Redis—在使用BRPOPLPUSH时清理处理队列（可靠）的更好方法

Question

环境再贴现2.8。17

我们已经使用与redis文档中描述的模式类似的模式在RPOPLPPUSH下实现了可靠队列

然而，考虑到其阻塞性质，我们使用BRPOPLPUSH，而LPUSH用于确保FIFO顺序。

生产者：使用LPUSH推送项目的多个线程（来自多个服务器）。

消费者：使用BRPOPLPUSH处理项目的多个线程（来自多个服务器）。

BRPOPLPUSH q processing-q

如文件所述，redis从队列“q”中弹出项目，同时将它们添加到“processing-q”中。

由于应用程序的多线程（异步）特性，我们无法控制使用者何时完成处理。

因此，如果我们使用LREM（根据文档）从processing-q中删除已处理元素，这将只删除processing-q的顶部元素。因为它不能保证是否已删除由相应消费者处理的实际元素。

因此，如果我们不做任何事情，处理q会继续增长（吞噬内存），恕我直言，这是非常糟糕的。

有什么建议或想法吗？

Answer 1

在一个类似的项目中，我使用备份队列的工作人员的主机名和进程ID。每个辅助角色都有自己的备份队列，如果辅助角色死亡，该项目不会丢失。

查看自述文件和实现以了解更多详细信息。

Answer 2

我将采用的方法是使用每个消费者的processing-q（例如processing-q：消费者id）。这将解决您当前的问题，但您仍然需要以某种方式处理崩溃的消费者。为此，我建议您也保留每个消费者最后一次弹出任务的时间，并定期检查超时情况。如果使用者已达到超时，请将其任务移回主队列并删除其队列。

Answer 3

您只需在对LREM的呼叫中包含要删除的作业。

LREM的形式如下：

LREM queue count "object"

它将从队列中删除等于“对象”的计数项。因此，要删除您的消费者线程正在处理的特定作业，您可以执行这样的操作。

LREM processing-q 1 "job_identifier"

有关更多信息，请参阅此处的文档：http://redis.io/commands/lrem

然后，为了处理崩溃的消费者和放弃的作业，您可以使用SETEX创建带过期的锁，并定期检查没有锁的作业。

所以整个过程是这样的：

制作人

1. RPUSH q"job_identifier"

消费者

1. SETEX-lock:processing-q:job\u标识符60（首先设置锁以避免竞争条件）
2. BRPOPLPUSH q处理队列
3. 处理作业
4. LREM处理队列“作业标识符”

已过期工作监控

1. 作业=LRANGE处理队列0-1

@NotAUser发布了一个开源java实现，如下所示：https://github.com/graknlabs/redisq/tree/master/src/main/java/ai/grakn/redisq