C#實現(xiàn)rabbitmq 延遲隊列功能實例代碼

更新時間：2017年04月22日 15:22:04 作者：ThisIsTest

本篇文章主要介紹了C#實現(xiàn)rabbitmq 延遲隊列功能實例代碼，具有一定的參考價值，感興趣的小伙伴們可以參考一下。

最近在研究rabbitmq,項目中有這樣一個場景：在用戶要支付訂單的時候，如果超過30分鐘未支付，會把訂單關掉。當然我們可以做一個定時任務，每個一段時間來掃描未支付的訂單，如果該訂單超過支付時間就關閉，但是在數(shù)據(jù)量小的時候并沒有什么大的問題，但是數(shù)據(jù)量一大輪訓數(shù)據(jù)庫的方式就會變得特別耗資源。當面對千萬級、上億級數(shù)據(jù)量時，本身寫入的IO就比較高，導致長時間查詢或者根本就查不出來，更別說分庫分表以后了。除此之外，還有優(yōu)先級隊列，基于優(yōu)先級隊列的JDK延遲隊列，時間輪等方式。但如果系統(tǒng)的架構中本身就有RabbitMQ的話，那么選擇RabbitMQ來實現(xiàn)類似的功能也是一種選擇。我們項目中用到了rabbitmq，可以做一個延遲隊列完美的解決這個問題。

rabbitmq本身不具有延時消息隊列的功能，但是可以通過TTL(Time To Live)、DLX(Dead Letter Exchanges)特性實現(xiàn)。其原理給消息設置過期時間，在消息隊列上為過期消息指定轉發(fā)器，這樣消息過期后會轉發(fā)到與指定轉發(fā)器匹配的隊列上，變向實現(xiàn)延時隊列。利用rabbitmq的這種特性，應該有了一個大概的思路。、

網(wǎng)上搜了一下 rabbitmq-delayed-message-exchange 這個插件也可以實現(xiàn)延遲隊列的功能。今天介紹的是如何用C#來實現(xiàn)。

首先了解一下TTL和DLX

消息的TTL（Time To Live）

消息的TTL就是消息的存活時間。RabbitMQ可以對隊列和消息分別設置TTL。對隊列設置就是隊列沒有消費者連著的保留時間，也可以對每一個單獨的消息做單獨的設置。超過了這個時間，我們認為這個消息就死了，稱之為死信。如果隊列設置了，消息也設置了，那么會取小的。所以一個消息如果被路由到不同的隊列中，這個消息死亡的時間有可能不一樣（不同的隊列設置）。這里單講單個消息的TTL，因為它才是實現(xiàn)延遲任務的關鍵。

Dead Letter Exchanges

Exchage的概念在這里就不在贅述。一個消息在滿足如下條件下，會進死信路由，記住這里是路由而不是隊列，一個路由可以對應很多隊列。

1. 一個消息被Consumer拒收了，并且reject方法的參數(shù)里requeue是false。也就是說不會被再次放在隊列里，被其他消費者使用。

2. 上面的消息的TTL到了，消息過期了。

3. 隊列的長度限制滿了。排在前面的消息會被丟棄或者扔到死信路由上。

Dead Letter Exchange其實就是一種普通的exchange，和創(chuàng)建其他exchange沒有兩樣。只是在某一個設置Dead Letter Exchange的隊列中有消息過期了，會自動觸發(fā)消息的轉發(fā)，發(fā)送到Dead Letter Exchange中去。

首先我建了兩個控制臺項目一個是生產者，一個是消費者。

生產者代碼如下

      var factory = new ConnectionFactory() { HostName = "127.0.0.1", UserName = "test", Password = "test" };
      using (var connection = factory.CreateConnection())
      {
        while (Console.ReadLine() != null)
        {
          using (var channel = connection.CreateModel())
          {

            Dictionary<string, object> dic = new Dictionary<string, object>();
            dic.Add("x-expires", 30000);
            dic.Add("x-message-ttl", 12000);//隊列上消息過期時間，應小于隊列過期時間 
            dic.Add("x-dead-letter-exchange", "exchange-direct");//過期消息轉向路由 
            dic.Add("x-dead-letter-routing-key", "routing-delay");//過期消息轉向路由相匹配routingkey 
            //創(chuàng)建一個名叫"zzhello"的消息隊列
            channel.QueueDeclare(queue: "zzhello",
              durable: true,
              exclusive: false,
              autoDelete: false,
              arguments: dic);

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

            //向該消息隊列發(fā)送消息message
            channel.BasicPublish(exchange: "",
              routingKey: "zzhello",
              basicProperties: null,
              body: body);
            Console.WriteLine(" [x] Sent {0}", message);
          }
        }
      }

      Console.ReadKey();

消費者代碼如下：

 var factory = new ConnectionFactory() { HostName = "127.0.01", UserName = "test", Password = "test" };

      using (var connection = factory.CreateConnection())
      {
        using (var channel = connection.CreateModel())
        {
          channel.ExchangeDeclare(exchange: "exchange-direct", type: "direct");
          string name = channel.QueueDeclare().QueueName;
          channel.QueueBind(queue: name, exchange: "exchange-direct", routingKey: "routing-delay");

          //回調，當consumer收到消息后會執(zhí)行該函數(shù)
          var consumer = new EventingBasicConsumer(channel);
          consumer.Received += (model, ea) =>
          {
            var body = ea.Body;
            var message = Encoding.UTF8.GetString(body);
            Console.WriteLine(ea.RoutingKey);
            Console.WriteLine(" [x] Received {0}", message);
          };

          //Console.WriteLine("name:" + name);
          //消費隊列"hello"中的消息
          channel.BasicConsume(queue: name,
                     autoAck: true,
                     consumer: consumer);

          Console.WriteLine(" Press [enter] to exit.");
          Console.ReadLine();
        }
      }

      Console.ReadKey();

效果：