Pub-Sub 模式和消息代理，如何确保所有订阅者完成对事件的处理

Posted 2023-03-10

【中文标题】Pub-Sub 模式和消息代理，如何确保所有订阅者完成对事件的处理

【英文标题】：Pub-Sub pattern and message broker, how to make sure all subscribers finished working on event

【发布时间】：2020-09-27 14:32:48

【问题描述】：

首先，我知道有一些好的（和轻量级的）消息代理可用，比如 NATS。如果这是一份工作，我当然会选择经过验证的解决方案，这更多是出于好奇心和理解的意愿。

假设我想构建一个像 CRM 这样的系统，并且我想将它基于微服务，以便它易于扩展并且可以适应工作负载。因为微服务应该是解耦的。进来了 pub-sub。为了使发布订阅按预期工作（发布者和订阅者的解耦），我需要一个消息传递系统。假设我想用 node.js 来实现这一点（充分意识到有很多更快的方法可以完成这项工作）。

我的“问题”或可能只是认知上的失败是我想知道如何确保所有订阅者都收到来自他们订阅的主题的消息？

客户端/前端向代理发送事件请求。代理可能会验证消息并将其放入预期的队列中。有 2 个微服务订阅了此队列。代理现在只是发送队列中最旧的事件，并回调两个微服务。

当其中一个微服务明显慢于另一个时，这不会导致问题吗？

我的意思是，只要我不想发回表明任务已由所有订阅者完成的确认消息，它就应该工作。客户端不知道事件请求涉及多少服务，因此无法跟踪它。所以需要broker来做。

这是否意味着，我需要将其包含在消息代理中？它会跟踪给定事件的订阅服务计算状态吗？

【参考方案1】：

经过更多研究并在床上醒着几个小时后，我得出的结论是，如果发布者想要接收响应/确认以跟踪状态，那么让多个订阅者订阅一个主题/主题应该被视为不好的做法发送的请求/消息/事件。

经过深思熟虑后，我得出的结论是，对于同一主题的多个订阅服务很可能永远不需要 - 至少在我的情况下，只要我正确设计服务。我能想到的唯一方案是在稍后的时间点添加某些功能，而不涉及已经部署的服务。这感觉像是对不合适的服务设计的修复。

然后我想了无论如何我该如何管理它并想出了 3 种方法。

首先是标准结构

我想不需要进一步的解释。不要介意一些方法的细节，这只是一个头脑风暴的版本，绝对不理想。显示模式就足够了。

方法 1 - 聚合器收集响应

由于代理跟踪每个订阅者，它总是知道（或可以轻松计算）预期响应的数量。因此，它可以将响应消息重定向到聚合器主题，该主题在发送/发布需要响应或成功消息的消息时自动创建（想想一些客户数据的更新 - 您显然想知道消息已收到通过并成功处理）。

当然，即使只有一个响应返回，聚合器也可能总是介于两者之间。这将减少要覆盖的案件数量。聚合器基本上是某种代理。但它仍然增加了 Broker 的复杂性。

方法 2 - 代理发布确认消息

首先：不要介意右边的连接混乱。它对我来说是一个草图，但远非整洁。

正在发布的每条消息都由代理以确认消息来回答。该消息被放置在消息单独的主题堆栈中。由于 Broker 知道每个 Subject 有多少个订阅者，它可以发回一个 Publisher 应该期望多少个响应。确认消息通常也有助于通知发布者他们的消息/事件/请求是否被接受（在这里考虑身份验证和授权模式）。

只要发布者始终需要响应，这将起作用。如果它没有消息可能会停留很长一段时间。超时可以解决这个问题。

方法 3 - 传输协议响应

这与方法 2 非常相似，不同之处在于传输协议用于通知发布者已发送请求的状态和预期的潜在响应数量。

由于大多数（如果不是所有）适用于这种拓扑的协议都提供了某种方式的响应消息，并且无论如何都应该使用这些方式来验证消息是否已成功发送，因此答案还可以包含一个有效负载通知客户不仅关心传输是否成功，还关心预期会有多少响应。

结论

我想说聚合器方法的开销太大，它需要更多的额外代码，而不仅仅是使用传输协议或消息系统本身。聚合器很有趣，因为客户端可以完全忽略服务，因此是解耦的。

消息系统的使用对于日志记录（潜在的调试）和 Sagas 的实现（事件链）也很有趣。

注意

我不提倡将这些方法中的任何一种作为最佳实践。我只想用我的研究结果来回答我自己的问题。

【参考方案2】：

正如您通过自己的思考发现的那样，使用 1 个队列很难实现 pub-sub/topic 模式，因为这样 1 个队列将不得不跟踪每个订阅者的消息。这对 1 个队列有很大的责任。

通常，pub-sub/topic 模式是使用多个队列实现的：

一个用于传入消息 每个订阅者（订阅者队列）一个。

订阅者队列充当特定地址的邮箱。如果您有 5 个订阅者，那么您将有 5 个订阅者队列。

代理将根据订阅者队列重新填充的配置速率将消息从传入队列分发到每个订阅者队列。

这允许每个队列仅处理 1 个特定订阅者，并且更容易跟踪订阅者通过确认消费的内容。此外，每个订阅者都可以按照自己的节奏使用其订阅者队列中的消息。