在 Celery 中，如何防止长时间延迟的任务阻塞较新的任务？答案

【问题标题】：In Celery, how can I keep long-delayed tasks from blocking newer ones?在 Celery 中，如何防止长时间延迟的任务阻塞较新的任务？
【发布时间】：2014-07-01 09:21:54
【问题描述】：

我有两种任务。任务 A 由 celerybeat 每小时生成一次。它立即运行，并生成一千个（或数千个）任务 B 实例，每个实例的 ETA 为未来一天。

在启动时，任务 A 的一个实例运行并生成一千个 B。从那时起，什么都没有发生。我应该看到另一个 A 每小时运行，另外一千个 B。但实际上我什么也没看到。

在冻结时，rabbitmqctl 显示 1000 条消息，其中 968 条准备就绪，32 条未确认。一小时后，有 1001 条消息，969 条准备就绪，32 条未确认。以此类推，每小时都会有一条新消息归类为就绪。大概发生的事情是工作人员正在预取 32 条消息，但无法对它们采取行动，因为他们的 ETA 仍在未来。同时，本应立即运行的较新任务无法运行。

处理这个问题的正确方法是什么？我猜我需要多个工人，也许还有多个队列（但我不确定后一点）。有没有更简单的方法？我试过摆弄 CELERYD_PREFETCH_MULTIPLIER 和 -Ofail （如此处所述：http://celery.readthedocs.org/en/latest/userguide/optimizing.html），但无法完成。我的问题和这个问题一样吗：[[Django Celery]] Celery blocked doing IO tasks？

无论如何：我能解决这个问题只是因为我对任务的性质和时间安排了如指掌。未来 ETA 的足够多的任务可以锁定整个系统，这似乎不是一个设计缺陷吗？如果我等待几个小时，然后杀死并重新启动工作程序，它会再次抓取前 32 个任务并冻结，即使此时队列中有准备好运行的任务。某些组件不应该足够聪明以查看 ETA 并忽略不可运行的任务吗？

附录：我现在认为这个问题是当 RabbitMQ 3.3 与 Celery 3.1.0 一起使用时的一个已知错误。更多信息在这里： https://groups.google.com/forum/#!searchin/celery-users/countdown|sort:date/celery-users/FiAAESOzezA/499OH-pylacJ

更新到 Celery 3.1.1 后，情况似乎好多了。任务 A 每小时运行一次（嗯，它有几个小时）并安排任务 B 的副本。这些似乎正在填满工作人员：未确认消息的数量继续增长。我得看看它是否可以无限制地生长。

【问题讨论】：

是的，RabbitMQ 3.3 打破了一切。我也遇到了同样的问题，Ask 很快发布了 Celery 3.1.11、Kombu 3.0.13 和 librabbitmq 1.5.0。
谢谢你，我花了一整天的时间试图找出问题所在，直到我发现这个！
你现在解决了吗？

标签： python rabbitmq celery scheduling

【解决方案1】：

这似乎是一个可以通过路由解决的问题： http://celery.readthedocs.org/en/latest/userguide/routing.html

使用路由时，您可以拥有多个填充不同类型任务的队列。如果您希望任务 B 不阻塞更多任务 A，您可以将它们放入具有不同优先级的单独工作队列中，这样您的工作人员将在充满任务 B 的大队列上工作，但是当任务 A 到达时，它会被下一个可用的工人。

这样做的额外好处是，您还可以将更多工作人员分配到人满为患的队列中，而这些工作人员只会从指定的高容量队列中拉出。

【讨论】：

【解决方案2】：

您目前有多少工作人员以什么并发运行？

提高工作人员的并发性可能有助于解决问题。如果线程 x 卡在任务 A 上，该任务需要很长时间或处于等待状态，则另一个线程可以处理其他预取任务

http://celery.readthedocs.io/en/latest/reference/celery.bin.worker.html#cmdoption-celery-worker-c

【讨论】：