architecture - 已发布并等待对 RabbitMQ/EasyNetQ 主题的回应 - 如何只获得你的？

Question

当发布者希望消息得到答复时，如何确保在您将其扩展时它只会获得(对自己的消息)相关的答复？

我们有一个客户端进程发布一条消息供服务器进程回答。此外，我们有一个“监听器”进程，它只需要在不发布任何内容的情况下同时使用问题和答案。此外，服务器进程将来可能会分解为多个进程，从而创建消息级联。我们不能使用请求/响应，因为我们需要监听器，然后再一次，当我们有级联时......此外，我们将有几个问题/答案类别，EasyNetQ 中的请求/响应不支持主题.

我们使用 EasyNetQ 的解决方案是简单的基于主题的发布/订阅:客户端发布到“question”主题，订阅“answer”，服务器订阅“question”并发布到“answer”，而听众只需订阅两者。

问题出在扩展客户端时。它的两个实例现在发布问题，但由于它们都订阅了一个“answer”主题，一个人可能会得到另一个实例发布的问题的答案，而不是自己的。

我们找到的解决方案是让客户端在订阅“answer”时使用一个唯一命名的队列——这样每个客户端都会得到所有的答案，只需要忽略那些不是他的。但是，此解决方案存在一些性能缺陷，并且还会导致每次客户端崩溃(或在开发期间重新启动等)时，唯一命名的队列在 RabbitMQ 中累积。

客户端，发送对象消息:

string corrId = Guid.NewGuid().ToString();

// Register the corrId in a dictionary
//...

var myMessage = new MyMessage {correlationId =corrId, realMessage = msg};
easyNetQBus.Subscribe<MyMessage>("mqClient"+uniqueSuffix, HandleMsg, x => x.WithTopic("answer"));
easyNetQBus.Publish(myMessage, "question");

// In HandleMsg, we see if we have issued questions with the correlation id that came with the answer (lookup in the dictionary) and if not, ignore it

服务器:

easyNetQBus.Subscribe<MyMessage>("mqServer", HandleMsg, x => x.WithTopic("question"));

// In HandleMsg, we publish the answer back to "answer" with the correlation id from the question

我们应该使用另一种模式吗？我们可以在每条消息中放入一个独特的主题/队列来发送答案，但这会使听众的生活和我提到的级联中 future 参与者的灵 active 变得复杂......

Answer 1

解决问题的一种方法是使用 Remote Procedure Call pattern .这提供了一种简单、可扩展的方式来为每个客户端提供一个独特的队列。

来自示例:

• 当客户端启动时，它创建一个匿名的独占回调队列。对于 RPC 请求，客户端发送一条消息，其中包含两个properties:ReplyTo，设置为回调队列和CorrelationId，为每个请求设置一个唯一值。
• 请求被发送到 rpc_queue 队列。 RPC worker(又名:服务器)正在等待该队列上的请求。当一个请求出现，它完成工作并将结果发送回消息客户端，使用来自 ReplyTo 属性的队列。
• 客户端等待回调队列中的数据。当一个消息出现时，它会检查 CorrelationId 属性。如果匹配来自请求的值，它将响应返回给申请。

如果您的回调队列是自动删除的排他队列 ( RabbitMQ queue documentation )，那么当客户端或服务器重新启动时，它们不应堆积起来。

With some workloads queues are supposed to be short lived. Whileclients can delete the queues they declare before disconnection, thisis not always convenient. On top of that, client connections can fail,potentially leaving unused resources (queues) behind.

There are three ways to make queue deleted automatically:

• Exclusive queues (covered below)
• TTLs (also covered below)
• Auto-delete queues

An auto-delete queue will be deleted when its last consumer iscancelled (e.g. using the basic.cancel in AMQP 0-9-1) or gone(closed channel or connection, or lost TCP connection with theserver).

If a queue never had any consumers, for instance, when all consumptionhappens using the basic.get method (the "pull" API), it won't beautomatically deleted. For such cases, use exclusive queues or queueTTL. (RabbitMQ queue documentation)