系列文章：

RabbitMQ從零到集群高可用(.NetCore5.0) - RabbitMQ簡介和六種作業模式詳解

RabbitMQ從零到集群高可用(.NetCore5.0) - 死信佇列，延時佇列

一、死信佇列

 描述：Q1佇列系結了x-dead-letter-exchange(死信交換機)為X2，x-dead-letter-routing-key（死信路由key)指向Q2(佇列2)

            P(生產者)發送訊息經X1(交換機1)路由到Q1(佇列1)，Q1的訊息觸發特定情況，自動把訊息經X2(交換機2)路由到Q2(佇列2)，C(消費者)直接訊息Q2的訊息，

特定情況有哪些呢：

1. 訊息被拒(basic.reject or basic.nack)并且沒有重新入隊(requeue=false)；
2. 當前佇列中的訊息數量已經超過最大長度(創建佇列時指定" x-max-length引數設定佇列最大訊息數量)，
3. 訊息在佇列中過期，即當前訊息在佇列中的存活時間已經超過了預先設定的TTL(Time To Live)時間；

這里演示情況1：

假如場景：Q1中佇列資料不完整，就算從新處理也會報錯，那就可以不ack，把這個訊息轉到死信佇列另外處理，

生產者：

 public static void SendMessage()
        {
            //死信交換機
            string dlxexChange = "dlx.exchange";
            //死信佇列
            string dlxQueueName = "dlx.queue";

            //訊息交換機
            string exchange = "direct-exchange";
            //訊息佇列
            string queueName = "queue_a";

            using (var connection = RabbitMQHelper.GetConnection())
            {
                using (var channel = connection.CreateModel())
                {

                    //創建死信交換機
                    channel.ExchangeDeclare(dlxexChange, type: ExchangeType.Direct, durable: true, autoDelete: false);
                    //創建死信佇列
                    channel.QueueDeclare(dlxQueueName, durable: true, exclusive: false, autoDelete: false);
                    //死信佇列系結死信交換機
                    channel.QueueBind(dlxQueueName, dlxexChange, routingKey: dlxQueueName);

                    // 創建訊息交換機
                    channel.ExchangeDeclare(exchange, type: ExchangeType.Direct, durable: true, autoDelete: false);
                    //創建訊息佇列,并指定死信佇列
                    channel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false, arguments:
                                        new Dictionary<string, object> {
                                             { "x-dead-letter-exchange",dlxexChange}, //設定當前佇列的DLX(死信交換機)
                                             { "x-dead-letter-routing-key",dlxQueueName}, //設定DLX的路由key，DLX會根據該值去找到死信訊息存放的佇列
                                         });
                    //訊息佇列系結訊息交換機
                    channel.QueueBind(queueName, exchange, routingKey: queueName);

                    string message = "hello rabbitmq message";
                    var properties = channel.CreateBasicProperties();
                    properties.Persistent = true;
                    //發布訊息
                    channel.BasicPublish(exchange: exchange,
                                         routingKey: queueName,
                                         basicProperties: properties,
                                         body: Encoding.UTF8.GetBytes(message));
                    Console.WriteLine($"向佇列:{queueName}發送訊息:{message}");
                }
            }
        }

消費者：

  public static void Consumer()
        {
            //死信交換機
            string dlxexChange = "dlx.exchange";
            //死信佇列
            string dlxQueueName = "dlx.queue";

            //訊息交換機
            string exchange = "direct-exchange";
            //訊息佇列
            string queueName = "queue_a";
            var connection = RabbitMQHelper.GetConnection();
            {
                //創建信道
                var channel = connection.CreateModel();
                {

                    //創建死信交換機
                    channel.ExchangeDeclare(dlxexChange, type: ExchangeType.Direct, durable: true, autoDelete: false);
                    //創建死信佇列
                    channel.QueueDeclare(dlxQueueName, durable: true, exclusive: false, autoDelete: false);
                    //死信佇列系結死信交換機
                    channel.QueueBind(dlxQueueName, dlxexChange, routingKey: dlxQueueName);

                    // 創建訊息交換機
                    channel.ExchangeDeclare(exchange, type: ExchangeType.Direct, durable: true, autoDelete: false);
                    //創建訊息佇列,并指定死信佇列
                    channel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false, arguments:
                                        new Dictionary<string, object> {
                                             { "x-dead-letter-exchange",dlxexChange}, //設定當前佇列的DLX
                                             { "x-dead-letter-routing-key",dlxQueueName}, //設定DLX的路由key，DLX會根據該值去找到死信訊息存放的佇列
                                         });
                    //訊息佇列系結訊息交換機
                    channel.QueueBind(queueName, exchange, routingKey: queueName);


                    var consumer = new EventingBasicConsumer(channel);
                    channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: true);
                    consumer.Received += (model, ea) =>
                    {
                        //處理業務
                        var message = Encoding.UTF8.GetString(ea.Body.ToArray());
                        Console.WriteLine($"佇列{queueName}消費訊息:{message},不做ack確認");
                        //channel.BasicAck(ea.DeliveryTag, false);
                        //不ack(BasicNack),且不把訊息放回佇列(requeue:false)
                        channel.BasicNack(ea.DeliveryTag, false, requeue: false);
                    };
                    channel.BasicConsume(queueName, autoAck: false, consumer);
                }
            }
        }

消費者加上channel.BasickNack()模擬訊息處理不了，不ack確認，

執行結果：

 RabbitMQ管理界面：

看到訊息佇列為queue_a，特性有DLX(死信交換機)，DLK（死信路由），因為消費端不nack，觸發了死信，被轉發到了死信佇列dlx.queue，

二、延時佇列

延時佇列其實也是配合死信佇列一起用，其實就是上面死信佇列的第二中情況，給佇列添加訊息過時時間（TTL），變成延時佇列，

 簡單的描述就是：P(生產者)發送訊息到Q1(延時佇列)，Q1的訊息有過期時間，比如10s，那10s后訊息過期就會觸發死信，從而把訊息轉發到Q2（死信佇列），

解決問題場景：像商城下單，未支付時取消訂單場景，下單時寫一條記錄入Q1,延時30分鐘后轉到Q2，消費Q2，檢查訂單，支付則不做操作，沒支付則取消訂單，恢復庫存，

生產者代碼：

 public static void SendMessage()
        {
            //死信交換機
            string dlxexChange = "dlx.exchange";
            //死信佇列
            string dlxQueueName = "dlx.queue";

            //訊息交換機
            string exchange = "direct-exchange";
            //訊息佇列
            string queueName = "delay_queue";

            using (var connection = RabbitMQHelper.GetConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    //創建死信交換機
                    channel.ExchangeDeclare(dlxexChange, type: ExchangeType.Direct, durable: true, autoDelete: false);
                    //創建死信佇列
                    channel.QueueDeclare(dlxQueueName, durable: true, exclusive: false, autoDelete: false);
                    //死信佇列系結死信交換機
                    channel.QueueBind(dlxQueueName, dlxexChange, routingKey: dlxQueueName);

                    // 創建訊息交換機
                    channel.ExchangeDeclare(exchange, type: ExchangeType.Direct, durable: true, autoDelete: false);
                    //創建訊息佇列,并指定死信佇列，和設定這個佇列的訊息過期時間為10s
                    channel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false, arguments:
                                        new Dictionary<string, object> {
                                             { "x-dead-letter-exchange",dlxexChange}, //設定當前佇列的DLX(死信交換機)
                                             { "x-dead-letter-routing-key",dlxQueueName}, //設定DLX的路由key，DLX會根據該值去找到死信訊息存放的佇列
                                             { "x-message-ttl",10000} //設定佇列的訊息過期時間
                                        });
                    //訊息佇列系結訊息交換機
                    channel.QueueBind(queueName, exchange, routingKey: queueName);

                    string message = "hello rabbitmq message";
                    var properties = channel.CreateBasicProperties();
                    properties.Persistent = true;
                    //發布訊息
                    channel.BasicPublish(exchange: exchange,
                                         routingKey: queueName,
                                         basicProperties: properties,
                                         body: Encoding.UTF8.GetBytes(message));
                    Console.WriteLine($"{DateTime.Now},向佇列:{queueName}發送訊息:{message}");
                }
            }
        }

消費者代碼：

 public static void Consumer()
        {
            //死信交換機
            string dlxexChange = "dlx.exchange";
            //死信佇列
            string dlxQueueName = "dlx.queue";
            var connection = RabbitMQHelper.GetConnection();
            {
                //創建信道
                var channel = connection.CreateModel();
                {
                    //創建死信交換機
                    channel.ExchangeDeclare(dlxexChange, type: ExchangeType.Direct, durable: true, autoDelete: false);
                    //創建死信佇列
                    channel.QueueDeclare(dlxQueueName, durable: true, exclusive: false, autoDelete: false);
                    //死信佇列系結死信交換機
                    channel.QueueBind(dlxQueueName, dlxexChange, routingKey: dlxQueueName);

                    var consumer = new EventingBasicConsumer(channel);
                    channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: true);
                    consumer.Received += (model, ea) =>
                    {
                        //處理業務
                        var message = Encoding.UTF8.GetString(ea.Body.ToArray());
                        Console.WriteLine($"{DateTime.Now}，佇列{dlxQueueName}消費訊息:{message}");
                        channel.BasicAck(ea.DeliveryTag, false);
                    };
                    channel.BasicConsume(dlxQueueName, autoAck: false, consumer);
                }
            }
        }

執行代碼：

 向延時佇列發送訊息，監聽死信佇列，發送和收到訊息時間剛好是設定的10s，

RabbitMQ管理界面：

三、延時佇列訊息設定不同過期時間

上面的延時佇列能解決訊息過期時間都是相同的場景，能不能解決訊息的過期時間是不一樣的呢？

例如場景：機器人客服，為了更像人為操作，收到訊息后要隨機3-10秒回復客戶，

1）佇列不設定TTL（訊息過期時間），把過期時間設定在訊息上，

生產者代碼：

public static void SendMessage()
        {
            //死信交換機
            string dlxexChange = "dlx.exchange";
            //死信佇列
            string dlxQueueName = "dlx.queue";

            //訊息交換機
            string exchange = "direct-exchange";
            //訊息佇列
            string queueName = "delay_queue";

            using (var connection = RabbitMQHelper.GetConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    //創建死信交換機
                    channel.ExchangeDeclare(dlxexChange, type: ExchangeType.Direct, durable: true, autoDelete: false);
                    //創建死信佇列
                    channel.QueueDeclare(dlxQueueName, durable: true, exclusive: false, autoDelete: false);
                    //死信佇列系結死信交換機
                    channel.QueueBind(dlxQueueName, dlxexChange, routingKey: dlxQueueName);

                    // 創建訊息交換機
                    channel.ExchangeDeclare(exchange, type: ExchangeType.Direct, durable: true, autoDelete: false);
                    //創建訊息佇列,并指定死信佇列，和設定這個佇列的訊息過期時間為10s
                    channel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false, arguments:
                                        new Dictionary<string, object> {
                                             { "x-dead-letter-exchange",dlxexChange}, //設定當前佇列的DLX(死信交換機)
                                             { "x-dead-letter-routing-key",dlxQueueName}, //設定DLX的路由key，DLX會根據該值去找到死信訊息存放的佇列
                                             //{ "x-message-ttl",10000} //設定佇列的訊息過期時間
                                        });
                    //訊息佇列系結訊息交換機
                    channel.QueueBind(queueName, exchange, routingKey: queueName);

                    string message = "hello rabbitmq message 10s后處理";
                    var properties = channel.CreateBasicProperties();
                    properties.Persistent = true;
                    properties.Expiration = "10000";//訊息的有效期10s

                    //發布訊息,延時10s
                    channel.BasicPublish(exchange: exchange,
                                         routingKey: queueName,
                                         basicProperties: properties,
                                         body: Encoding.UTF8.GetBytes(message));
                    Console.WriteLine($"{DateTime.Now},向佇列:{queueName}發送訊息:{message},延時:10s");



                    string message2 = "hello rabbitmq message 5s后處理";
                    var properties2 = channel.CreateBasicProperties();
                    properties2.Persistent = true;
                    properties2.Expiration = "5000";//訊息有效期5s

                    //發布訊息,延時5s
                    channel.BasicPublish(exchange: exchange,
                                         routingKey: queueName,
                                         basicProperties: properties2,
                                         body: Encoding.UTF8.GetBytes(message2));
                    Console.WriteLine($"{DateTime.Now},向佇列:{queueName}發送訊息:{message2},延時:5s");


                }
            }
        }

消費者代碼還是上面延時佇列的不變，先試下效果，

 生產者向佇列中發送一條延時10s的訊息再發一條延時5秒的訊息，但消費者卻先拿到延時10s的，再拿到延時5秒的，我想要的結果是先拿到延時5s的再拿到延時10s的，是什么原因呢，

原因是：佇列是先進先出的，而RabbitMQ只會對首位第一條訊息做檢測，第一條沒過期，那么后面的訊息就會阻塞住等待前面的過期，

解決辦法：增加一個消費者對延時佇列消費，不ack，把第一條訊息放到佇列尾部，一直讓訊息在流動，這樣就能檢測到了，

2）新增消費者代碼：

public static void DelayConsumer()
        {
            //延時佇列
            string queueName = "delay_queue";
            var connection = RabbitMQHelper.GetConnection();
            {
                //創建信道
                var channel = connection.CreateModel();
                {
                    var consumer = new EventingBasicConsumer(channel);
                    channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: true);
                    consumer.Received += (model, ea) =>
                    {
                        //處理業務
                        var message = Encoding.UTF8.GetString(ea.Body.ToArray());
                        Thread.Sleep(20);//訊息少的時候可以加個睡眠時間減少IO
                        channel.BasicNack(ea.DeliveryTag, false, requeue: true);
                    };
                    channel.BasicConsume(queueName, autoAck: false, consumer);

                }
            }
        }

執行效果：

這會得到了想要的效果，

RabbitMQ管理界面：

標籤：.NET技术

上一篇：紙殼CMS 3.4.4 新增搜索引擎訪問日志

下一篇：記一次 .NET 某流媒體獨角獸 API 句柄泄漏分析