python - 安排一个redis作业，用python-rq安排另一个redis作业

Question

我有两种作业:一种是我想串行运行的，另一种是我想同时并行运行的。但是，我希望并行作业以串行方式安排(如果您仍在关注)。即:

1. 做 A。
2. 等待 A，做 B。
3. 等待 B，同时执行 2 个以上版本的 C。

我认为它有 2 个 redis 队列，一个只有一个 worker 的 serial_queue。还有一个 parallel_queue，上面有多个 worker。

serial_queue.schedule(
    scheduled_time=datetime.utcnow(),
    func=job_a,
     ...)    
serial_queue.schedule(
    scheduled_time=datetime.utcnow(),
    func=job_b,
     ...)

def parallel_c():
    for task in range(args.n_tasks):
        queue_concurrent.schedule(
            scheduled_time=datetime.utcnow(),
            func=job_c,
            ...)

serial_queue.schedule(
    scheduled_time=datetime.utcnow(),
    func=parallel_c,
     ...)

但是目前这个设置给出了错误AttributeError: module '__main__' has no attribute 'schedule_fetch_tweets' .如何为 python-rq 正确打包此函数？

Answer 1

该解决方案需要一些技巧，因为您必须导入当前脚本，就好像它是一个外部模块一样。

举个例子。 schedule_twitter_jobs.py 的内容为:

from redis import Redis
from rq_scheduler import Scheduler
import schedule_twitter_jobs
# we are importing the very module we are executing

def schedule_fetch_tweets(args, queue_name):
    ''' This is the child process to schedule'''

    concurrent_queue = Scheduler(queue_name=queue_name+'_concurrent', connection=Redis())
    # this scheduler is created based on a queue_name that will be passed in
    for task in range(args.n_tasks):
        scheduler_concurrent.schedule(
            scheduled_time=datetime.utcnow(),
            func=app.controller.fetch_twitter_tweets,
            args=[args.statuses_backfill, fill_start_time])

serial_queue = Scheduler(queue_name='myqueue', connection=Redis())
serial_queue.schedule(
'''This is the first schedule.'''
   scheduled_time=datetime.utcnow(),
   func=schedule_twitter_jobs.schedule_fetch_tweets,
   #now we have a fully-qualified reference to the function we need to schedule.
   args=(args, ttl, timeout, queue_name)
   #pass through the args to the child schedule
   )