Celery worker 不会在当前任务完成后接手新任务

Question

我有三个任务：

@app.task(name='timey')
def timey():
    print "timey"
    while True:
        pass
    return 1

@app.task(name='endtimey')
def endtimey():
    for i in range(10):
        print "ENDTIMEY", time()
        sleep(3)
    return 1

@app.task(name='nexttask')
def nexttask(n):
    print "NEXT TASK"
    return 1

如果我唯一要做的就是将 endtimey 和 nexttask 链接在一起 -

chain(endtimey.s() | nexttask.s()).delay()

一切正常。我在芹菜日志中看到 ENDTIMEY <current time> 打印十次，然后是 NEXT TASK 。但是，如果我用无限任务 timey 填满 7 个工人，然后将 endtimey 和 nexttask 链接在一起 -

for i in range(7):
    timey.s().delay()
chain(endtimey.s() | nexttask.s()).delay()

所有 timey 任务将由 8 个工人中的 7 个完成，endtimey 将运行在第 8 个工人身上完成，之后日志将显示nexttask已收到，但nexttask不会运行。

这是为什么？

此外，如果我杀死 celery 服务器然后重新启动它，nexttask 将是运行的第一件事。

这是一个人为的例子，但我在更复杂的情况下遇到了一个问题，即 celery worker 在完成当前任务后没有选择排队的任务。如果我在那种情况下重新启动芹菜，自由工人将再次开始接任务。

Answer 1

听起来问题出在 celery 的默认预取行为上。每个工作人员将在当前处于最大容量时提前保留一定数量的任务，这被称为 Prefetch Multiplier.

这样做的原因是，当您有大量短任务时，如果任务已经预取并准备好立即执行，您的总体吞吐量会高得多。

问题是，当您有很多长运行任务或混合了长任务和短任务时，任务可能会被一个忙碌的工作人员预留和阻塞，即使其他工作人员有空处理一下。

因此，在您的情况下，您可能需要将预取乘数降低到 1。

Celery worker 不会在当前任务完成后接手新任务

Celery worker will not pick up a new task after the current one is finished

python

django

rabbitmq

celery