Net分佈式系統之四：RabbitMQ消息隊列應用

時間 2019-11-19

原文原文鏈接

消息通訊組件Net分佈式系統的核心中間件之一，應用與系統高併發，各個組件之間解耦的依賴的場景。本框架採用消息隊列中間件主要應用於兩方面：一是解決部分高併發的業務處理；二是經過消息隊列傳輸系統日誌。目前業界使用較多的消息隊列組件有RabbitMQ、ActiveMQ、MSMQ、kafka、zeroMQ等，本文對系統架構之MQ Component詮釋，並採用RabbitMQ做爲消息隊列中間件。html

圖1- 消息隊列組件示意圖web

1、RabbitMQ介紹算法

　　RabbitMQ是一款基於AMQP（消息隊列協議），由Erlang開發的開源消息隊列組件。是一款優秀的消息隊列組件，他由兩部分組成：服務端和客戶端，客戶端支持多種語言的驅動，如：.Net、JAVA、Erlang等。RabbitMQ與其餘消息隊列組件性能比較，在此不做介紹，網上有大把的資料。緩存

2、RabbitMQ原理簡介session

圖2- RabbitMQ結構示意圖架構

　　RabbitMQ中間件分爲服務端（RabbitMQ Server）和客戶端（RabbitMQ Client），服務端能夠理解爲是一個消息的代理消費者，客戶端又分爲消息生產者（Producer）和消息消費者（Consumer）。併發

　　一、消息生產者（Producer）：主要生產消息並將消息基於TCP協議，經過創建Connection和Channel，將消息傳輸給RabbitMQ Server，對於Producer而言基本就完成了工做。框架

　　二、服務端（RabbitMQ Server）：主要負責處理消息路由、分發、入隊列、緩存和出列。主要由三部分組成：Exchange、RoutingKey、Queue。運維

　　（1）Exchange：用於接收消息生產者發送的消息，有三種類型的exchange：direct, fanout,topic，不一樣類型實現了不一樣的路由算法；tcp

　　　　A. direct exchange：將與routing key 比配的消息，直接推入相對應的隊列，建立隊列時，默認就建立同名的routing key。

　　　　B. fanout exchange：是一種廣播模式，忽略routingkey的規則。

　　　　C. topic exchange：應用主題，根據key進行模式匹配路由，例如：若爲abc*則推入到全部abc*相對應的queue；若爲abc.#則推入到abc.xx.one ,abc.yy.two對應的queue。

　　（2）RoutingKey：是RabbitMQ實現路由分發到各個隊列的規則，並結合Binging提供於Exchange使用將消息推送入隊列；

　　（3）Queue：是消息隊列，能夠根據須要定義多個隊列，設置隊列的屬性，好比：消息移除、消息緩存、回調機制等設置，實現與Consumer通訊；

　　三、消息消費者（Consumer）：主要負責消費Queue的消息，一樣基於TCP協議，經過創建Connection和Channel與Queue傳輸消息，一個消息能夠給多個Consumer消費；

　　四、關鍵名詞說明：Connection、Channel、Binging等；

　　（1）Connection：是創建客戶端與服務端的鏈接。

　　（2）Channel：是基於Connection之上創建通訊通道，由於每次Connection創建TCP協議通訊開銷及性能消耗較大，因此一次創建Connection後，使用多個Channel通道通訊減小開銷和提升性能。

　　（3）Binging：是一個捆綁定義，將exchange和queue捆綁，定義routingkey相關策略。

3、RabbitMQ安裝部署

　　以上對RabbitMQ簡介，接下來咱們經過實際搭建消息隊列服務實踐。RabbitMQ服務端能運行於Window、Linux和Mac平臺，客戶端也支持多種技術的實現。本次咱們將在Linux之CentOS7平臺搭建。

　　一、安裝Erlang運行環境

　　　　因爲RabbitMQ使用Erlang技術開發，因此須要先安裝Erlang運行環境後，才能安裝消息隊列服務。

　　　　（1）配置系統能正常訪問公網，設置默認網關

 
         route add  
         default  
         gw 192.168.1.1

　　　　（2）安裝erlang

 
         su -c  
         'rpm -Uvh http://download.fedoraproject.org/pub/epel/7/x86_64/e/epel-release-7-5.noarch.rpm' 
        
         sudo yum install erlang

　　　　（3）檢查erlang是否安裝成功

erl

　　　　（4）安裝成功

　　二、安裝RabbitMQ服務端

　　　　（1）下載安裝包

 
         wget http: 
         //www.rabbitmq.com/releases/rabbitmq-server/v3.6.0/rabbitmq-server-3.6.0-1.noarch.rpm

　　　　（2）安裝和配置RabbitMQ服務端，3.6.0版本：

 
         rpm --import https: 
         //www.rabbitmq.com/rabbitmq-signing-key-public.asc 
        
         yum install rabbitmq-server-3.6.0-1.noarch.rpm

　　　　（3）啓用web管理插件

 
         rabbitmq-plugins enable rabbitmq_management

　　　　（4）啓動RabbitMQ

 
         chkconfig rabbitmq-server  
         on 
        
         /sbin/service rabbitmq-server start

　　　　（5）防火牆開通端口

 
         # firewall-cmd --permanent --zone=public --add-port=5672/tcp 
        
         # firewall-cmd --permanent --zone=public --add-port=15672/tcp 
        
         # firewall-cmd --reload

　　　　（6）rabbitmq默認會建立guest帳號，只能用於localhost登陸頁面管理員，本機訪問地址：http://localhost:15672/

 
         rabbitmqctl add_user test test 
        
         rabbitmqctl set_user_tags test administrator<br>rabbitmqctl set_permissions -p / test  
         ".*"  
         ".*"  
         ".*"

　　　　　 RabbitMQ 管理員頁面。

4、RabbitMQ應用

　　本章節描述，web應用生產的日誌，經過rabbitmq傳輸，而後日誌服務接收消息隊列的消息。

圖3- 功能結構示意圖

　　本系統採用官方的Client，經過nuget引用。

　　一、Web應用生產業務日誌

 
         [HttpPost] 
        
         public  
         ActionResult Create() 
        
         { 
        
         this 
         .HttpContext.Session[ 
         "mysession" 
         ] = DateTime.Now.ToString( 
         "yyyy-MM-dd hh:mm:ss" 
         ); 
        
         var  
         txt = Request.Form[ 
         "txtSite" 
         ].ToString(); 
        
         RabbitMQHelper helper =  
         new  
         RabbitMQHelper(); 
        
         helper.SendMsg(txt +  
         ",操做日誌，時間:"  
         + DateTime.Now.ToString( 
         "yyyy-MM-dd hh:mm:ss" 
         )); 
        
         return  
         RedirectToAction( 
         "Index" 
         ); 
        
         }

　　頁面效果圖。

　　二、日誌服務接收日誌消息

　　　　基於window form開發一個日誌處理服務，並將接收的消息打印出來。

 
         private  
         void  
         btnReceive_Click( 
         object  
         sender, EventArgs e) 
        
         { 
        
         isConnected =  
         true 
         ; 
        
         using  
         ( 
         var  
         channel = connection.CreateModel()) 
        
         { 
        
         channel.QueueDeclare( 
         "MyLog" 
         ,  
         false 
         ,  
         false 
         ,  
         false 
         ,  
         null 
         ); 
        
         var  
         consumer =  
         new  
         QueueingBasicConsumer(channel); 
        
         channel.BasicConsume( 
         "MyLog" 
         ,  
         true 
         , consumer); 
        
         while  
         (isConnected) 
        
         { 
        
         var  
         ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue(); 
        
         var  
         body = ea.Body; 
        
         var  
         message = Encoding.UTF8.GetString(body); 
        
         txtMsg.Text += message +  
         "\r\n" 
         ; 
        
         } 
        
         } 
        
         }