.net使用RabbitMQ

時間 2019-11-26

原文原文鏈接

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

1、生產者數據庫

1.建立鏈接connection:無論是生產者仍是消費者都須要先於RabbitMQ服務器鏈接，才能進行數據交換瀏覽器

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

3.聲明交換器、隊列網絡

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

5.經過交換器BasicPublish數據到隊列學習

2、消費者fetch

1.建立鏈接 :和生產者同樣spa

2.建立通道:和生產者同樣調試

3.聲明交換器、隊列

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

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

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

3、出現的錯誤

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

這個顯然是你的服務端網絡不通或者服務沒有啓用、端口配置不正確致使的，建議你先檢查先端口、服務是否啓動成功。能夠用telnet或者netstat -a等命令，排除本地服務是否有誤等問題。排除服務端，再調試客戶端,以前本身也有檢查了，RabbitMQ也是啓動的，用瀏覽器登陸http://localhost:15672也能登陸，感受是至關的納悶，本身重啓了下RabbitMQ以後又試了一下沒想到成功了，感受到好納悶，如今還不知道是什麼緣由致使的。好神奇的樣子。

4、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();
                    }
                }
            }
        }
    }
}

View Code

上面的代碼分別建立了兩個路由和兩個隊列，一種是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();
                }
            }
        }

    }
}

View Code

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

4.實驗截圖

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