芹菜活动任务因工人关闭而丢失

Question

我正在使用 celery=4.0.0，当任务运行较长时间并且工作人员在其间关闭时，我丢失了所有这些任务。如何重新启动所有丢失的任务。提前致谢。 芹菜 = 4.0.0 redis = 2.10.5

Answer 1

您应该重写 celery 信号方法。 信号列表； http://docs.celeryproject.org/en/latest/userguide/signals.html

为 celery 任务创建模型。例如它的名字是 CeleryTask。 在任务运行之前，在 CeleryTask 模型中创建新的任务对象。当 celery 重新启动并且 worker 准备好时，任务会再次运行。

tasks.py

from celery.signals import worker_ready, task_prerun

def on_task_prerun(*args, **kwargs):
    from .models import CeleryTask

    task_properties = kwargs['task']

    ctask = CeleryTask.objects.create(
        task_name=task_properties.name,
        task_args=task_properties.request.args
    )

def on_worker_ready(*args, **kwargs):
    from .models import CeleryTask
    from celery.execute import send_task

    tasks = CeleryTask.objects.all()

    for ctask in tasks:
        send_task(ctask.task_name, ctask.task_args)

worker_ready.connect(on_worker_ready, dispatch_uid='on_worker_ready')
task_prerun.connect(on_task_prerun, dispatch_uid='on_task_prerun')