芹菜中如何向特定队列发送周期性任务 [英] How to send periodic tasks to specific queue in Celery

问题描述

默认情况下，芹菜将所有任务发送到‘celery’队列，但您可以通过添加额外参数来更改此行为：

@task(queue='celery_periodic')
def recalc_last_hour():
    log.debug('sending new task')
    recalc_hour.delay(datetime(2013, 1, 1, 2)) # for example

日程安排程序设置：

CELERYBEAT_SCHEDULE = {
   'installer_recalc_hour': {
        'task': 'stats.installer.tasks.recalc_last_hour',
        'schedule': 15  # every 15 sec for test
    },
}
CELERYBEAT_SCHEDULER = "djcelery.schedulers.DatabaseScheduler"

运行工作进程：

python manage.py celery worker -c 1 -Q celery_periodic -B -E

此方案未按预期工作：此工作进程将定期任务发送到"celery"队列，而不是"celery_periical"。我如何修复它？

P.S.芹菜==3.0.16

推荐答案

我找到了此问题的解决方案：

1)首先，我更改了周期性任务的配置方式。我使用了这样的@PERIONIC_TASK修饰符：

@periodic_task(run_every=crontab(minute='5'),
               queue='celery_periodic',
               options={'queue': 'celery_periodic'})
def recalc_last_hour():
    dt = datetime.utcnow()
    prev_hour = datetime(dt.year, dt.month, dt.day, dt.hour) 
                - timedelta(hours=1)
    log.debug('Generating task for hour %s', str(prev_hour))
    recalc_hour.delay(prev_hour)

2)我在@periical_task的参数中写了两次celery_periical：

• queue=‘celery_periodic’选项在您从代码(.Delay或.Apply_Async)调用任务时使用

• options={‘queue’：‘celery_periical’}选项在芹菜节拍调用时使用。

我确信，如果您使用CELERYBEAT_SCHEDUE变量配置定期任务，同样的事情也是可能的。

更新。此解决方案适用于CELERYBEAT_SCHEDULER的基于数据库和基于文件的存储。

这篇关于芹菜中如何向特定队列发送周期性任务的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！