1 RabbitMQ簡介
RabbitMQ是一個由erlang開發的AMQP(Advanced Message Queue )的開源實現,官網地址:http://www.rabbitmq.com。RabbitMQ作為一個消息代理,主要負責接收、存儲和轉發消息,它提供了可靠的消息機制和靈活的消息路由,並支持消息集群和分布式部署,常用於應用解耦,耗時任務隊列,流量削鋒等場景。本系列文章將系統介紹RabbitMQ的工作機制,代碼驅動和集群配置,本篇主要介紹RabbitMQ中一些基本概念,常用的RabbitMQ Control命令,最后寫一個C#驅動的簡單栗子。先看一下RabbitMQ的基本結構:
上圖是RabbitMQ的一個基本結構,生產者Producer和消費者Consumer都是RabbitMQ的客戶端,Producer負責發送消息,Consumer負責消費消息。
接下來我們結合這張圖來理解RabbitMQ的一些概念:
Broker(Server):接受客戶端連接,實現AMQP消息隊列和路由功能的進程,我們可以把Broker叫做RabbitMQ服務器。
Virtual Host:一個虛擬概念,一個Virtual Host里面可以有若干個Exchange和Queue,主要用於權限控制,隔離應用。如應用程序A使用VhostA,應用程序B使用VhostB,那么我們在VhostA中只存放應用程序A的exchange,queue和消息,應用程序A的用戶只能訪問VhostA,不能訪問VhostB中的數據。
Exchange:接受生產者發送的消息,並根據Binding規則將消息路由給服務器中的隊列。ExchangeType決定了Exchange路由消息的行為,例如,在RabbitMQ中,ExchangeType有Direct、Fanout、Topic和Header四種,不同類型的Exchange路由規則是不一樣的(這些以后會詳細介紹)。
Queue:消息隊列,用於存儲還未被消費者消費的消息,隊列是先進先出的,默認情況下先存儲的消息先被處理。
Message:就是消息,由Header和Body組成,Header是由生產者添加的各種屬性的集合,包括Message是否被持久化、由哪個Message Queue接受、優先級是多少等,Body是真正傳輸的數據,內容格式為byte[]。
Connection:連接,對於RabbitMQ而言,其實就是一個位於客戶端和Broker之間的TCP連接。
Channel:信道,僅僅創建了客戶端到Broker之間的連接Connection后,客戶端還是不能發送消息的。需要在Connection的基礎上創建Channel,AMQP協議規定只有通過Channel才能執行AMQP的命令,一個Connection可以包含多個Channel。之所以需要Channel,是因為TCP連接的建立和釋放都是十分昂貴的。
2 RabbitMQ安裝
因為RabbitMQ是用erlang語言開發的,所以我們在安裝RabbitMQ前必須要安裝erlang支持。
1 Windows平台安裝
1 安裝erlang
首先下載erlang,直接下載最新版本,當前下載的是 OTP 21.3 Windows 64-bit Binary File ,下載完成后一直下一步安裝即可。
2 安裝RabbitMQ
下載Windows平台的RabbtMQ,當前下載的是 rabbitmq-server-3.7.14.exe ,下載完成后一直下一步安裝即可。
3 安裝Web管理插件
打開RabbitMQ Command Prompt,執行命令 rabbitmq-plugins enable rabbitmq_management 即可完成Web監控插件的安裝。
安裝完成后,打開瀏覽器輸入 http://127.0.0.1:15672/ 使用默認賬號[ name:guest / password:guest ]登錄后界面如下,使用這個UI插件我們可以輕松的查看RabbitMQ中的交換機(exchange),隊列(queue)等內容,也可以對exchange,queue,用戶等進行添加、修改、刪除操作。
到這一步Windows平台安裝RabbitMQ完成了。 打開服務管理器,RabbitMQ已經在正常運行了,如下:
2 Centos安裝RabbitMQ
1 安裝RabbitMQ
這里虛擬機系統為Centos7,采用的安裝方式是yum安裝,為了簡單,這里直接使用官方提供的erlang和RabbitMQ-server的自動安裝腳本(官方安裝文檔),逐行執行下邊的代碼就可以安裝完成erlang和RabbitMQ。
#安裝socat yum install socat #安裝erlang curl -s https://packagecloud.io/install/repositories/rabbitmq/erlang/script.rpm.sh | sudo bash yum -y install erlang #安裝rabbitmq-server curl -s https://packagecloud.io/install/repositories/rabbitmq/rabbitmq-server/script.rpm.sh | sudo bash yum -y install rabbitmq-server #啟動rabbitmq服務 systemctl start rabbitmq-server #添加web管理插件 rabbitmq-plugins enable rabbitmq_management
補充:如果安裝完成后,執行RabbitMQ執行命令特別慢,或者出現報錯【rabbitmq unable to perform an operation on node xxx@xxx】,解決方法:
編輯hosts,執行命令 vim /etc/hosts ,添加本機IP(或者虛擬機IP)
命令執行結束后,使用瀏覽器訪問 http://127.0.0.1:15672/ 也會出現web管理界面。通過上邊的安裝步驟安裝的RabbitMQ會生成Unit文件,所以我們可以使用Systemd管理RabbitMQ服務,以下是幾條常用的命令:
#啟動RabbitMQ服務 systemctl start rabbitmq-server #停止RabbitMQ服務 systemctl stop rabbitmq-server #查看RabbitMQ運行狀態 systemctl status rabbitmq-server #重啟RabbitMQ服務 systemctl restart rabbitmq-server
2 RabbitMQ Control工具
使用Web管理界面可以實現RabbitMQ的大部分常用功能,但是有些功能WebUI是做不到的,如:開啟/關閉RabbitMQ應用程序和集群的管理等。RabbitMQ Control是RabbitMQ的命令行管理工具,可以調用所有的RabbitMQ內置功能,主命令是rabbitmqctl ,下邊是一個查詢用戶列表的命令,注意需要切換到sbin目錄下執行:
為了方便的使用RabbitMQ Control工具,我們最好添加一個環境變量,Windows默認安裝時在PATH中添加一條: C:\Program Files\RabbitMQ Server\rabbitmq_server-3.7.14\sbin ,不是默認安裝的話找到對應的安裝目錄添加PATH。按照上的安裝方法,Centos可以直接使用RabbitMQ Control工具,不需要多余的配置。如果想詳細了解RabbitMQ Control工具,可以參考RabbitMQ Control的官方文檔。
這里總結了一些最常用到的RabbitMQ Controll命令,有興趣的小伙伴可以試着運行一下這些命令,如在命令行工具中使用命令 rabbitmqctl add_user <username> <password> 添加一個新用戶。
1 基本控制命令
基本控制命令主要用於啟動、停止應用程序、runtime等
#停止rabbitmq和runtime rabbitmqctl shutdown #停止erlang節點 rabbitmqctl stop #啟用rabbitmq rabbitmqctl start_app #停止rabbitmq rabbitmqctl stop_app #查看狀態 rabbitmqctl status #查看環境 rabbitmqctl environment #rabbitmq恢復最初狀態,內部的exchange和queue都清除 rabbitmqctl reset
2 服務狀態管理
這些命令主要用於用於查看exchang、channel、binding、queue、consumers:
#返回queue的信息 list_queues [-p <vhostpath>] [<queueinfoitem> ...] #返回exchange的信息 list_exchanges [-p <vhostpath>] [<exchangeinfoitem> ...] #返回綁定信息 list_bindings [-p <vhostpath>] [<bindinginfoitem> ...] #返回鏈接信息 list_connections [<connectioninfoitem> ...] #返回目前所有的channels list_channels [<channelinfoitem> ...] #返回consumers list_consumers [-p <vhostpath>]
3 用戶管理命令
這些命令主要用於添加、修改、刪除用戶及管理用戶權限
#在rabbitmq的內部數據庫添加用戶 add_user <username> <password> #刪除一個用戶 delete_user <username> #改變用戶密碼 change_password <username> <newpassword> #清除用戶密碼,禁止用戶登錄 clear_password <username> #設置用戶tags,就是設置用戶角色 set_user_tags <username> <tag> # 查看用戶列表 list_users #創建一個vhost add_vhost <vhostpath> #刪除一個vhosts delete_vhost <vhostpath> #列出vhosts list_vhosts [<vhostinfoitem> ...] #針對一個vhosts 給用戶賦予相關權限 set_permissions [-p <vhostpath>] <user> <conf> <write> <read> #清除一個用戶對vhost的權限 clear_permissions [-p <vhostpath>] <username> #列出所有用戶對某一vhost的權限 list_permissions [-p <vhostpath>] #列出某用戶的訪問權限 list_user_permissions <username>
4 集群管理命令
#clusternode表示node名稱,--ram表示node以ram node加入集群中。默認node以disc node加入集群,在一個node加入cluster之前,必須先停止該node的rabbitmq應用,即先執行stop_app。 join_cluster <clusternode> [--ram] #顯示cluster中的所有node cluster_status #設置集群名字 set_cluster_name <clustername> #修改集群名字 rename_cluster_node <oldname> <newname> #改變一個cluster中node的模式,該節點在轉換前必須先停止,不能把一個集群中唯一的disk node轉化為ram node change_cluster_node_type <disc | ram> #遠程刪除一個節點,刪除前必須該節點必須先停止 rabbitmqctl forget_cluster_node rabbit@rabbit1 #同步鏡像隊列 sync_queue <queuename> #取消同步隊列 cancel_sync_queue <queuename> #清空隊列中所有消息 purge_queue [-p vhost] <queuename>
這里列舉的很多命令是現階段用不到的,如集群控制相關的命令,這些命令的用法會在以后的章節中逐漸理解。
3 C#驅動RabbitMQ
1 一個簡單的栗子
作為開發者,我們最在意的還是怎么在代碼中使用RabbitMQ,可以通過官方RabbitMQ開發文檔來學習RabbitMQ的使用,這里以.NET為例演示一下RabbitMQ的最基本用法。創建兩個Console應用,一個作為發送消息的生產者(Producer),一個作為接受消息的消費者(Consumer),生產者向隊列寫入消息,消費者接受這條消息,結構如下:
兩個控制台應用都要添加RabbitMQ.Client包,命令如下:
Install-Package RabbitMQ.Client
生產者(Producer)代碼:
class Program { static void Main(string[] args) { var factory = new ConnectionFactory() { //rabbitmq-server所在設備ip,這里就是本機 HostName = "127.0.0.1", UserName = "wyy",//用戶名 Password = "123321"//密碼 }; //第一步:創建連接connection using (var connection = factory.CreateConnection()) { //第二步:創建通道channel using (var channel = connection.CreateModel()) { //第三步:聲明交換機exchang channel.ExchangeDeclare(exchange: "myexchange", type: ExchangeType.Direct, durable: true, autoDelete: false, arguments: null); //第四步:聲明隊列queue channel.QueueDeclare(queue: "myqueue", durable: true, exclusive: false, autoDelete: false, arguments: null); Console.WriteLine("生產者准備就緒...."); //第五步:綁定隊列到交互機 channel.QueueBind(queue:"myqueue", exchange:"myexchange", routingKey:"mykey"); string message = ""; //第六步:發送消息 //在控制台輸入消息,按enter鍵發送消息 while (!message.Equals("quit", StringComparison.CurrentCultureIgnoreCase)) { message = Console.ReadLine(); var body = Encoding.UTF8.GetBytes(message); //基本發布 channel.BasicPublish(exchange: "myexchange", routingKey: "mykey", basicProperties: null, body: body); Console.WriteLine($"消息【{message}】已發送到隊列"); } } } Console.ReadKey(); } }
消費者(Consumer)代碼:
class Program { static void Main(string[] args) { var factory = new ConnectionFactory() { //rabbitmq-server所在設備ip,這里就是本機 HostName = "127.0.0.1", UserName = "wyy",//用戶名 Password = "123321"//密碼 }; //第一步:創建連接connection using (var connection = factory.CreateConnection()) { //第二步:創建通道channel using (var channel = connection.CreateModel()) { //第三步:聲明隊列queue channel.QueueDeclare(queue: "myqueue", durable: true, exclusive: false, autoDelete: false, arguments: null); //第四步:定義消費者 var consumer = new EventingBasicConsumer(channel); consumer.Received += (model, ea) => { var body = ea.Body; var message = Encoding.UTF8.GetString(body); Console.WriteLine($"接受到消息【{message}】"); }; Console.WriteLine("消費者准備就緒...."); //第五步:處理消息 channel.BasicConsume(queue: "myqueue", autoAck: true, consumer: consumer); Console.ReadLine(); } } } }
依次運行Producer和Consumer兩個應用程序,運行結果如下:
注意:上邊的代碼在生產者和消費者的代碼中都聲明了exchange和queue,這主要是為了讓這兩個程序可以按任意順序啟動,如:我們只在生產者代碼中定義了exchange和queue,卻先啟動消費者,這會讓造成消費者找不到自己需要的exhange和queue(出現404錯誤)。實際開發中創建exchange/queue、綁定隊列以及設置routingKey這些工作,都可以通過WebUI管理界面或者使用Rabbitmq Control工具完成。
QueueDeclare方法用於聲明隊列,ExchangeDeclare用於聲明交換機,我們在使用這兩個方法聲明時,可以設置隊列和交換機的屬性,如queue的名字,長度限制,exchange是否持久化、交換機類型等。
2 QueueDeclare方法詳解
在上邊的栗子中我們使用了聲明隊列的方法 QueueDeclareOk QueueDeclare(string queue, bool durable, bool exclusive, bool autoDelete, IDictionary<string, object> arguments) ,該方法通過參數設置隊列的特性。這里介紹一下該方法 中幾個參數的作用,先看一個完整的聲明隊列的栗子:
//聲明隊列newsQueue channel.QueueDeclare(queue: "myqueue", durable: false, exclusive: false, autoDelete: false, arguments: new Dictionary<string, object>() { //隊列中消息的過期時間是8s { "x-message-ttl",1000*8 }, //隊列60s沒有被使用,則刪除該隊列 {"x-expires",1000*60 }, //隊列最多保存100條消息 {"x-max-length",100 }, //隊列中ready類型消息總共不能超過1000字節 {"x-max-length-bytes",1000 }, //當隊列消息滿了時,丟棄傳來后續消息 {"x-overflow","reject-publish" }, //丟棄的消息發送到deadExchange交換機 {"x-dead-letter-exchange","deadExchange" }, //丟棄的消息發送到deadExchange交換機時的RoutingKey {"x-dead-letter-routing-key","deadKey" }, //隊列中最大的優先級等級為10(在Publish消息時對每條消息設置優先級) {"x-max-priority",10 }, //設置隊列默認為lazy {"x-queue-mode","lazy" } });
QueueDeclare方法的參數如下:
queue:隊列名字;
durable:是否持久化。設置為true時,隊列信息保存在rabbitmq的內置數據庫中,服務器重啟時隊列也會恢復(注意:重啟后隊列內部的消息不會恢復,怎么實現消息持久化以后會詳細介紹);
exclusive:是否排外。設置為true時只有首次聲明該隊列的Connection可以訪問,其他Connection不能訪問該隊列;且在Connection斷開時,隊列會被刪除(即使durable設置為true也會被刪除);
autoDelete:是否自動刪除。設置為true時,表示在最后一條使用該隊列的連接(Connection)斷開時,將自動刪除這個隊列;
arguments:設置隊列的一些其它屬性,為Dictionary<string,object>類型,下表總結了arguments中可以設置的常用屬性。
參數名 | 作用 | 示例 |
Message TTL | 設置隊列中消息的有效時間 | { "x-message-ttl",1000*8 },設置隊列中的所有消息的有效期為8s; |
Auto expire | 自動刪除隊列。一定的時間內隊列沒有被使用,則自動刪除隊列 | {"x-expires",1000*60 },設置隊列的過期時長為60s,如果60s沒有隊列被訪問,則刪除隊列; |
Max length | 隊列能保存消息的最大條數 | {"x-max-length",100 },設置隊列最多保存100條消息; |
Max length bytes | 隊列中ready類型消息的總字節數 | {"x-max-length-bytes",1000 }, 設置隊列中ready類型消息總共不能超過1000字節; |
Overflow behaviour | 當隊列消息滿了時,再接收消息時的處理方法。有兩種處理方案:默認為"drop-head"模式,表示從隊列頭部丟棄消息;"reject-publish "表示不接收后續的消息 |
{"x-overflow","reject-publish" },設置當隊列消息滿了時,丟棄傳來后續消息; |
Dead letter exchange | 用於存儲被丟棄的消息的交換機名。Overflow behaviour 的兩種處理方案中丟棄的消息都會發送到這個交換機 | {"x-dead-letter-exchange","beiyongExchange" },設置丟棄的消息發送到名字位beiyongExchange的交換機; |
Dead letter routing key | 被丟棄的消息發送到Dead letter exchange時的使用的routing Key | {"x-dead-letter-routing-key","deadKey" },設置丟棄的消息發送到beiyongExchange交換機時的RoutingKey值是"deadKey"; |
Maximum priority | 設置隊列中消息優先級的最大等級,在publish消息時可以設置單條消息的優先級等級 | {"x-max-priority",10 },設置中消息優先級的最大等級為10; |
Lazy mode | 設置隊列的模式,如果設置為Lazy表示隊列中消息盡可能存放在磁盤中,以減少內存占用;不設置時消息都存放在隊列中,用以盡可能快的處理消息 | {"x-queue-mode","lazy"},3.6以后版本可用,設置隊列中消息盡可能存放在磁盤中,以減少內存占用。在消息擁堵時和消息持久化配置使用可以減少內存占用。 |
3 ExchangeDeclare方法詳解
聲明交換機的方法 void ExchangeDeclare(string exchange, string type, bool durable, bool autoDelete, IDictionary<string, object> arguments) 可以設置交換機的特性,這里簡單介紹一下這個方法的幾個參數:
channel.ExchangeDeclare(exchange: "myexchange", type: ExchangeType.Direct, durable: true, autoDelete: false, arguments: new Dictionary<string, object> {
{"alternate-exchange","BeiyongExchange" }//如果消息不能路由到該交換機,就把消息路由到備用交換機BeiyongExchange上
});
exchange:交換機名字。
type:交換機類型。exchange有direct、fanout、topic、header四種類型,在下一篇會詳細介紹;
durable:是否持久化。設置為true時,交換機信息保存在rabbitmq的內置數據庫中,服務器重啟時交換機信息也會恢復;
autoDelete:是否自動刪除。設置為true時,表示在最后一條使用該交換機的連接(Connection)斷開時,自動刪除這個exchange;
arguments:其他的一些參數,類型為Dictionary<string,object> 。
小結
本節主要介紹了RabbitMQ的基本概念,在Windows和Centos上的安裝方法,及RabbitMQ Control工具的基本使用,最后演示了一個C#驅動RabbitMQ的栗子,並詳細介紹了聲明queue和exchange的方法。通過這一節我們大概了解了RabbitMQ的基本使用。以后的章節會逐漸介紹RabbitMQ的四種exchange、兩種Consumer的特點和使用場景,以及消息確認、優先級、持久化等,最后搭建一個高可用的RabbitMQ集群,如果文中有錯誤的話,希望大家可以指出,我會及時修改,謝謝!
參考文章
【1】 https://www.cnblogs.com/dwlsxj/p/RabbitMQ.html
【2】https://www.cnblogs.com/operationhome/p/10483840.html