RabbitMQ - 了解消费者已启动答案

【问题标题】：RabbitMQ - Knowing Consumer is upRabbitMQ - 了解消费者已启动
【发布时间】：2016-07-05 17:30:55
【问题描述】：

我有几个关于 RabbitMQ 编码的问题...我对这个世界并不陌生，并且有一些问题是基于提供给我的用于实现的设计...

如果我从消费者发送 BasicAck 或 BasicAack，它是否只删除来自队列的相应消息还是将其传递给发布者？
如何确保 Publisher 仅在 Consumer 准备好进行处理时才向 Server 发送消息？

设计上说 Publisher 需要等待并知道 Consumer 的处理何时完成才能在客户端执行某些任务（取决于成功/失败）。

我已经尝试了下面的代码，但出队立即从队列中删除了消息，而我没有发送任何 Ack 或 Nack。我很困惑

发布者代码：

using (var connection = factory.CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    channel.QueueDeclare("test", durable, false, false, null);
                    channel.TxSelect();
                    var properties = channel.CreateBasicProperties();
                    properties.SetPersistent(true);

                    string message = "Hello World!";
                    var body = Encoding.UTF8.GetBytes(message);



                    channel.BasicPublish("", "test", properties, body);
                    channel.TxCommit();
                    Console.WriteLine(" [x] Sent {0}", message);
                }
            }

消费者代码

 using (var connection = factory.CreateConnection())
        {
            using (var channel = connection.CreateModel())
            {
                channel.QueueDeclare("test", durable, false, false, null);

                var consumer = new QueueingBasicConsumer(channel);
                channel.BasicConsume("test", true, consumer);

                Console.WriteLine(" [*] Waiting for messages." +
                                         "To exit press CTRL+C");
                while (true)
                {
                    var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();

                    var body = ea.Body;
                    var message = Encoding.UTF8.GetString(body);
                    Console.WriteLine(" [x] Received {0}", message);
                }
            }
        }

注意：我意识到channel.BasicConsume("test", true, consumer); 的 noAck 为 true。我把它改成了channel.BasicConsume("test", false, consumer);

当我使用channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false); 时，我可以看到消息已从队列中删除但是，Publisher 是如何知道 Consumer 处理成功的呢？

【问题讨论】：

标签： c# rabbitmq

【解决方案1】：

如何确保发布者仅在消费者准备好处理时才向服务器发送消息？

你不能。更重要的是，你不应该。使用消息传递架构的目的是忘记这类问题。检查this。

此外，如果队列是持久的，RabbitMQ 将为您存储这些消息，直到有人准备好处理它们。

但是，Publisher 怎么知道 Consumer 处理成功了呢？

不，不会。 ack 仅在 RabbitMQ 和您的消费者之间或 RabbitMQ 和您的生产者之间。请查看this 了解有关 ack/nack 的一些详细信息。

如果我能正确理解你的话，你想在这里实现的是一种“健谈”架构，其中消费者也是发布者的“响应消息”的生产者，比如“嘿，我是消息 XX 完成，一切正常”。

执行此操作的简单方法是将您的消费者设置为生产者，并将您的生产者设置为消费者。您只需要在消息中添加一个 guid 或某种唯一 id，当您完成它时，您将在另一个队列上发送一条消息，该 id 作为内容通知原始发布者（即消费者这个“响应”队列），工作已成功完成。

希望对你有帮助:)

【讨论】：

谢谢；我知道您已经为我的问题提供了解决方案；哪一种是标准或传统方式实现？我希望是第一个，但不是“健谈”架构！
如果您需要让生产者知道您的消费者“完成”了工作，则需要在频道上做出响应；响应可以在同一个队列或另一个队列上，但要发送另一条消息。否则，您可以检查 sagas，检查此 docs.masstransit-project.com/en/latest/overview/saga.html 但有点复杂（并且您需要公共交通）。也许您可以重新考虑您的架构，试图消除响应消息的必要性:)
我完全同意你的回复；这取决于我们的建筑师想要什么;）感谢您的帮助

【解决方案2】：

这是一个重复的问题。检查这个线程，这是一个正确答案的类似问题：Why doesn't Channel.waitForConfirmsOrDie block?

【讨论】：

【解决方案3】：

或者你可以使用 RPC 模式

https://www.rabbitmq.com/tutorials/tutorial-six-dotnet.html

【讨论】：