.net使用RabbitMQ

前面的兩篇博文算是把RabbitMQ的基礎了解了下，今天學習.Net 中RabbitMQ的使用。原本這篇博文是應該上周寫的，可在自己使用的過程中出現了一個問題bug:就是在連接服務端時，一直報下面的錯誤:None of the specified endpoints were reachable.自己也上網查了好久，又問了網友，這里要感謝博客園的園有的幫助。一般使用這些技術的使用都是有一定的步驟，有固有的套路，熟悉了套路，用起來就會很方便。像ADO.Net的五大對象也是，操作數據庫先進行連接connection,然后使用command過濾出要選擇的數據，使用DataReader或DataSet、DataAdapter,在RabbitMQ的使用當中也有基本固定的步驟。

一、生產者

1.創建連接connection:不管是生產者還是消費者都需要先於RabbitMQ服務器連接，才能進行數據交換

2.創建通道 Channel:生產者、消費者的消息傳遞是在通道下傳遞的

3.聲明交換器、隊列

4.交換器與隊列進行綁定

5.通過交換器BasicPublish數據到隊列

二、消費者

1.創建連接 :和生產者一樣

2.創建通道:和生產者一樣

3.聲明交換器、隊列

4.交換器與隊列進行綁定

5.通過BasicGet方法獲取隊列中的數據

上面一、二是大致的基本步驟，按照大致的步驟來基本不會出現大的問題，其實生產者和消費者的前4個步驟基本一樣，主要是第5個步驟，一個是生產BasicPublish發布，一個是獲取Get。

三、出現的錯誤

在ConnectionFactory創建連接對象時出現上面提到的bug:None of the specified endpoints were reachable.這個問了下博客園博友的，博友的回復如下:

這個顯然是你的服務端網絡不通或者服務沒有啟用、端口配置不正確導致的，建議你先檢查先端口、服務是否啟動成功。可以用telnet或者netstat -a等命令，排除本地服務是否有誤等問題。排除服務端，再調試客戶端,之前自己也有檢查了，RabbitMQ也是啟動的，用瀏覽器登錄http://localhost:15672也能登錄，感覺是相當的納悶，自己重啟了下RabbitMQ之后又試了一下沒想到成功了，感覺到好納悶，現在還不知道是什么原因導致的。好神奇的樣子。

四、demo

1.首先在生產者端和消費者端引入RabbitMQ

 2.生產者端

using RabbitMQ.Client;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace RMQProducter
{
    class Program
    {
        /// <summary>
        /// 連接配置
        /// </summary>
        private static readonly ConnectionFactory rabbitMqFactory = new ConnectionFactory(){
            UserName = "cywadmin",
            Password = "123456",
            Port = 5672,
            VirtualHost= "cywVirtualHost"
        };
        /// <summary>
        /// 路由名稱
        /// </summary>
        const string ExchangeName = "cyw.exchange";

        //隊列名稱
        const string QueueName = "cyw.queue";

        /// <summary>
        /// 路由名稱
        /// </summary>
        const string TopExchangeName = "topic.cyw.exchange";

        //隊列名稱
        const string TopQueueName = "topic.cyw.queue";


        static void Main(string[] args)
        {
            DirectExchangeSendMsg();
           // TopicExchangeSendMsg();
            Console.WriteLine("按任意值，退出程序");
            Console.ReadKey();
        }
        /// <summary>
        ///  單點精確路由模式
        /// </summary>
        public static void DirectExchangeSendMsg()
        {
            using (IConnection conn = rabbitMqFactory.CreateConnection())
            {
                using (IModel channel = conn.CreateModel())
                {
                    channel.ExchangeDeclare(ExchangeName, "direct", durable: true, autoDelete: false, arguments: null);
                    channel.QueueDeclare(QueueName, durable: true, autoDelete: false, exclusive: false, arguments: null);
                    channel.QueueBind(QueueName, ExchangeName, routingKey: QueueName);

                    var props = channel.CreateBasicProperties();
                    props.Persistent = true;
                    string vadata = Console.ReadLine();
                    while (vadata != "exit")
                    {
                        var msgBody = Encoding.UTF8.GetBytes(vadata);
                        channel.BasicPublish(exchange: ExchangeName, routingKey: QueueName, basicProperties: props, body: msgBody);                        
                        Console.WriteLine(string.Format("***發送時間:{0}，發送完成，輸入exit退出消息發送",
                            DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")));
                        vadata = Console.ReadLine();
                    }                   
                }
            }
        }

        public static void TopicExchangeSendMsg()
        {
            using (IConnection conn = rabbitMqFactory.CreateConnection())
            {
                using (IModel channel = conn.CreateModel())
                {
                    channel.ExchangeDeclare(TopExchangeName, "topic", durable: false, autoDelete: false, arguments: null);
                    channel.QueueDeclare(TopQueueName, durable: false, autoDelete: false, exclusive: false, arguments: null);
                    channel.QueueBind(TopQueueName, TopExchangeName, routingKey: TopQueueName);
                    //var props = channel.CreateBasicProperties();
                    //props.Persistent = true;
                    string vadata = Console.ReadLine();
                    while (vadata != "exit")
                    {
                        var msgBody = Encoding.UTF8.GetBytes(vadata);
                        channel.BasicPublish(exchange: TopExchangeName, routingKey: TopQueueName, basicProperties: null, body: msgBody);
                        Console.WriteLine(string.Format("***發送時間:{0}，發送完成，輸入exit退出消息發送", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")));
                        vadata = Console.ReadLine();
                    }
                }
            }
        }
    }
}

上面的代碼分別創建了兩個路由和兩個隊列，一種是DirectExchange，一種是TopicExchange，驗證時需要生產者和消費者使用同一種的ExChange。

3.消費者端

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace RMQCustomer
{
    class Program
    {
        /// <summary>
        /// 連接配置
        /// </summary>
        private static readonly ConnectionFactory rabbitMqFactory = new ConnectionFactory() {
            HostName = "127.0.0.1",
            UserName = "cywadmin",
            Password = "123456",
            Port = 15672,
            VirtualHost = "cywVirtualHost"
        };
        /// <summary>
        /// 路由名稱
        /// </summary>
        const string ExchangeName = "cyw.exchange";

        //隊列名稱
        const string QueueName = "cyw.queue";

        /// <summary>
        /// 路由名稱
        /// </summary>
        const string TopExchangeName = "topic.cyw.exchange";

        //隊列名稱
        const string TopQueueName = "topic.cyw.queue";


        static void Main(string[] args)
        {
            DirectAcceptExchange();
            //DirectAcceptExchangeEvent();
            //DirectAcceptExchangeTask();
            //TopicAcceptExchange();
            Console.WriteLine("按任意值，退出程序");
            Console.ReadKey();
        }

        public static void DirectAcceptExchange()
        {
            using (IConnection conn = rabbitMqFactory.CreateConnection())
            {
                using (IModel channel = conn.CreateModel())
                {
                    channel.ExchangeDeclare(ExchangeName, "direct", durable: true, autoDelete: false, arguments: null);
                    channel.QueueDeclare(QueueName, durable: true, autoDelete: false, exclusive: false, arguments: null);
                    channel.QueueBind(QueueName, ExchangeName, routingKey: QueueName);
                    while (true)
                    { 
                        BasicGetResult msgResponse = channel.BasicGet(QueueName, noAck: true);
                        if (msgResponse != null)
                        {
                            var msgBody = Encoding.UTF8.GetString(msgResponse.Body);
                            Console.WriteLine(string.Format("***接收時間:{0}，消息內容：{1}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"),msgBody));
                        }

                        //BasicGetResult msgResponse2 = channel.BasicGet(QueueName, noAck: false);

                        ////process message ...

                        //channel.BasicAck(msgResponse2.DeliveryTag, multiple: false);
                        System.Threading.Thread.Sleep(TimeSpan.FromSeconds(1));
                    }
                }
            }
        }

        public static void DirectAcceptExchangeEvent()
        {
            using (IConnection conn = rabbitMqFactory.CreateConnection())
            {
                using (IModel channel = conn.CreateModel())
                {
                    //channel.ExchangeDeclare(ExchangeName, "direct", durable: true, autoDelete: false, arguments: null);
                    channel.QueueDeclare(QueueName, durable: true, autoDelete: false, exclusive: false, arguments: null);
                    //channel.QueueBind(QueueName, ExchangeName, routingKey: QueueName);
                    var consumer = new EventingBasicConsumer(channel);
                    consumer.Received += (model, ea) =>
                    {                        
                        var msgBody = Encoding.UTF8.GetString(ea.Body);
                        Console.WriteLine(string.Format("***接收時間:{0}，消息內容：{1}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"), msgBody));
                    };
                    channel.BasicConsume(QueueName, noAck: true, consumer: consumer);

                    //已過時用EventingBasicConsumer代替
                    //var consumer2 = new QueueingBasicConsumer(channel);
                    //channel.BasicConsume(QueueName, noAck: true, consumer: consumer);
                    //var msgResponse = consumer2.Queue.Dequeue(); //blocking
                    //var msgBody2 = Encoding.UTF8.GetString(msgResponse.Body);

                    Console.WriteLine("按任意值，退出程序");
                    Console.ReadKey();
                }
            }
        }

        public static void DirectAcceptExchangeTask()
        {
            using (IConnection conn = rabbitMqFactory.CreateConnection())
            {
                using (IModel channel = conn.CreateModel())
                {
                    //channel.ExchangeDeclare(ExchangeName, "direct", durable: true, autoDelete: false, arguments: null);
                    channel.QueueDeclare(QueueName, durable: true, autoDelete: false, exclusive: false, arguments: null);
                    channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);//告訴broker同一時間只處理一個消息
                    //channel.QueueBind(QueueName, ExchangeName, routingKey: QueueName);
                    var consumer = new EventingBasicConsumer(channel);
                    consumer.Received += (model, ea) =>
                    {
                        var msgBody = Encoding.UTF8.GetString(ea.Body);
                        Console.WriteLine(string.Format("***接收時間:{0}，消息內容：{1}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"), msgBody));                        
                        int dots = msgBody.Split('.').Length - 1;
                        System.Threading.Thread.Sleep(dots * 1000);
                        Console.WriteLine(" [x] Done");
                        //處理完成，告訴Broker可以服務端可以刪除消息，分配新的消息過來
                        channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
                    };
                    //noAck設置false,告訴broker，發送消息之后，消息暫時不要刪除，等消費者處理完成再說
                    channel.BasicConsume(QueueName, noAck: false, consumer: consumer);

                    Console.WriteLine("按任意值，退出程序");
                    Console.ReadKey();
                }
            }
        }

        public static void TopicAcceptExchange()
        {
            using (IConnection conn = rabbitMqFactory.CreateConnection())
            {
                using (IModel channel = conn.CreateModel())
                {
                    channel.ExchangeDeclare(TopExchangeName, "topic", durable: false, autoDelete: false, arguments: null);
                    channel.QueueDeclare(TopQueueName, durable: false, autoDelete: false, exclusive: false, arguments: null);
                    channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
                    channel.QueueBind(TopQueueName, TopExchangeName, routingKey: TopQueueName);
                    var consumer = new EventingBasicConsumer(channel);
                    consumer.Received += (model, ea) =>
                    {
                        var msgBody = Encoding.UTF8.GetString(ea.Body);
                        Console.WriteLine(string.Format("***接收時間:{0}，消息內容：{1}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"), msgBody));
                        int dots = msgBody.Split('.').Length - 1;
                        System.Threading.Thread.Sleep(dots * 1000);
                        Console.WriteLine(" [x] Done");
                        channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
                    };
                    channel.BasicConsume(TopQueueName, noAck: false, consumer: consumer);

                    Console.WriteLine("按任意值，退出程序");
                    Console.ReadKey();
                }
            }
        }

    }
}

消費者端也是兩個路由兩個隊列，在實現DirectExchange時使用了三種方式，DirectAcceptExchange是基於時間輪詢的，每隔一段時間獲取一次，DirectAcceptExchangeEvent、DirectAcceptExchangeTask是基於事件的，當消息到達時觸發事件，獲取數據。

4.實驗截圖

 參考:http://www.cnblogs.com/xibei666/p/5931267.html#3680686 感謝園主，我這篇博文用的都是他的例子。