python celery - 如何在运行时将 CELERYBEAT_SCHEDULE 任务添加到工作人员？答案

【问题标题】：python celery - how to add CELERYBEAT_SCHEDULE task at runtime to a worker?python celery - 如何在运行时将 CELERYBEAT_SCHEDULE 任务添加到工作人员？
【发布时间】：2016-02-14 14:30:49
【问题描述】：

我创建了一个具有单个 celerybeat 调度任务的 celery worker，该任务以 5 秒的时间间隔运行。如何在不停止的情况下向 celery worker 动态添加另一个节拍任务？

例子

app.conf.update(
   CELERY_TASK_RESULT_EXPIRES=3600,
   CELERY_TIMEZONE = 'UTC',
   CELERYBEAT_SCHEDULE = {
    'long-run-5-secs': {
        'task': 'test_proj.tasks.test',
        'schedule': timedelta(seconds=5),
        'args': (16, )
    }
   }
)

通过上面的配置，我可以成功运行 celery worker 的 beat 模式。

现在我需要动态添加以下节拍时间表：

'long-run-2-secs': {
    'task': 'test_proj.tasks.test',
    'schedule': timedelta(seconds=2),
    'args': (14, ) },

谢谢

【问题讨论】：

标签： python celery scheduler celerybeat djcelery

【解决方案1】：

我一直在为同样的问题寻找解决方案。我担心你将不得不等待 Celery ver.4.0。目前仅开发版本支持动态任务调度：http://docs.celeryproject.org/en/master/userguide/periodic-tasks.html#beat-entries

【讨论】：

【解决方案2】：

一种可能的方法是将任务存储在数据库中并动态添加删除任务。您可以使用数据库支持的 celery beat 调度程序。请参阅https://django-celery-beat.readthedocs.io/en/latest/。 PeriodicTask 数据库存储周期性任务。您可以使用数据库命令 (Django ORM) 来操作周期性任务。

这就是我处理动态任务的方式（动态创建和停止任务）。

from django_celery_beat.models import PeriodicTask, IntervalSchedule, CrontabSchedule

chon_schedule = CrontabSchedule.objects.create(minute='40', hour='08', day_of_week='*', day_of_month='*', month_of_year='*') # To create a cron schedule. 
schedule = IntervalSchedule.objects.create(every=10, period=IntervalSchedule.SECONDS) # To create a schedule to run everu 10 min.
PeriodicTask.objects.create(crontab=chon_schedule, name='name_to_identify_task',task='name_of_task') # It creates a entry in the database describing that periodic task (With cron schedule).
task = PeriodicTask.objects.create(interval=schedule, name='run for every 10 min', task='for_each_ten_min', ) # It creates a periodic task with interval schedule

每当您更新 PeriodicTask 时，此表中的计数器也会递增，它告诉 celery beat 服务重新加载从数据库中调度。

因此您无需重新启动或终止节拍。如果您想在满足特定条件时停止任务，那么

periodic_task = PeriodicTask.objects.get(name='run for every 10 min')
periodic_task.enabled = False
periodic_task.save()

当 enabled 为 False 时，周期性任务变为空闲。您可以通过 enable = True 再次激活它。

如果您不再需要该任务，则只需删除该条目即可。

【讨论】：