django - 如何在 celery 中为每个用户生成队列？

Question

所以我正在尝试将 Web 请求中的阻塞内容作为后台任务并利用队列。我也是消息传递和发布/订阅的新手。用户将数据推送到那里并进行处理，稍后会通知用户。我为此做了一个 celery 设置，发现它不能满足我的用例，即为每个用户设置私有(private)队列来完成他们自己的任务。

我尝试指定缺少队列的创建和工作人员产卵期间(发送队列名称以逗号分隔)，并将它们列在队列设置中，如之前在互联网上对“使用 celery 创建动态队列”的回答中所述。它会创建队列，但当我在设置和命令行中指定的队列名称与指定名称不同时，它不会创建队列。解决方案是使用不满足用例的队列名称生成更多工作人员，因为将有数百万个数据处理请求。

我发现 python-rq 有 Queue 对象初始化及其名称，我认为它创建了新队列。如果是这样，转向 RQ 是否正确？

redis_conn = Redis()
q = Queue('some_queue', connection=redis_conn)

我想要的是每个用户在后台为自己的任务排队。我没有在 celery 中看到任何在线创建动态队列的解决方案(没有在命令行中指定队列名称或设置)。 python-rq 似乎有那个解决方案。我的权衡是从 RabbitMQ 和 celry 转移到 redis。

有没有办法在 celery 中真正实现每个用户的队列？如果是，请列出步骤。还是这种设计模式不对？ pubsub 会满足用例吗？

Answer 1

Celery worker 仅从 task_queues 定义的队列中消费使用 -Q 在命令行上设置或给出选项。但是，这可以通过命令行或代码动态更改。一定要有 task_create_missing_queues在设置中启用(这是默认设置)，以便在您开始使用它们时自动创建新队列。

因此，在为给定用户发送任务之前，您必须指示工作人员开始从用户队列中消费。这可以通过使用 add_consumer 从命令行实现。控制命令，或使用 app.control.add_consumer() 的代码方法。这些操作是幂等的，因此如果工作人员已经从队列中消费，则什么也不会发生。如果队列尚不存在，则会自动创建。