如何在celery中生成每个用户的队列？

Question

因此，我尝试将阻塞内容从web请求转移到后台任务，并利用队列。我也是消息传递和pub/sub的新手。用户将数据推送到那里，数据会被处理，然后用户会收到通知。我为此做了一个芹菜设置，发现它不能满足我的用例，即每个用户都有自己的任务的私有队列。

我尝试指定在worker派生期间创建丢失的队列(发送队列名称，用逗号分隔)，并将它们列在队列设置中，就像之前在internet上回答“用芹菜动态创建队列”时所说的那样。它创建队列，但当我在设置和命令行中指定的队列名称与指定的名称不同时，它不会创建队列。解决方案是产生更多具有队列名称的工作进程，这不能满足用例，因为将有数百万个数据处理请求。

我发现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()方法从代码实现。这些操作是独立的，所以如果worker已经从队列中消费，则不会发生任何事情。如果队列还不存在，它会自动创建。