使用 Celery 动态创建队列

Question

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

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

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

Answer 1

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

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

可以建立一种系统，将高优先级消息发送到专用队列和消费者，类似地，普通或低优先级消息由一个或多个队列处理。

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