RabbitMQ_Consumer Acknowledgements and Publisher Confirms(6)

Posted on | In | Visitors

Background:

Sometimes the consumers may crash or have connection issues. In this case,We may miss the queue messages if we set the acknowledgement to auto confirm, Please ses the following code:
Receive project code:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
ConsumeMsg(args, "DirectTestQueue");
 static void ConsumeMsg(string[] args, string queueName)
     {
         var factory = new ConnectionFactory() { HostName = "localhost" };
         using (var connection = factory.CreateConnection())
         {
             using (var channel = connection.CreateModel())
             {
                 channel.QueueDeclare(queue: queueName, durable: false, exclusive: false, autoDelete: false, arguments: null);
                 var consumer = new EventingBasicConsumer(channel);
                 consumer.Received += (model, ea) =>
                 {
                     var message = Encoding.UTF8.GetString(ea.Body);
                    Console.WriteLine("queue:"+queueName);
                     Console.WriteLine("[x] Received {0}", message);
                     Thread.Sleep(60000);
                     Console.WriteLine(" [x] Done");
                     //channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false); //Manually send message acknowledgments

                 };
                 channel.BasicConsume(queue: queueName, autoAck: true, consumer: consumer);
                 Console.WriteLine(" Press [enter] to exit.");
                 Console.ReadLine();
             }
         }
     }



This consumer will sleep 1 minutes once consumer receive th message. Then please press CTR+C to quit this console application,you will find no message in the queue, We have missed the message due to consumer issues. How to handle this scenario?

  1. We only need to do a bit changes for our codeļ¼ŒPlease see the code as below:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
static void ConsumeMsg(string[] args, string queueName)
       {
           var factory = new ConnectionFactory() { HostName = "localhost" };
           using (var connection = factory.CreateConnection())
           {
               using (var channel = connection.CreateModel())
               {
                   channel.QueueDeclare(queue: queueName, durable: false, exclusive: false, autoDelete: false, arguments: null);
                   var consumer = new EventingBasicConsumer(channel);
                   consumer.Received += (model, ea) =>
                   {
                       var message = Encoding.UTF8.GetString(ea.Body);
                      Console.WriteLine("queue:"+queueName);
                       Console.WriteLine("[x] Received {0}", message);
                       Thread.Sleep(60000);
                       Console.WriteLine(" [x] Done");
                       channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false); //Manually send message acknowledgments

                   };
                   channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer);
                   Console.WriteLine(" Press [enter] to exit.");
                   Console.ReadLine();
               }
           }
       }

2.Run the receive project and quit this console application once the message found in the screen

2.Check the queue via RabbitMQ management once you quit this console application,You will find this queue message is still in the queue