Celery 按照调用顺序执行任务（在运行时）

Question

我有一个由链中的子任务组成的任务。如何确保在第一个任务完成之前不会开始第二次调用此任务？

@shared_task
def task(user):
    res = chain(subtask_1.s(), # each subtask takes ~1 hour
            subtask_2.s(),
            subtask_3.s())

    return res.apply_async()

django 视图现在可能会触发调用此任务：

# user A visits page that triggers task
task.delay(userA)
# 10 seconds later, while task() is still executing, user B visits page
task.delay(userB) 

这会导致任务相互竞争，而不是按顺序执行。例如。一旦工作人员完成了第一个任务的subtask_1()，它就会开始处理第二个任务的subtask_1()，而不是第一个任务的subtask_2() 和subtask_3()。

有没有办法优雅地避免这种情况？我想问题是子任务添加到队列中的顺序。

1. 我已经设置了worker --concurreny=1，但是这仍然不会改变他从队列中消费的顺序。
2. Official docs (task cookbook) 似乎提供了一个我不理解的解决方案，不幸的是对我不起作用。
3. 也许在任务链之后包含一个阻塞机制，带有while not res.ready(): sleep(1) 类型的hack？

Answer 1

您可以等待第一个任务完成，然后像这样执行第二个任务。

res = task.delay(userA)
res.get() # will block until finished
task.delay(userB)

但它会阻塞调用线程，直到第一个线程完成。您可以链接任务以避免阻塞，但为此您必须稍微修改任务签名以接受任务结果作为参数。

@shared_task
def task(_, user): signature takes one extra argument
    # skipped

和

from celery.canvas import chain


chain(task.s(None, userA), task.s(userB))()