Rabbitmq-设计消息重放服务

Question

我正在尝试设计一种重播机制，使用户能够重播队列中的消息。 我为包含多个队列和多个消费者的交换提出的最佳设计是：

1. 创建一个记录器服务，它将：

   • 创建一个队列并将所有路由键绑定到它。
   • 消耗来自交易所的所有消息。
   • 将所有消息保存到数据库。

2. 订阅者请求重播。

   • 每个订阅者创建一个新的交换器、队列，并使用与其常规队列相同的绑定来绑定它。
   • 订阅者向 Web 服务器发送休息请求，以使用过滤器（开始日期等）开始重播。请求包含其重播交换名称。
   • Web服务器从DB中提取数据并将其发布到特定交易所
   • 可以添加改进，例如附加 RequestId 并回显它。

问题：
1. 这有道理吗？
2.我是在发明轮子吗？有没有兔子固有的解决方案？插件？
3. 创建多个交易所是否是一种好的做法？
在此解决方案中，为每个队列创建一个交换器，以便发布相同的消息。

另一种解决方案：
1.为每个队列创建一个附加队列“ReplayQueue”。设置 TTL（假设一个月）。
2. 每次用户请求重放时，让他从自己的 ReplayQueue 中重放，而不需要确认。

这个解决方案有点问题，因为。

• 为了重播最后一天，消费者必须获取之前 29 天的所有内容并将其过滤掉。
• 此解决方案可扩展 - 队列将变得更大（与可以扩展的数据库存储不同）。

Answer 1

1. 这有道理吗？

是的

2. 我在发明轮子吗？有没有兔子固有的解决方案？插件？

你并不是在重新发明轮子。据我所知，没有兔子解决方案，也没有开箱即用的解决方案。

我认为你的第一个解决方案很好。

Another solution

是非常有问题的，因为健康的兔子是空的，而兔子不是数据存储。

您将有一个队列 (

STORE

)，所有已发布的消息都应路由到该队列。您可以考虑使用主题交换，而不是将

STORE

与所有绑定键绑定。代价是绑定键不能包含

. # *

以及消息路由时的轻微开销。

STORE

队列将与绑定键

#

绑定。

你可以看看firehose。

为什么要进行网络请求？您可以使用 Rabbit 进行 RPC 调用:

• 订阅者发送 amqp 请求以使用过滤器（开始日期等）开始重播。
• 请求包含

  reply-to

  队列名称。该队列可能是客户端和请求独占的。
• RPC 服务器从数据库中提取数据并将其发布到回复队列

您还可以查看直接回复模式。

3. 创建多个交易所是否是一种好的做法？

是的，只要您需要就可以。对于您的具体情况，我认为没有必要为每个订阅者进行交换。该请求已包含队列名称。您可以简单地使用默认交换器

amq.direct

发布到此队列，路由键等于队列名称。如果您想要一个交换，我将创建一个唯一的交换（例如“重播”），并让每个订阅者将其重播队列绑定到该交换。 “重播”交换可以是约定或与请求一起发送。

Answer 2

第一种方案是可行的。鉴于rabbit MQ附带了

tracing plugin

，将消息存储在数据库中变得更加容易，因为它简单地归结为从绑定到

amq.rabbitmq.trace

交换的队列中进行消费。

此交换特定于

vhost

，每个虚拟主机都有自己的

amq.rabbitmq.trace

交换。此外，在创建新跟踪时，可以限制启用跟踪的队列/交换，从而可以灵活地在不需要时禁用跟踪。

请参阅以下链接来配置跟踪：
https://www.rabbitmq.com/firehose.html
https://www.rabbitmq.com/blog/2011/09/09/rabbitmq-tracing-a-ui-for-the-firehose/

Answer 3

自 RabbitMQ v3.9.0（2021 年 7 月 23 日）起，RabbitMQ 支持类似 kafka 的具有重放功能的主题（在 RabbitMQ 中称为时间旅行），请参阅 v3.9.0 发行说明了解发布详细信息。

检查 RabbitMQ Streams 文档