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

Question

撤销@periodic_task 上的任务发送Discarding revoked tasks & Due task to workers.

celery-workers-screenshot

[2018-09-17 12:23:50,864: INFO/MainProcess] Received task: cimexapp.tasks.add[xxxxxxx]
[2018-09-17 12:23:50,864: INFO/MainProcess] Discarding revoked task: cimexapp.tasks.add[xxxxxxx] [2018-09-17 12:24:00,865: INFO/Beat] Scheduler: Sending due task cimexapp.tasks.add (cimexapp.tasks.add) [2018-09-17 12:24:00,869: INFO/MainProcess] Received task: cimexapp.tasks.add[xxxxxxx]
[2018-09-17 12:24:00,869: INFO/MainProcess] Discarding revoked task: cimexapp.tasks.add[xxxxxxx] [2018-09-17 12:24:10,865: INFO/Beat] Scheduler: Sending due task cimexapp.tasks.add (cimexapp.tasks.add) [2018-09-17 12:24:10,868: INFO/MainProcess] Received task: cimexapp.tasks.add[xxxxxxx]
[2018-09-17 12:24:10,869: INFO/MainProcess] Discarding revoked task: 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() 函数的任务/工作人员。

Answer 1

一种可能的方法是将任务存储在数据库中并动态添加删除任务。您可以使用数据库支持的 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 再次激活它。

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

在创建项目模型对象时，也要创建周期性任务。只需根据您的方案创建 cron 计划或间隔计划。然后创建PeriodicTask 对象，您可以将Project.name 赋予PeriodicTask 名称（这样您就可以轻松地将项目对象与PeriodicTask 对象关联起来。就是这样，从那一刻起，任务将由celery beat 处理。

如果您想动态禁用或启用定期任务，只需在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