首页 文章

当前工作完成后，芹菜 Worker 不会接受新任务

提问于
浏览 次
2

我有三个任务：

@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

如果我唯一能做的就是连锁超时和下一步 -

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

一切都按预期工作 . 我在芹菜日志中看到 ENDTIMEY <current time> 打印十次，然后是 NEXT TASK . 但是，如果我用无限任务 timey 填满7名 Worker ，然后将 endtimey 和 nexttask 链在一起 -

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

所有 timey 任务将由8名工作人员中的7人接收， endtimey 将在第8名工作人员上运行，之后日志将显示已收到 nexttask ，但 nexttask 将无法运行 .

为什么是这样？

此外，如果我杀死芹菜服务器然后重新启动它， nexttask 将是第一个运行 .

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

python django rabbitmq celery

1 回答

  • 4

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

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

    问题是，当你有很多长时间运行的任务或者长短任务的混合时，即使其他工作人员可以处理它，任务也可以被繁忙的工作人员保留和阻止 .

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

    回复于

相关问题