RabbitMQ EventingBasicConsumer 丢失消息答案

【问题标题】：RabbitMQ EventingBasicConsumer lose messagesRabbitMQ EventingBasicConsumer 丢失消息
【发布时间】：2017-07-02 11:42:11
【问题描述】：

我正在开发一个使用 Rabbit MQ 进行消息传递的应用程序。我使用 EventingBasicConsumer、Exchange 和 QueueBind。为了测试，我运行我的应用程序来接收消息，然后关闭我的以太网控制器。在我打开它后，重新开始接收消息，但大约有 500 条消息丢失。我的代码：

private void DoWork()
    {
        try
        {
            var connection = ConnectionFactory.CreateConnection();
            IModel model = connection.CreateModel();

            // Configure the Quality of service for the model. Below is how what each setting means.
            // BasicQos(0="Dont send me a new message untill I’ve finshed",  1= "Send me one message at a time", false ="Apply to this Model only")
            model.BasicQos(0, 1, false);

            model.ExchangeDeclare(Options.RabbitConnectionOptions.Exchange, RabbitConstants.ExchangeType, RabbitConstants.ExchangeDurable, RabbitConstants.ExchangeAutoDelete);
            var queueDeclareOk = model.QueueDeclare("SomeSubsruberQueue3", RabbitConstants.QueueDurable, RabbitConstants.QueueExclusive, RabbitConstants.ExchangeAutoDelete);


            var queueName = queueDeclareOk.QueueName;

            foreach (var optionsBindingKey in Options.BindingKeys)
            {
                Logger.Debug($"QueueBind for {nameof(optionsBindingKey)}={optionsBindingKey}");
                model.QueueBind(queueName, Options.RabbitConnectionOptions.Exchange, optionsBindingKey);
            }


            var consumer = new EventingBasicConsumer(model);

            consumer.Received += (ch, ea) =>
            {
                try
                {
                    var body = ea.Body;
                    var message = Encoding.UTF8.GetString(body);
                    var routingKey = ea.RoutingKey;
                    var messageToLog = $" [x] Received '{routingKey}':'{message}'";                        

                    Logger.Info(messageToLog);
                    Console.WriteLine($"receavedCount={++receavedCount}");

                }
                catch (Exception e)
                {
                    Console.WriteLine(e);
                    throw;
                }
            };

            model.BasicConsume(queueName, RabbitConstants.QueueAutoAck, consumer);
        }
        catch (Exception e)
        {
            Logger.Error(e);
            Console.WriteLine(e);
            throw;
        }
    }

如何防止丢失消息？

附：这里的日志显示丢失的消息。 “id”是连续的，所以你可以看到，id 从 926 到 1299 的按摩丢失了

2017-02-14 10:52:01.4916 <Root><Subscribers><Subscriber Id="922" Name="1c" />
2017-02-14 10:52:01.4916 <Root><Subscribers><Subscriber Id="923" Name="1c" />
2017-02-14 10:52:01.4916 <Root><Subscribers><Subscriber Id="924" Name="1c" />
2017-02-14 10:52:01.5056 <Root><Subscribers><Subscriber Id="925" Name="1c" />
2017-02-14 10:52:22.5606 <Root><Subscribers><Subscriber Id="1300" Name="1c" />
2017-02-14 10:52:22.5606 <Root><Subscribers><Subscriber Id="1301" Name="1c" />
2017-02-14 10:52:22.5676 <Root><Subscribers><Subscriber Id="1302" Name="1c" />
2017-02-14 10:52:22.6046 <Root><Subscribers><Subscriber Id="1303" Name="1c" />

UPD：

如果我修改：

model.BasicConsume(queueName,true, consumer);

到

model.BasicConsume(queueName,false, consumer);

并使用明确的 ASK

((EventingBasicConsumer)ch).Model.BasicAck(ea.DeliveryTag, false);

我的行为很奇怪：

2017-02-14 13:06:35.9835| DeliveryTag=23, <Subscriber Id="778" Name="1c" /> 
2017-02-14 13:06:36.0295| DeliveryTag=24, <Subscriber Id="779" Name="1c" /> 
2017-02-14 13:06:57.3285| DeliveryTag=26, <Subscriber Id="782" Name="1c" /> 
2017-02-14 13:06:57.3755| DeliveryTag=27, <Subscriber Id="783" Name="1c" />

没有 DeliveryTag=25 ！

【问题讨论】：

loose = 它的宽松，loose = 丢失.....

标签： c# .net rabbitmq amqp rabbitmq-exchange

【解决方案1】：

不要使用自动确认。阅读一条消息，对它做任何你想做的事情，然后当你完成它时明确地确认。

【讨论】：

【解决方案2】：

Ок，谢谢 Emil Vikström，我必须使用明确的 ASK：

model.BasicConsume(queueName,false, consumer);

并在处理消息后询问：

((EventingBasicConsumer)ch).Model.BasicAck(ea.DeliveryTag, false);

关于丢失消息的其他问题：rabbitMq 在将消息返回到队列时重新排序消息。所以它不会丢失，它只是稍后会收到

【讨论】：