python - 基于集合作为数据结构而不是列表的分布式任务队列

Question

我想知道是否有一种方法可以设置 RabbitMQ 或 Redis 与 Celery 一起工作，这样当我将任务发送到队列时，它不会进入任务列表，而是进入一组任务根据我的任务的有效负载进行键控，以避免重复。

这是我的设置以获得更多上下文: python + celery 。我已经尝试过 RabbitMQ 作为后端，现在我正在使用 Redis 作为后端，因为我不需要 100% 的可靠性、更容易使用、内存占用小等。

我有大约 1000 个 ID 需要重复完成。我的数据管道的第 1 阶段由调度程序触发，它输出第 2 阶段的任务。任务仅包含需要完成工作的 ID，实际数据存储在数据库中。我可以运行任何组合或序列的第 1 阶段和第 2 阶段任务而不会造成伤害。

如果第 2 阶段没有足够的处理能力来处理第 1 阶段输出的任务量，我的任务队列就会越来越大。如果任务队列使用集合而不是列表作为基础数据结构，则不必是这种情况。

是否有现成的解决方案可以从列表切换到集合作为分布式任务队列？ celery 有这个能力吗？我最近看到 Redis 刚刚发布了一个队列系统的 alpha 版本，所以它还没有准备好用于生产。

我应该以不同的方式构建我的管道吗？

Answer 1

您可以使用外部数据结构来存储和监控 celery 队列的当前状态。1.让我们以redis键值为例。每当您将任务推送到 celery 中时，您都会在 redis 中将带有“id”字段的键标记为 true。

1. 在尝试推送带有任何 'id' 的新任务之前，您将检查带有 'id' 的键在 Redis 中是否为真，如果是，则跳过推送任务。

2. 要在适当的时候清除 key ，您可以使用after_return。 celery 的处理程序，在任务返回时运行。此处理程序将取消设置 redis 中的键“id”，从而清除下一个任务推送的锁。

此方法确保您在 celery 队列中运行的每个任务 ID 只有一个实例。您还可以通过在 redis 键上使用 INCR 和 DECR 命令，在推送任务和任务的 after_return 时，将其增强为每个 ID 仅允许 N 个任务。