RabbitMq應用一_ZenDei技術網路在線

RabbitMq應用一 RabbitMQ的具體概念，百度百科一下，我這裡說一下我的理解，如果有少或者不對的地方，歡迎糾正和補充。一個項目架構，小的時候，一般都是傳統的單一網站系統，或者項目，三層架構，到現在的MVC架構。隨著用戶訪問量越來越多，系統業務越來越多，會出現以下問題： 1.修改完大量代碼 ...

RabbitMq應用一

RabbitMQ的具體概念，百度百科一下，我這裡說一下我的理解，如果有少或者不對的地方，歡迎糾正和補充。

一個項目架構，小的時候，一般都是傳統的單一網站系統，或者項目，三層架構，到現在的MVC架構。隨著用戶訪問量越來越多，系統業務越來越多，會出現以下問題：

　　　1.修改完大量代碼後，不敢更新，因為都是集成在一起，互相耦合性非常強，一處報錯，滿盤皆掛；

　　2.整個項目文件夾，層級越來越多，對新來的同事很不友好，文件不可避免的會亂放，重覆的過多，甚至為了緊急更新，會把很多原本的需要編譯的代碼，挪到一般處理程式中，

　　　　時間越長，越會發現，整個代碼結構像一鍋粥一樣；

　　3.會有很多地方需要記錄日誌，郵件，簡訊等等很多需要非同步的操作，如果訪問量過高，會把這個系統拖垮。

上述問題出現一定時間後，一定會重構整個，進行業務分離，SOA架構服務化，這就涉及到多個應用相互之間的通信，常見的方式，是通過API的方式通過JSON的方式，進行數據交互，

這種做法實時性很高，但是對單個業務系統的高峰期壓力還是非常大的，需要對但業務API系統進行負載均衡，這時候，如果說把一些要求實時性相對低一些，並且特別消耗性能的請求，摘出去慢慢處理的話，消息隊列就派上用場了，引入的消息隊列就成了消息處理的緩衝區。消息隊列引入的非同步通信機制，使得發送方和接收方都不用等待對方返回成功消息，就可以繼續執行下麵的代碼，從而提高了數據處理的能力。尤其是當訪問量和數據流量較大的情況下，就可以結合消息隊列與後臺任務，通過避開高峰期對大數據進行處理，就可以有效降低資料庫和程式處理數據的負荷。

一搭建環境

RabbitMq是由erlang語言開發，所有到先安裝erlang語言的環境，傳送門下載，並且安裝，預設安裝後會自動配置一個環境變數ERLANG_HOME ，如果沒配置的話，手動配置一下，指向erlang安裝目錄就可以了。

安裝好erlang語言環境，我們去下載rabbitmq服務了。地址http://www.rabbitmq.com/download.html,下載RabbitMq Server.

下載完，安裝完畢後，打開安裝目錄到sbin下　

我們右鍵CMD，用管理員身份打開，然後切到上圖這個目錄下，執行3行命令

rabbitmq-service install
rabbitmq-service enable
rabbitmq-service start
開啟rabbitMq的服務，這時候可以自己查看服務是否開啟。
安裝好以後，我們使用rabbitmqctl list_users命令，是列出當前服務用戶的列表，

這個用戶是我自己添加的，剛開始的時候是預設有一個用戶guest
我們可以自己添加用戶，並且設置密碼，設置許可權，設置管理員操作，還可以刪除用戶，更改密碼

rabbitmqctl  add_user  feige  habi  添加用戶，賬號feige,密碼habi
rabbitmqctl  set_permissions  feige  ".*"  ".*"  ".*"給feige這個用戶設置對所有消息隊列設置和配置，讀，寫的許可權
rabbitmqctl  set_user_tags feige administrator給feige這個用戶設置成管理員

rabbitmqctl delete_user feige  這個是刪除用戶

rabbitmqctl change_password feige  1234修改feige的密碼

運行下麵命令來啟用管理插件：rabbitmq-plugins enable rabbitmq_management
預設埠:http://localhost:15672/#/這個是可以監控的後臺。



綜上，RabbitMq的環境已經部署完畢，一些基本命令已經熟悉了，下麵開始使用

二基本使用

在.NET中使用RabbitMQ需要下載RabbitMQ的客戶端程式集，可以到官網下載，下載解壓後就可以得到RabbitMQ.Client.dll，這就是RabbitMQ的客戶端。

在使用RabitMQ之前，需要對下麵的幾個基本概念說明一下：

RabbitMQ是一個消息代理。他從消息生產者(producers)那裡接收消息，然後把消息送給消息消費者（consumer）在發送和接受之間，他能夠根據設置的規則進行路由，緩存和持久化。

我們用VS2012+，可以直接從nuget控制台命令，添加

Install-Package RabbitMq.Client -version 3.6.5

生產(Producing)意思就是發送。發送消息的程式就是一個生產者(producer)。

隊列(queue)就是郵箱的名稱。消息通過你的應用程式和RabbitMQ進行傳輸，它們只能存儲在隊列（queue）中。隊列（queue）容量沒有限制，你要存儲多少消息都可以——基本上是一個無限的緩衝區。多個生產者（producers）能夠把消息發送給同一個隊列，同樣，多個消費者（consumers）也能從同一個隊列（queue）中獲取數據。隊列可以畫成這樣（圖上是隊列的名稱）：

消費（Consuming）和獲取消息是一樣的意思。一個消費者（consumer）就是一個等待獲取消息的程式

通常，消息生產者，消息消費者和消息代理不在同一臺機器上。

下麵，我們用.net代碼，來分別做一個製造者，發送消息的控制台，一個消費者，接收消息的控制台，當然，我們在本機做一下實驗，。

/// <summary>
    /// 消息生產者，客戶端消息存進隊列中
    /// </summary>
    class Program
    {
        static void Main(string[] args)
        {
            //創建鏈接工廠，設置目標，用戶，密碼
            var factory = new ConnectionFactory();
            factory.HostName = "127.0.0.1";
            factory.UserName = "feiyang";
            factory.Password = "123456";
            //開啟當前服務設置的用戶的鏈接
            using (var connection =  factory.CreateConnection())
            {
                //開啟一個頻道
                using (var channel = connection.CreateModel())
                {
                    //創建一個隊列
                    channel.QueueDeclare("firstQueue",false,false,false,null);
                    byte[] body = null;
                    //消息是以二進位數組的形式傳輸的，所以如果消息是實體對象的話，需要序列化和然後轉化為二進位數組。
                    for (int i = 0; i < 100000; i++)
                    {
                        body = Encoding.UTF8.GetBytes("這是第-----"+i+"-----條消息");
                        channel.BasicPublish("", "firstQueue", null, body);
                        Console.Write("成功發送第-----"+i+"-----條消息！");
                    }
                    Console.ReadKey();
                }
            }
        }
    }

上述是消費者控制台代碼，我下麵那個迴圈時測試的10W條數據，不斷的發送和另一個控制台不斷的獲取消息，測試下來還不錯性能。。。這個例子發送的消息比較簡單，直接轉成二進位就可以了，但是如果我們

用到對象的話，就要先反序列化，再轉成二進位。

 /// <summary>
    /// 消息消費者讀取消息隊列
    /// </summary>
    class Program
    {
        static void Main(string[] args)
        {
            var factory = new ConnectionFactory();
            factory.HostName = "127.0.0.1";
            factory.UserName = "feiyang";
            factory.Password = "123456";
            using (var connection  = factory.CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    //還是連接到哪個隊列
                    channel.QueueDeclare("firstQueue",false,false,false,null);
                    //定義消息接受者
                    var customer = new QueueingBasicConsumer(channel);
                    //從指定隊列獲取消息
                    channel.BasicConsume("firstQueue",true,customer);
                    //開始不斷迴圈出隊列的消息
                    while (true)
                    {
                        var ea = (BasicDeliverEventArgs)customer.Queue.Dequeue();
                        //將消息二進位轉回字元串
                        var msg = Encoding.UTF8.GetString(ea.Body);
                        Console.WriteLine(msg);
                    }
                    //sw.Stop();
                    //Console.WriteLine("共用時" + sw.ElapsedTicks + "毫秒");
                    //Console.ReadKey();
                }
            }
        }
    }