时间戳

首页 > 解决方案 > 工人不运行时执行芹菜任务

django - 工人不运行时执行芹菜任务

问题描述

即使没有芹菜工人，我也希望执行芹菜任务。该 celery 任务应该像正常功能一样工作，并且无论如何都会执行。

标签： djangocelerydjango-celerycelery-task

解决方案

有几个选项可以实现这一目标。broadcast推荐的方法使用控制命令明确地向工作人员查询已注册任务的列表：

app = Celery(broker='amqp://')

reply = app.control.broadcast('registered', reply=True)
tasks = set(t for n in reply for w in n.values() for t in w)

if 'my_task_name' in tasks:
    # there is an online worker that can handle the task
    # => use async mode
    app.send_task('my_task_name', args=(...), kwargs={...}, ...)
    # or use my_task.apply_async(...)
else:
    # there is no online worker for the task
    # => call the function directly
    my_task(...)

或者，您可以尝试将任务直接发送给代理并在短时间内检查结果。如果在那个时间之后任务仍然有PENDING结果，想必没有工人可以处理这个任务。如果是这种情况，您可以将任务作为普通函数调用：

app = Celery(broker='amqp://', backend='rpc://')

result = app.send_task('my_task_name', args=(...), kwargs={...}, ..., expiration=5)
time.sleep(5)
if result.state == 'PENDING':
    # presumably the task didn't start because there hasn't been any worker
    my_task(...)

不过，人们必须对第二种方法保持谨慎。任务挂起可能还有其他原因（例如，它们很忙并且预取了最大数量的任务）。此外，此示例假定您的代理支持 AMQP 过期（例如RabbitMQ）并且您不会忽略任务结果。

推荐阅读