使用Celery创建动态队列

Question

这是我的情况:

当用户登录到我的网站时，我为该用户排队一堆任务(通常每个任务需要100毫秒，每个用户有100个任务)。这些任务被排队到默认的 Celery 队列中，我有数百个工作进程在运行。我使用 Websockets 显示用户实时进度，以便在后端完成任务时进行更新。如果只有 1 或 2 个用户活跃，则一切都很好。

现在，如果几个并发用户登录到我的网站，后面的用户将排在最初的用户之后排队，他们的任务会饿死(因为所有任务都进入同一个队列)。我的想法是为每个用户创建一个动态队列，以确保公平性。然而根据 Celery 文档 (http://docs.celeryproject.org/en/latest/userguide/routing.html#defining-queues)，似乎我需要静态地定义队列。

有没有关于在我的情况下使用 celery 的最佳实践建议？

- El Diablo

---

动态队列不会让排队失去意义吗？每个用户有成百上千的任务听起来很多。这些任务不能被分组在一起，以便一个工作人员可以一次性处理所有用户的任务吗？这样用户就可以拥有专门负责登录过程的工作人员了。所有登录任务需要同步运行吗？ - Joe Doherty

@JoeDoherty - 我不明白你说动态队列如何会破坏排队的目的。但是，你建议将每个用户任务批处理为一个大任务的想法非常有趣，肯定可以缓解我的问题 - 谢谢！从设计角度来看，我试图实现一个执行小/快速任务（而不是长时间运行任务）的大型工作池。不过，我仍然很想知道是否有在运行时创建队列的选项 - 即使我批量处理用户任务，我也可以看到这很有用。 - El Diablo

@JoeDoherty - 关于“所有登录任务是否需要同步运行”的问题 - 答案是“不需要”。要求是它们都在合理的时间内完成，因为用户正在等待网站上任务的结果完成。 - El Diablo

我忘记用户正在等待每一个任务完成。此外，我对于可以解决您的问题的动态队列的想法是错误的。您可以将它们全部分批处理，然后在每个阶段完成时报告回来，类似于this。 - Joe Doherty

Answer 1

http://docs.celeryproject.org/en/latest/userguide/workers.html#queues

celery -A proj control add_consumer foo -d worker1.local

使用app.control.add_consumer()方法可以动态实现相同的功能：

app.control.add_consumer('foo', reply=True)
[{u'worker1.local': {u'ok': u"already consuming from u'foo'"}}]

app.control.add_consumer('foo', reply=True,
destination=['[e-mail]'])

- esdotzed

Answer 2

在运行时引用它时，您可以动态地将任务分配给队列，请参阅calling.html#routing-options。 如果启用了CELERY_CREATE_MISSING_QUEUES，则此方法可行。

- Gautam