防止多个 NodeJS 实例处理 Redis PubSub 事件答案

【问题标题】：Prevent multiple instances of NodeJS handling Redis PubSub event防止多个 NodeJS 实例处理 Redis PubSub 事件
【发布时间】：2020-11-03 08:51:03
【问题描述】：

我有一个有趣的难题；我使用 Redis PubSub 模式在我的 NodeJS 微服务网络中发出事件。我离开了 RabbitMQ，因为我需要多播消息，以便多个微服务可以接收和处理相同的事件。因此，如果我发出 LOGOUT 事件，UserService 和 WebSocketService 都应该听到它。

在我的客户端部署在每个微服务的多个实例运行的环境中之前，这一直很好。现在，给定 3 个 UserService 实例，所有实例都连接到 Redis，它们都接收并处理 LOGOUT 事件，这很糟糕。

因此，一方面我需要不同的微服务来听到事件，但另一方面我需要防止同一微服务的重复副本全部处理事件。摇滚，遇到困难的地方。

到目前为止，我最好的想法有点老套，但类似于：

不引发事件，而是将事件列表写入 Redis 缓存
UserService 和 WebSocketService 可以每 3 秒读取一次该列表并检查需要处理的新事件
找到相关事件后，UserService 会将其“名称”添加到正在处理该事件的服务列表中
当 WebSocketService 看到事件时，它仍然能够处理它并将其“名称”添加到处理程序列表中
当 UserService 的重复实例看到该事件时，它会在处理程序列表中看到其“名称”并忽略该事件

我不喜欢这个解决方案，因为列表会在内存中不断增长，而不是在短暂的消息中。此外，我必须开始添加代码来管理已检查的事件；否则每个周期，整个列表都必须由所有服务的所有实例再次解析。

欢迎提出想法。

【问题讨论】：

标签： node.js redis publish-subscribe

【解决方案1】：

收到通知后，使用 LPOP 使用 Redis 上的事件列表，并且仅在该列表中确实有内容时才采取行动。

这将清理事件列表并保证只有一个消费者会实际执行操作。

【讨论】：

谢谢；我想你误会了。我不想保证每个事件有一个处理程序。在上面的示例中，我希望 UserService 和 WebSocketService 都独立处理事件。我要保证的是 UserService 的重复实例不处理该事件（我有多个相同服务的进程正在运行）。只有一个 UserService 实例和一个 WebSocketService 实例应该处理该事件。
好的。在我的脑海中，我会让 2 个不同的服务消耗 2 个 dirrerent 事件列表。我认为根据 Pub/Sub 的通知采取行动是一种反模式，消费者应该对事件采取行动，而 Pub/Sub 只是为了防止客户轮询列表并更快地做出反应。

【解决方案2】：

对于您描述的场景，我将使用两种模式：使用redlock 实现的事件队列（或更具体地说，每个消费者集群的事件队列）和 PubSub 模式来提醒所有人消费者知道一个新的事件已经准备好被消费了。考虑到您提出的解决方案，这应该更具反应性，因为没有轮询，但对队列的访问是事件驱动的。

【讨论】：

你能说得更具体一点吗？具体来说，发射器不应该知道消息是给谁的，所以当他发射并且同一微服务的所有重复实例都接收到事件时，会发生什么？谁排队？
发射器应该同时执行@BeetleJuice 操作。首先将事件排入队列（或队列中：一个用于 UserService 集群，一个用于 WebSocketService 集群），然后发布该事件。所有消费者都将收到该事件：对于每个集群，第一个能够从其特定队列（可能是工作负载较低的队列）中将事件出列的人将处理它。
感谢您的想法。来自用户 vins 的 Redis Streams 建议似乎是对问题陈述的更自然的解决方案

【解决方案3】：

Redis streams 似乎对您的用例有所帮助。你检查了吗？

我们可以有多个消费者组。与 pub/sub 一样，每个消费者组都应该从流中接收消息。（这里的流就像发布/订阅频道/主题）。但是，当一个消费者组中有多个节点时，只有一个节点会处理该事件。不是全部。节点可以确认它已处理的消息。

重启后，节点不会处理已发送的消息
流将继续增长。我们可以根据大小进行修剪。我们可以保留最新的 N 个元素。

【讨论】：

谢谢分享。到目前为止，这似乎是解决问题的原生方式

【解决方案4】：

嘿，我在多个实例上运行服务器后遇到了同样的问题。

就我而言，我使用 mongodb 作为数据库，所以每当我收到 keySpace 通知我为此创建了一个写锁到我的数据库特定订阅。

所以我自己创建了一个写锁。每当我订阅您订阅的那个特定密钥我增加了密钥关于那个特定的文件 1. 通过更新我得到 nModified:1 在我的情况下，一旦它通过创建一些被修改相应的条件。

所以一旦服务器通过创建一些条件发送多个请求我设法输入了该条件并更新了该密钥，并且一次执行我想要的任何任务。

您可以通过创建自己的写锁来做到这一点。

冰雹编码

【讨论】：