我们如何在 celery periodic_task 中完成停止任务？

Question

Revoking a task on @periodic_task sends Discarding revoked tasks & Due task to workers.

celery-workers-screenshot

[2018-09-17 12:23:50,864: INFO/MainProcess] 收到任务: cimexapp.tasks.add[xxxxxxx]<br> [2018-09-17 12:23:50,864: INFO/MainProcess] 丢弃已撤销的任务：cimexapp.tasks.add[xxxxxxx] [2018-09-17 12:24:00,865: INFO/Beat] 调度程序：发送到期任务 cimexapp.tasks.add (cimexapp.tasks.add) [2018-09-17 12:24:00,869: INFO/MainProcess] 收到任务: cimexapp.tasks.add[xxxxxxx]<br> [2018-09-17 12:24:00,869: INFO/MainProcess] 丢弃已撤销的任务：cimexapp.tasks.add[xxxxxxx] [2018-09-17 12:24:10,865: INFO/Beat] 调度程序：发送到期任务 cimexapp.tasks.add (cimexapp.tasks.add) [2018-09-17 12:24:10,868: INFO/MainProcess] 收到任务: cimexapp.tasks.add[xxxxxxx]<br> [2018-09-17 12:24:10,869: INFO/MainProcess] 丢弃已撤销的任务：cimexapp.tasks.add[xxxxxxx]

tasks.py

@periodic_task(run_every=timedelta(seconds=10),options={"task_id":"xxxxxxx"})
def add():
     call(["ping","-c10","google.com"])


def stop():
    x = revoke("xxxxxxx",terminate=True,signal="KILL")
    print(x)
    print('DONE')

我创建了 task_id 名称，这样我就可以很容易地通过调用 id 来杀死它。

如何完全阻止它发送任务？我不想用

杀死所有工人

pkill -9 -f 'celery worker'
celery -A PROJECTNAME 控件关闭

我只想停止 add() 功能的 tasks/workers。

Answer 1

一种可能的方法是将任务存储在数据库中并动态添加删除任务。您可以使用数据库支持的芹菜节拍调度程序。参考 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

Whenever you update a PeriodicTask a counter in this table is also incremented, which tells the celery beat service to reload the schedule from the database.

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

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

当 enabled 为 False 时，周期性任务变为空闲。您可以通过 enable = True.

再次激活它

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

在创建项目模型对象时也创建周期性任务。只需根据您的场景创建 cron 计划或间隔计划。然后创建 PeriodicTask 对象，您可以将 Project.name 命名为 PeriodicTask（这样您就可以轻松地将项目对象与 PeriodicTask 对象相关联。就是这样，从那一刻起，任务将由芹菜节拍处理。

如果您想动态禁用或启用定期任务，只需在 PeriodicTask 中设置启用标志，如下所示

    task = PeriodicTask.objects.get(name='task_name')
    task.enabled = False
    task.save()

Answer 2

在Celery 4.2.2中，您可以通过执行以下命令删除所有节拍任务

celery -A YourProjectName purge

记得替换 YourProjectName。有关详细信息，请查看此 page

我们如何在 celery periodic_task 中完成停止任务？

How do we completed stop tasks in celery periodic_task?

django

celery

celery-task

django-celery

celerybeat