将 Celery 任务函数的 .delay() 和periodic_task 发送到不同的队列

Question

我有一个代码库，其中包含几个应用程序，每个应用程序都有 tasks.py，并且总共有 100 个这样的函数

@periodic_task(run_every=crontab(minute='20'))
def sync_calendar_availability_and_prices(listing_id_list=None, reapply_rules_after_sync=False):

它采用 celery 定期任务定义的旧格式，但在 celery==4.1 上运行良好。

这些通过 beat 每隔几个小时或几分钟执行一次，我也使用 .delay() 在代码库中将它们称为 ad-hoc。我希望所有 .delay() 调用都进入某个 celery 队列manual_call_queue 并定期节拍触发调用相同功能的调用去periodic_beat_fired_queue - 这是一个简单的 1-2 行配置更改在全局级别的某个地方这样做？

我使用 rabbitmq、celery、django 和 django-celery-beat

Answer 1

要将周期性任务发送到特定队列，请发送队列/选项 arg。

@periodic_task(run_every=crontab(minute='20'), queue='manual_call_queue', options={'queue': 'periodic_beat_fired_queue'})
def sync_calendar_availability_and_prices(listing_id_list=None, reapply_rules_after_sync=False):

queue='manual_call_queue' 在使用 .delay 或 .apply_async 调用任务时使用

options={'queue': 'periodic_beat_fired_queue'} 用于 celery beat 调用任务时使用。