Запланируйте задание redis, которое планирует другое задание redis, с помощью python-rq - PullRequest
Новая
       37

Запланируйте задание redis, которое планирует другое задание redis, с помощью python-rq

0 голосов
/ 27 августа 2018

У меня есть два вида заданий: те, которые я хочу выполнять в последовательном режиме, и те, которые я хочу выполнять параллельно в параллельном режиме. Однако я хочу, чтобы параллельные задания планировались в последовательном порядке (если вы все еще подписываетесь). То есть:

  1. До А.
  2. Ждите A, делайте B.
  3. Ждите B, делайте 2+ версии C одновременно.

Я подумал, что у него 2 очереди redis, в serial_queue только один рабочий. И параллельная очередь с несколькими работниками. 

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?

1 Ответ

0 голосов
/ 28 августа 2018

Решение требует немного гимнастики, так как вам нужно импортировать текущий скрипт, как если бы это был внешний модуль .

Так, например. Содержимое 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
   )
...