基于集合而不是列表的数据结构的分布式任务队列答案

【问题标题】：Distributed Task Queue Based on Sets as a Data Structure instead of Lists基于集合而不是列表的数据结构的分布式任务队列
【发布时间】：2015-07-28 11:21:23
【问题描述】：

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

这是我的设置以获取更多上下文：蟒蛇+芹菜。我已经尝试过 RabbitMQ 作为后端，现在我使用 Redis 作为后端，因为我不需要 100% 的可靠性、更易于使用、内存占用少等。

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

如果第 2 阶段没有足够的处理能力来处理第 1 阶段输出的任务量，我的任务队列就会不断增长。如果任务队列使用集合而不是列表作为底层数据结构，则不必如此。

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

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

【问题讨论】：

使用 RabbitMQ，您可以为每个唯一 ID 延迟创建队列，最大队列深度 (x-max-length) 为 1 。发布和订阅 1000 个不同的队列需要额外的内务处理，但会根据需要删除重复项。
这正是我正在寻找的那种解决方法类型的逻辑，但这种特殊的解决方案似乎很乏味，而且我更愿意远离 RabbitMQ。不过，感谢您的创意！

【解决方案1】：

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

在尝试使用任何 'id' 推送新任务之前，您将检查具有 'id' 的键在 redis 中是否为真，如果是，则跳过推送任务。
要在适当的时候清除密钥，您可以使用 celery 的 after_return 处理程序，该处理程序在任务返回时运行。此处理程序将取消设置 redis 中的键“id”，从而为下一个任务推送清除锁定。

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

【讨论】：

【解决方案2】：

在 Redis 中为您的作业队列使用 SortedSet。它确实是一个 Set，因此如果您将完全相同的数据放入其中，则不会在其中添加新值（它绝对需要是完全相同的数据，您不能覆盖 Redis 中 SortedSet 中使用的哈希函数）。

您需要一个分数才能与 SortedSet 一起使用，您可以使用时间戳（值作为双精度值，例如使用 unixtime），这样您就可以根据需要获取最新的项目/最旧的项目。 ZRANGEBYSCORE 可能是您要查找的命令。 http://redis.io/commands/zrangebyscore

此外，如果您需要其他行为，您可以将所有内容包装在 Lua 脚本中，以实现原子行为和自定义驱逐策略（如果需要）。例如，调用“get”脚本来获取作业并将其从队列中自动删除，或者在背压过大时驱逐数据等。

【讨论】：

【解决方案3】：

您在第 2 阶段的任务是否可以检查工作是否已经完成，如果已经完成，则不再执行该工作？这样，即使您的任务列表会增加，您需要完成的工作量也不会增加。

我还没有遇到关于集合/列表的解决方案，我认为还有很多其他方法可以解决这个问题。

【讨论】：

任务队列本身的大小是个问题。根据您使用哪种软件作为排队解决方案，您最终会占用内存或占用数据库中的空间。这些都不是可取的。我更喜欢行为良好的系统。
对，我明白了。大概即使您确实有使用集合的解决方案，您也会排队已经完成的工作？我想知道是否有一种方法可以在某处记录“已完成的工作或计划完成的工作”，并且仅当不在该列表中时才将任务添加到队列中。