python - 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在 celery 日志中。但是，如果我用无限任务填满 7 个 worker timey , 然后链 endtimey和 nexttask一起——

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

所有 timey任务将由 8 名 worker 中的 7 名承担，并且 endtimey将在第 8 个 worker 上运行它的过程，之后日志将显示 nexttask已收到，但nexttask不会运行。

这是为什么？

此外，如果我终止 celery 服务器然后重新启动它，nexttask将是第一个运行的东西。

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

Answer 1

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

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

问题是，当您有很多长时间运行的任务或混合了长任务和短任务时，任务可能会被保留并被忙碌的工作人员阻塞，即使其他工作人员可以处理它也是如此。

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