在发布/订阅和消息传递队列中为多个消费者提供消息传递保证

要求

系统发生一些状态更改，并且系统的其他多个部分也必须知道这一点（我们称它们为观察者），以便他们可以基于当前状态执行某些操作，观察者的操作很重要，如果某些操作的观察者不在线（由于某些问题当前不收听，但是很快会回来），在所有观察者获得消息之前，不应丢弃该消息。

尝试使用pub / sub模型完成此操作，这是我的发现（（如果理解错误，请更正）-

发布者针对特定主题创建事件，多个订阅者可以使用 same 消息。此模型要么不提供任何传递保证（以redis形式），要么提供一次传递保证（使用消息队列），即。当使用者之一确认一条消息时，该消息将被丢弃（rabbitmq）。

示例

在数据库中创建了一个新的Person Profile实体

现在

现在，这两项任务都很重要，不相关并且可以并行运行。

现在，在队列模型中，如果订阅服务由于某种原因而关闭，则BG验证过程将确认该消息，该消息将从队列中删除，或者如果它像大多数pub / sub一样被触发而忘了传递，无论如何都不能保证这两种服务。

还有一点是，两个任务都不相关，不需要一个接一个地触发。

简而言之，我需要确保所有消费者都收到相同消息，并且他们应该能够分别确认他们，消息应该是只有在所有消费者都承认之后才驱逐以上两种方法中的任何一种都不这样做。

这里我想念什么吗？我应该如何解决这个问题？

如前所述，Redis Pub / Sub数据结构无法控制。

但是您可以使用Redis Streams轻松地做到这一点。

流将允许您使用XADD命令发布消息，然后控制哪些使用者正在处理该消息并确认该消息已被处理。

您可以查看这些示例应用程序，这些示例应用程序提供了有关Java的示例：

链接：

Getting Started with Redis Streams and Java
Redis Streams in Action （该项目显示了如何使用ADD / ACK / PENDING / CLAIM以及使用Redis Streams和SpringData构建防错流应用程序）

RabbitMQ's model明确支持此方案，该方案将“交换”与“队列”分开：

这里要理解的关键是，当我们谈论消费者“订阅”队列时，“ pub-sub”设置的“订阅”部分实际上是从交换机到队列的路由。

因此RabbitMQ pub-sub系统可能看起来像这样：

在数据库中创建了一个新的Person Profile实体
此事件作为消息发布给“事件” topic exchange，路由键为“ entity.profile.created”
交换将消息的副本路由到多个队列：
- 已将“ verification_service”队列绑定到此交换，以接收与“ entity.profile。＃”匹配的所有消息的副本。
- 已将“ subscription_setup_service”队列绑定到此交换，以接收与“ entity.profile.created”匹配的所有消息的副本。
正在使用的脚本对此路由一无所知，它们只知道消息将出现在队列中，用于与其相关的事件：
- 验证服务在“ verification_service”队列中提取消息的副本，进行处理并确认
- 订阅设置服务在“ subscription_setup_service”队列中获取消息的副本，进行处理并确认
如果有多个使用脚本查看 same 队列，它们将在它们之间共享消息在该队列中，但仍完全独立于其他任何队列。

下面是this interactive visualisation tool的屏幕截图，显示了这种情况：