2. 程式人生 > 其它 >RabbitMQ死性佇列，延時佇列

RabbitMQ死性佇列，延時佇列

阿新 • • 發佈：2021-09-01

1.前言

之前一直聽說過RabbitMQ的死性佇列，延時佇列，但是沒有深入的瞭解，結合自己實際工作中使用到的RabbitMQ，現在整理出來當做個人學習筆記。

2.死性佇列介紹

　　死性佇列：DLX（dead-letter-exchange）,利用DLX，當訊息在一個佇列中變成死性（dead message）之後他能被重新push到另外一個Exchange，那麼這個Exchange就是DLX。

　　訊息變成死性一般有一下三種情況：

　　　　（1）訊息被拒(basic.reject or basic.nack)並且沒有重新入隊(requeue=false)；

　　　　（2）當前佇列中的訊息數量已經超過最大長度(建立佇列時指定" x-max-length引數設定佇列最大訊息數量)。

　　　　（3）訊息在佇列中過期，即當前訊息在佇列中的存活時間已經超過了預先設定的TTL(Time To Live)時間；

3.死性處理過程

　　　　（1）DLX也是一個正常的交換機（Exchange），和一般的Exchange沒有區別，他能在任何的佇列上指定，實際上就是設定某個佇列的屬性。

　　　　（2）當這個佇列中有死性時，RabbitMQ就會自動的將這個訊息重新發送到設定的Exchange中，進而被路由到另一個佇列。

　　　　（3）可以監聽到這個佇列（DLX）做特殊處理。

4.demo演示

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

生產者：

using RabbitMQ.Client;
using System;
using System.Collections.Generic;
using System.Text;

namespace RabbitMqClient
{
    class Program
    {
        static IConnectionFactory factory = new ConnectionFactory()
        {
            HostName = "127.0.0.1",
            UserName = "admin",
            Password 
 
= "admin",
            VirtualHost = "/"
        };
        static void Main(string[] args)
        {
            Console.WriteLine("Hello Producer!");
            SendMessage();
            Console.ReadLine();
        }

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

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

            using (var connection = factory.CreateConnection())
            {
                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}");
                }
            }
        }
    }
}

其中在佇列queue_a上設定屬性值x-dead-letter-exchange，x-dead-letter-routing-key

消費者：

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Text;

namespace RabbitMqServer
{
    class Program
    {
        static IConnectionFactory factory = new ConnectionFactory()
        {
            HostName = "127.0.0.1",
            UserName = "admin",
            Password = "admin",
            VirtualHost = "/"
        };
        static void Main(string[] args)
        {
            Console.WriteLine("Hello Consumer!");
            Consumer();
            Console.ReadLine();
        }
        public static void Consumer()
        {
            //死信交換機
            string dlxexChange = "dlx.exchange";
            //死信佇列
            string dlxQueueName = "dlx.queue";

            //訊息交換機
            string exchange = "direct-exchange";
            //訊息佇列
            string queueName = "queue_a";
            var connection = factory.CreateConnection();
            {
                //建立通道
                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的web端看下

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

5.延時佇列

延時佇列是配合死性佇列一起使用，給佇列新增訊息過時時間（TTL）變成延時佇列。

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

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

生產者：

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

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

            using (var connection = factory.CreateConnection())
            {
                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會根據該值去找到死信訊息存放的佇列
                                            { "x-message-ttl",10000}
                                        });
                    //訊息佇列繫結訊息交換機
                    channel.QueueBind(queueName, exchange, routingKey: queueName);

                    string message = "hello rabbitmq message";
                    var properties = channel.CreateBasicProperties();
                    properties.Persistent = true;
                    //釋出訊息
                    for (int i = 0; i < 10; i++)
                    {
                        message += i;
                        channel.BasicPublish(exchange: exchange,
                                         routingKey: queueName,
                                         basicProperties: properties,
                                         body: Encoding.UTF8.GetBytes(message));
                        Console.WriteLine($"{DateTime.Now}向佇列:{queueName}傳送訊息:{message}");
                        System.Threading.Thread.Sleep(1000);
                    }

                }
            }
        }

消費者：

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

            var connection = factory.CreateConnection();
            {
                //建立通道
                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);
                        //不ack(BasicNack),且不把訊息放回佇列(requeue:false)
                        //channel.BasicNack(ea.DeliveryTag, false, requeue: false);
                    };
                    channel.BasicConsume(dlxQueueName, autoAck: false, consumer);
                }
            }
        }

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

6.延時訊息設定不同過期時間

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

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

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

生產者：

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

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

            using (var connection = factory.CreateConnection())
            {
                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 10s後處理";
                    var properties = channel.CreateBasicProperties();
                    properties.Persistent = true;
                    properties.Expiration = "10000";
                    //釋出訊息
                    channel.BasicPublish(exchange: exchange,
                                     routingKey: queueName,
                                     basicProperties: properties,
                                     body: Encoding.UTF8.GetBytes(message));
                    Console.WriteLine($"{DateTime.Now}向佇列:{queueName}傳送訊息:{message}");


                    message = "hello rabbitmq message 5s後處理";
                    properties = channel.CreateBasicProperties();
                    properties.Persistent = true;
                    properties.Expiration = "5000";
                    channel.BasicPublish(exchange: exchange,
                                    routingKey: queueName,
                                    basicProperties: properties,
                                    body: Encoding.UTF8.GetBytes(message));
                    Console.WriteLine($"{DateTime.Now}向佇列:{queueName}傳送訊息:{message}");

                }
            }
        }

消費者程式碼和上面一致。

結論:

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

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

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

新增消費者程式碼：

　　

public static void Consumer_NormalConsumer()
        {
            //死信佇列
            string dlxQueueName = "queue_a";

            var connection = factory.CreateConnection();
            {
                //建立通道
                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());
                        //Console.WriteLine($"{DateTime.Now}佇列{dlxQueueName}消費訊息:{message}");
                        //不ack(BasicNack),且不把訊息放回佇列(requeue:false)
                        System.Threading.Thread.Sleep(20);
                        channel.BasicNack(ea.DeliveryTag, false, requeue: true);
                    };
                    channel.BasicConsume(dlxQueueName, autoAck: false, consumer);
                }
            }
        }

本文來自部落格園，作者：可樂加冰-Mr-Wang，轉載請註明原文連結：https://www.cnblogs.com/helloworld-wang/p/15213854.html

相關推薦

RabbitMQ死性佇列，延時佇列

1.前言 之前一直聽說過RabbitMQ的死性佇列，延時佇列，但是沒有深入的瞭解，結合自己實際工作中使用到的RabbitMQ，現在整理出來當做個人學習筆記。

RabbitMQ從零到叢集高可用.NetCore(.NET5) - 死信佇列，延時佇列（轉）

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

Rabbitmq的死信佇列和延時佇列

一、死信佇列 死信佇列其實和普通的佇列沒啥大的區別，都需要建立自己的Queue、Exchange,然後通過RoutingKey繫結到Exchange上去，只不過死信佇列的RoutingKey和Exchange要作為引數，繫結到正常的佇列上去，一種應用場

面試官：RabbitMQ過期時間設定、死信佇列、延時佇列怎麼設計？

哈嘍！大家好，我是小奇，一位不靠譜的程式設計師 小奇打算以輕鬆幽默的對話方式來分享一些技術，如果你覺得通過小奇的文章學到了東西，那就給小奇一個贊吧

TP6(thinkphp6)佇列與延時佇列

安裝 在此我就不再略過TP6的專案建立過程了，大致就是安裝composer工具，安裝成功以後，再使用composer去建立專案即可。

SpringBoot使用RabbitMQ延時佇列(小白必備)

1.什麼是MQ MQ，是一種跨程序的通訊機制，用於上下游傳遞訊息。 在網際網路架構中，MQ是一種非常常見的上下游“邏輯解耦+物理解耦”的訊息通訊服務。

rabbitmq 延時佇列 外掛方式實現 每條訊息都延時自己時間

上篇文章的延時是加到佇列上的 通過死信過時推送 ，缺點就是不能每條訊息定義自己的過時時間而且每次有新的過時時間，要新建一個交換機和佇列

【日常摘要】- RabbitMq實現延時佇列

簡介 什麼是延時佇列？ 一種帶有延遲功能的訊息佇列 過程： 使用場景 比如存在某個業務場景

springboot-rabbitmq:實現延時佇列

延時佇列應用於什麼場景 延時佇列顧名思義，即放置在該佇列裡面的訊息是不需要立即消費的，而是等待一段時間之後取出消費。那麼，為什麼需要延遲消費呢？我們來看以下的場景 網上商城下訂單後30分鐘後沒有完成支付，

Springboot整合Rabbitmq實現延時佇列之rabbitmq_delayed_message_exchange外掛方式

很多時候我們想定時去做某件事情的時候我們會首先想到定時任務，quartz是個不錯的選擇，但是也有缺點，假如配置在專案中，叢集部署會有重複執行的問題，如果持久化在mysql中，解決了叢集的問題，但是過於依賴mysql，

docker安裝rabbitmq延時佇列外掛

docker安裝rabbitMQ延時佇列外掛（delayed_message_exchange） 1. 查詢Docker容器中的RabbitMQ映象

運用.NetCore例項講解RabbitMQ死信佇列,延時佇列

目錄一、死信佇列二、延時佇列三、延時訊息設定不同過期時間四、延時訊息用延時外掛的方式實現一、死信佇列

RabbitMQ延時佇列應用場景

應用場景 我們系統未付款的訂單，超過一定時間後，需要系統自動取消訂單並釋放佔有物品

基於golang的簡單分散式延時佇列服務的實現

一、引言 背景 我們在做系統時，很多時候是處理實時的任務，請求來了馬上就處理，然後立刻給使用者以反饋。但有時也會遇到非實時的任務，比如確定的時間點發布重要公告。或者需要在使用者做了一件事情的X分鐘/Y小時

一口氣說出Java 6種延時佇列的實現方法(面試官也得服)

五一期間原計劃是寫兩篇文章，看一本技術類書籍，結果這五天由於自律性過於差，禁不住各種誘惑，我連電腦都沒開啟過，計劃完美宣告失敗。所以在這能看出和大佬之間的差距，人家沒白沒夜的更文，比你優秀的人比你更努

延時佇列 DelayQueue

當用戶超時未支付時，給使用者發提醒訊息。另一種場景是，超時未付款，訂單自動取消。通常，訂單建立的時候可以向延遲佇列種插入一條訊息，到時間自動執行。其實，也可以用臨時表，把這些未支付的訂單放到一個臨時表

DelayQueue延時佇列

一、DelayQueue是什麼 　　DelayQueue是一個無界的BlockingQueue，用於放置實現了Delayed介面的物件，其中的物件只能在其到期時才能從佇列中取走。這種佇列是有序的，即隊頭物件的延遲到期時間最長。注意：不能將nul

詳解Java中的延時佇列 DelayQueue

當用戶超時未支付時，給使用者發提醒訊息。另一種場景是，超時未付款，訂單自動取消。通常，訂單建立的時候可以向延遲佇列種插入一條訊息，到時間自動執行。其實，也可以用臨時表，把這些未支付的訂單放到一個臨時表

延時佇列 DelayQueue 的簡單使用

技術標籤：佇列java 使用場景：物聯網應用中，硬體版本升級。需要通過線上操作，對某一型別裝置，在某個時間點進行升級。

延時佇列的6個實現方案

一、延時佇列的應用 什麼是延時佇列？顧名思義：首先它要具有佇列的特性，再給它附加一個延遲消費佇列訊息的功能，也就是說可以指定佇列中的訊息在哪個時間點被消費。延時佇列在專案中的應用還是比較多的，尤其像電商