使用 Celery 动态创建队列

Question

我正在使用 Django、Celery 和 RabbitMQ 编写一个邮件列表管理器。当有消息进来时，会为每个接收者执行一个任务。所有任务都进入一个队列，一个或多个工作人员使用队列中的任务，构建电子邮件消息并发送它们。

单个队列会导致公平性问题：如果一条消息进入一个大型邮件列表，则会将大量任务添加到队列中，并且其他消息无法通过其他较小的邮件列表，直到所有消息都到达大名单已发送。我怎样才能找到解决这个问题的方法？

从概念上讲，一个解决方案是为每个邮件列表创建一个队列，并让工作人员从各个队列循环中消费任务。考虑到我需要能够动态创建新的邮件列表，这在 Celery 中是否可行？我还没有看到动态创建队列或使工作人员从新队列中消费的功能。

Answer 1

如下图所示，考虑一个使用主题交换而不是直接交换的系统。

与直接交换不同，主题交换允许我们将不同的消息路由到不同的队列。这是通过为每条消息设置 routing_key 并绑定某些队列以仅接受具有特定路由键的消息来完成的。

可以设置一个系统，将高优先级发送到专用队列和消费者，类似地，正常或低优先级消息由一个或多个队列处理。

Celery 支持跨多个工作人员的负载平衡（工作人员的数量取决于硬件）。像 BROKER_POOL_LIMIT、PREFETCH_LIMIT 等 Celery 配置设置有助于更好地平衡负载并减少拥塞。