将 pubsub 消息转换为 websocket 事件

Posted 2023-02-16

【中文标题】将 pubsub 消息转换为 websocket 事件

【英文标题】：Converting pubsub messages to websocket events

【发布时间】：2021-11-25 05:49:58

【问题描述】：

当前架构

我有一个为特定资源生成事件的微服务（我们称之为发布者）。有一个 Web 应用程序显示资源的更新列表，但此视图不直接与微服务对话。每当 Web 应用程序请求资源列表时，都会有一个视图服务调用发布者。 为了显示最新的更新，Web 应用程序使用长轮询方法调用视图服务，视图服务调用发布者（视图服务做的不止这些，例如从不同的其他来源收集数据，但这可能不相关这里）。发布者还使用视图服务当前未使用的 pubsub 发布所有资源的任何更新。

建议的更改

为了减少 API 调用并提高系统效率，我正在尝试实现 websockets 而不是长轮询。从理论上讲，它的工作方式就像网络应用程序将订阅来自视图服务的事件。视图服务将监听来自发布者的资源更新事件，每当发布订阅主题有任何消息时，它都会向网络应用发送一个事件。

问题

我现在遇到的 websocket 问题是视图服务是使用 Kubernetes 部署的，因此可以同时运行多个 pod。 Web 应用程序的不同实例可以监听来自不同 pod 的事件，因此可能会发生 pubsub 消息被 pod A 接收，但需要此资源的 websocket 侦听器连接到 pod B。如果 pod A ack消息丢弃它，因为它找不到任何事件侦听器，我们将丢失事件并且网络应用程序将没有更新的数据。如果 pod A nack 消息，以便它可以被任何其他可能受益于该消息的 pod 侦听，则可能没有 pod 具有任何 websocket 事件侦听器可以从该消息中受益，并且该消息将继续循环永远阻塞 pubsub 队列。

可能的解决方案

我想到的第一个解决方案是为不同的 pod 创建不同的订阅，这样每个 pod 将至少接收一次相同的事件，并且我们不会阻塞 pubsub 队列。然而，这种方法的挑战在于 Pod 随时可能死亡，从而导致订阅被遗弃，几周后我将处理大量消息溢出的废弃订阅。

另一种解决方案是拥有一个存储 pubsub 消息的数据库，不同的 pod 会查询它以接收事件，定期检查任何侦听器并将其从那里删除，但是当没有侦听器时它并不能解决问题事件的监听器。此外，我不想仅仅因为这个问题而添加数据库（当前的长轮询架构比这要好得多）。

第三种解决方案是在发布者内部实现 websocket，但是，这是最不可能的解决方案，因为那里的代码库非常庞大，没有人喜欢在那里添加任何新功能。

最终的解决方案是始终只拥有一个视图服务 pod，但这违背了拥有微服务和在 Kubernetes 上的目的。此外，它不可扩展。

问题

是否有任何推荐的方式或任何其他方式我可以使用 websockets 将 pubsub 事件连接到 web 应用程序而不会增加不必要的复杂性？如果其他地方有可用的示例，我会喜欢一个示例。

【问题讨论】：

您也可以尝试在 softwareengineering.stackexchange.com 上发布您的问题，因为它在某种程度上是一个设计/咨询问题。

【参考方案1】：

没有简单的解决方案。首先，在 websocket 模式中，pod 负责将事件发送到 web 应用程序。因此，将正确的后端事件收集到 Web 应用程序。在这个设计中，pod 需要过滤正确的消息来传递。

最幼稚的实现是在所有 pod 中（因此在所有订阅中）复制所有消息，但这并不是真正有效且耗时（除了丢弃所有消息很耗时）。

---

我们可以想象一个更有效的解决方案，并在每个 pod 上创建一个订阅列表，每个打开的 webapp 频道一个。在这些订阅上，您可以添加过滤器参数。当然，发布者需要添加一个属性以允许订阅对其进行过滤。

会话结束后，必须删除订阅。

如果发生崩溃，我建议这种模式：在数据库中存储订阅 ID、过滤器内容（webapp 会话）和负责过滤和交付的 pod ID。然后，检测 pod 崩溃，或者运行一个调度的 pod，检查是否所有正在运行的 pod 都注册到数据库中。如果一个 pod 在数据库中并且没有运行，则删除所有相关的订阅。

如果您能够实时检测到 pod 崩溃，您可以将活动的 webapp 会话分派给其他正在运行的 pod，或者在新创建的 pod 上。

---

如您所见，设计并不简单，需要控制、检查和垃圾回收。

【讨论】：

您好，感谢您的回答。这与我提出的类似。第二种方法似乎更好，但我真的不想要与服务关联的数据库，除非没有其他选择。即使在这种情况下，人们也更愿意使用当前的长轮询系统，而不是为视图服务添加单独的数据库。