如何使用Celery,RabbitMQ和Django确保每个用户的任务执行顺序？

Question

我正在运行Django,Celery和RabbitMQ.我想要实现的是确保与一个用户相关的任务按顺序执行(具体来说,当时一个,我不希望每个用户的任务并发)

• 每当为用户添加新任务时,它应该取决于最近添加的任务.其他功能可能包括不将任务添加到队列,如果此类型的任务排队等待该用户但尚未启动.

我做了一些研究,并且:

• 我无法找到一种方法将新创建的任务与已在Celery中排队的任务链接起来,链似乎只能链接新任务.
• 我认为使用自定义RabbitMQ消息处理程序可以实现这两种功能,尽管它可能很难编码.
• 我也读过关于celery-tasktree的内容,这可能是确保执行顺序的最简单方法,但是如何将新任务与已经" applied_async"task_tree或队列相关联？有什么办法可以使用这个包实现额外的无重复功能吗？

编辑:在芹菜食谱中也有这个"锁定"的例子,因为这个概念很好,我看不到一种可能的方法让它在我的情况下按预期工作 - 只是如果我无法为用户获取锁定,任务必须重试,但这意味着将它推到队列的末尾.

这里最好的做法是什么？

Answer 1

如果您配置 celery 工作线程，使其一次只能执行一项任务（请参阅worker_concurrency设置），那么您可以在每个用户的基础上强制执行所需的并发性。使用类似的方法

NUMBER_OF_CELERY_WORKERS = 10

def get_task_queue_for_user(user):
    return "user_queue_{}".format(user.id % NUMBER_OF_CELERY_WORKERS)

根据用户id获取任务队列，每个用户的每个任务都会被分配到同一个队列中。工作人员需要配置为仅使用单个任务队列中的任务。

它会像这样发挥：

1. 用户49触发任务

2. 任务发送至user_queue_9

3. 当唯一一个正在侦听的 celery Workeruser_queue_9准备好使用新任务时，该任务就会被执行

但这是一个很奇怪的答案，因为

• 每个队列只需要一个 celery 工作人员是一个脆弱的系统 - 如果 celery 工作人员停止，整个队列也会停止

• 工人运转效率低下