已发布并等待 RabbitMQ / EasyNetQ 主题的回复 - 如何仅获取您的主题？

当发布者期望得到消息的答案时，如何确保在扩展时它只能得到（与其自己的消息相关的）答案？

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

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

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

我们找到的解决方案是让客户端在订阅“时使用唯一命名的队列”answer“ - 这样每个客户端都会得到所有的答案，只需要忽略不属于他的答案。但是，这个解决方案有一些性能缺陷，并且还会导致每次客户端崩溃（或只是在开发过程中重新启动等）。

客户端，发送对象消息：

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

Server:

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

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

解决你的问题的一种方法是使用远程过程调用模式 https://www.rabbitmq.com/tutorials/tutorial-six-dotnet.html。这提供了一种简单、可扩展的方式来为每个客户端提供唯一的队列。

从例子来看：

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

如果您的回调队列是自动删除的，则独占队列（RabbitMQ 队列文档 https://www.rabbitmq.com/queues.html），那么当客户端或服务器重新启动时，它们不应该堆积起来。

对于某些工作负载，队列应该是短暂的。尽管 客户端可以删除他们在断开连接之前声明的队列，这 并不总是方便。最重要的是，客户端连接可能会失败， 可能会留下未使用的资源（队列）。

有3种方法可以让队列自动删除：

• 独占队列（如下所述）
• TTL（下面也有介绍）
• 自动删除队列

当最后一个消费者被删除时，自动删除队列将被删除 取消（例如使用basic.cancel在 AMQP 0-9-1 中）或消失 （关闭通道或连接，或丢失与 服务器）。

如果一个队列从来没有任何消费者，例如，当所有消费 发生使用basic.get方法（“拉”API），它不会 自动删除。对于这种情况，请使用独占队列或队列 TTL。 (RabbitMQ 队列文档 https://www.rabbitmq.com/queues.html)