将 celery 并发设置为每个队列 1 个工作人员答案

【问题标题】：Setting celery concurrency to 1 worker per queue将 celery 并发设置为每个队列 1 个工作人员
【发布时间】：2011-07-07 00:39:51
【问题描述】：

我本质上是在 celery 中使用 rabbitmq 队列作为穷人的同步。例如，当某些对象被更新（并且成本很高）时，我会根据它们的对象 ID 将它们轮询到一组 10 个队列。首先这是一种常见的模式还是有更好的方法。

其次，对于 celeryd，并发级别选项 (CELERY_CONCURRENCY) 似乎设置了所有队列中的工作人员数量。这违背了使用队列进行同步的目的，因为队列可以由多个工作人员提供服务，这意味着在对同一对象执行不同操作时可能会出现竞争条件。

有没有办法设置并发级别（或工作池选项），以便每 N 个队列有一个工作人员？

谢谢斯里

【问题讨论】：

除此之外，我想避免使用不同配置运行多个 celeryd 实例的选项（即，每个实例都设置为 CELERY_CONCURRENCY = 1 并且只监听 10 个队列的一个子集）。如果这是唯一的方法，那就太糟糕了。

【解决方案1】：

为什么不简单地通过使用 memcache 或 nosql db 来实现全局任务锁定系统？通过这种方式，您可以避免任何竞争条件。

【讨论】：

嘿，Mauro，很好的例子。感谢那。但是有几个问题： 1. 这不会让工作人员闲置，而锁还没有被另一个线程释放吗？我可以增加线程数，但它并没有真正解决问题。 2. 我可以看到这扩展到需要跨系统的多个锁，这意味着增加了复杂性或增加了死锁。 3. 最后我想避免外部依赖只是为了获得正确的并发性。我可以看到内存缓存的一般用途，但这似乎很长。
基本上你可以在找到锁的时候做你想做的事情，你可以实现你自己的行为，你可以重试任务或者让他失败，作品可以继续接收其他任务。基本上你要做的就是为你的任务创建一个基类，扩展 celery.task 并覆盖 call 方法和 after_return 方法。这样您就不必在每个任务逻辑中编写锁定行为。查看此演示文稿的幻灯片编号 21-22 slideshare.net/fireantology/… cmets 字符还不够

【解决方案2】：

关于你问题的第一部分，我在这里提出并回答了一个类似的问题：Route to worker depending on result in Celery?

基本上，您可以根据密钥（在您的情况下是 ID）直接路由到工作人员。它避免了对单个锁定点的任何需要。希望它有用，即使这个问题已有 2 年历史了:)

【讨论】：