消息隊列之 RabbitMQ
消息隊列之 RabbitMQ
關于消息隊列，從前年開始斷斷續續看了些資料，想寫很久了，但一直沒騰出空，近來分別碰到幾個朋友聊這塊的技術選型，是時候把這塊的知識整理記錄一下了。
市面上的消息隊列產品有很多，比如老牌的 ActiveMQ、RabbitMQ ，目前我看最火的 Kafka ，還有 ZeroMQ ，去年底阿里巴巴捐贈給 Apache 的 RocketMQ ，連 redis 這樣的 NoSQL 數據庫也支持 MQ 功能?？傊@塊知名的產品就有十幾種，就我自己的使用經驗和興趣只打算談談 RabbitMQ、Kafka 和 ActiveMQ ，本文先講 RabbitMQ ，在此之前先看下消息隊列的相關概念。
什么叫消息隊列
消息（Message）是指在應用間傳送的數據。消息可以非常簡單，比如只包含文本字符串，也可以更復雜，可能包含嵌入對象。
消息隊列（Message Queue）是一種應用間的通信方式，消息發送后可以立即返回，由消息系統來確保消息的可靠傳遞。消息發布者只管把消息發布到 MQ 中而不用管誰來取，消息使用者只管從 MQ 中取消息而不管是誰發布的。這樣發布者和使用者都不用知道對方的存在。
為何用消息隊列
從上面的描述中可以看出消息隊列是一種應用間的異步協作機制，那什么時候需要使用 MQ 呢？
以常見的訂單系統為例，用戶點擊【下單】按鈕之后的業務邏輯可能包括：扣減庫存、生成相應單據、發紅包、發短信通知。在業務發展初期這些邏輯可能放在一起同步執行，隨著業務的發展訂單量增長，需要提升系統服務的性能，這時可以將一些不需要立即生效的操作拆分出來異步執行，比如發放紅包、發短信通知等。這種場景下就可以用 MQ ，在下單的主流程（比如扣減庫存、生成相應單據）完成之后發送一條消息到 MQ 讓主流程快速完結，而由另外的單獨線程拉取MQ的消息（或者由 MQ 推送消息），當發現 MQ 中有發紅包或發短信之類的消息時，執行相應的業務邏輯。
以上是用于業務解耦的情況，其它常見場景包括最終一致性、廣播、錯峰流控等等。
RabbitMQ 特點
RabbitMQ 是一個由 Erlang 語言開發的 AMQP 的開源實現。
AMQP ：Advanced Message Queue，高級消息隊列協議。它是應用層協議的一個開放標準，為面向消息的中間件設計，基于此協議的客戶端與消息中間件可傳遞消息，并不受產品、開發語言等條件的限制。
RabbitMQ 最初起源于金融系統，用于在分布式系統中存儲轉發消息，在易用性、擴展性、高可用性等方面表現不俗。具體特點包括：
可靠性（Reliability） RabbitMQ 使用一些機制來保證可靠性，如持久化、傳輸確認、發布確認。
靈活的路由（Flexible Routing） 在消息進入隊列之前，通過 Exchange 來路由消息的。對于典型的路由功能，RabbitMQ 已經提供了一些內置的 Exchange 來實現。針對更復雜的路由功能，可以將多個 Exchange 綁定在一起，也通過插件機制實現自己的 Exchange 。
消息集群（Clustering） 多個 RabbitMQ 服務器可以組成一個集群，形成一個邏輯 Broker 。
高可用（Highly Available Queues） 隊列可以在集群中的機器上進行鏡像，使得在部分節點出問題的情況下隊列仍然可用。
多種協議（Multi-protocol） RabbitMQ 支持多種消息隊列協議，比如 STOMP、MQTT 等等。
多語言客戶端（Many Clients） RabbitMQ 幾乎支持所有常用語言，比如 Java、.NET、Ruby 等等。
管理界面（Management UI） RabbitMQ 提供了一個易用的用戶界面，使得用戶可以監控和管理消息 Broker 的許多方面。
跟蹤機制（Tracing） 如果消息異常，RabbitMQ 提供了消息跟蹤機制，使用者可以找出發生了什么。
插件機制（Plugin System） RabbitMQ 提供了許多插件，來從多方面進行擴展，也可以編寫自己的插件。
RabbitMQ 中的概念模型
消息模型
所有 MQ 產品從模型抽象上來說都是一樣的過程： 消費者（consumer）訂閱某個隊列。生產者（producer）創建消息，然后發布到隊列（queue）中，最后將消息發送到監聽的消費者。
RabbitMQ 基本概念
上面只是最簡單抽象的描述，具體到 RabbitMQ 則有更詳細的概念需要解釋。上面介紹過 RabbitMQ 是 AMQP 協議的一個開源實現，所以其內部實際上也是 AMQP 中的基本概念：
Message 消息，消息是不具名的，它由消息頭和消息體組成。消息體是不透明的，而消息頭則由一系列的可選屬性組成，這些屬性包括routing-key（路由鍵）、priority（相對于其他消息的優先權）、delivery-mode（指出該消息可能需要持久性存儲）等。
Publisher 消息的生產者，也是一個向交換器發布消息的客戶端應用程序。
Exchange 交換器，用來接收生產者發送的消息并將這些消息路由給服務器中的隊列。
Binding 綁定，用于消息隊列和交換器之間的關聯。一個綁定就是基于路由鍵將交換器和消息隊列連接起來的路由規則，所以可以將交換器理解成一個由綁定構成的路由表。
Queue 消息隊列，用來保存消息直到發送給消費者。它是消息的容器，也是消息的終點。一個消息可投入一個或多個隊列。消息一直在隊列里面，等待消費者連接到這個隊列將其取走。
Connection 網絡連接，比如一個TCP連接。
Channel 信道，多路復用連接中的一條獨立的雙向數據流通道。信道是建立在真實的TCP連接內地虛擬連接，AMQP 命令都是通過信道發出去的，不管是發布消息、訂閱隊列還是接收消息，這些動作都是通過信道完成。因為對于操作系統來說建立和銷毀 TCP 都是非常昂貴的開銷，所以引入了信道的概念，以復用一條 TCP 連接。
Consumer 消息的消費者，表示一個從消息隊列中取得消息的客戶端應用程序。
Virtual Host 虛擬主機，表示一批交換器、消息隊列和相關對象。虛擬主機是共享相同的身份認證和加密環境的獨立服務器域。每個 vhost 本質上就是一個 mini 版的 RabbitMQ 服務器，擁有自己的隊列、交換器、綁定和權限機制。vhost 是 AMQP 概念的基礎，必須在連接時指定，RabbitMQ 默認的 vhost 是 / 。
Broker 表示消息隊列服務器實體。
AMQP 中的消息路由
AMQP 中消息的路由過程和 Java 開發者熟悉的 JMS 存在一些差別，AMQP 中增加了 Exchange 和 Binding 的角色。生產者把消息發布到 Exchange 上，消息最終到達隊列并被消費者接收，而 Binding 決定交換器的消息應該發送到那個隊列。
Exchange 類型
Exchange分發消息時根據類型的不同分發策略有區別，目前共四種類型：direct、fanout、topic、headers 。headers 匹配 AMQP 消息的 header 而不是路由鍵，此外 headers 交換器和 direct 交換器完全一致，但性能差很多，目前幾乎用不到了，所以直接看另外三種類型：
direct
消息中的路由鍵（routing key）如果和 Binding 中的 binding key 一致， 交換器就將消息發到對應的隊列中。路由鍵與隊列名完全匹配，如果一個隊列綁定到交換機要求路由鍵為“dog”，則只轉發 routing key 標記為“dog”的消息，不會轉發“dog.puppy”，也不會轉發“dog.guard”等等。它是完全匹配、單播的模式。
fanout
每個發到 fanout 類型交換器的消息都會分到所有綁定的隊列上去。fanout 交換器不處理路由鍵，只是簡單的將隊列綁定到交換器上，每個發送到交換器的消息都會被轉發到與該交換器綁定的所有隊列上。很像子網廣播，每臺子網內的主機都獲得了一份復制的消息。fanout 類型轉發消息是最快的。
topic
topic 交換器通過模式匹配分配消息的路由鍵屬性，將路由鍵和某個模式進行匹配，此時隊列需要綁定到一個模式上。它將路由鍵和綁定鍵的字符串切分成單詞，這些單詞之間用點隔開。它同樣也會識別兩個通配符：符號“#”和符號“”。#匹配0個或多個單詞，匹配不多不少一個單詞。
RabbitMQ 安裝
一般來說安裝 RabbitMQ 之前要安裝 Erlang ，可以去Erlang官網下載。接著去RabbitMQ官網下載安裝包，之后解壓縮即可。根據操作系統不同官網提供了相應的安裝說明：Windows、Debian / Ubuntu、RPM-based Linux、Mac
如果是Mac 用戶，個人推薦使用 HomeBrew 來安裝，安裝前要先更新 brew：
brew?update
接著安裝 rabbitmq 服務器：
brew?install?rabbitmq
這樣 RabbitMQ 就安裝好了，安裝過程中會自動其所依賴的 Erlang 。
RabbitMQ 運行和管理 啟動 啟動很簡單，找到安裝后的 RabbitMQ 所在目錄下的 sbin 目錄，可以看到該目錄下有6個以 rabbitmq 開頭的可執行文件，直接執行 rabbitmq-server 即可，下面將 RabbitMQ 的安裝位置以 . 代替，啟動命令就是：
./sbin/rabbitmq-server
啟動正常的話會看到一些啟動過程信息和最后的 completed with 7 plugins，這也說明啟動的時候默認加載了7個插件。
后臺啟動
如果想讓 RabbitMQ 以守護程序的方式在后臺運行，可以在啟動的時候加上 -detached 參數：
./sbin/rabbitmq-server?-detached
查詢服務器狀態
sbin 目錄下有個特別重要的文件叫 rabbitmqctl ，它提供了 RabbitMQ 管理需要的幾乎一站式解決方案，絕大部分的運維命令它都可以提供。 查詢 RabbitMQ 服務器的狀態信息可以用參數 status ：
./sbin/rabbitmqctl?status
該命令將輸出服務器的很多信息，比如 RabbitMQ 和 Erlang 的版本、OS 名稱、內存等等
關閉 RabbitMQ 節點
我們知道 RabbitMQ 是用 Erlang 語言寫的，在Erlang 中有兩個概念：節點和應用程序。節點就是 Erlang 虛擬機的每個實例，而多個 Erlang 應用程序可以運行在同一個節點之上。節點之間可以進行本地通信（不管他們是不是運行在同一臺服務器之上）。比如一個運行在節點A上的應用程序可以調用節點B上應用程序的方法，就好像調用本地函數一樣。如果應用程序由于某些原因奔潰，Erlang 節點會自動嘗試重啟應用程序。 如果要關閉整個 RabbitMQ 節點可以用參數 stop ：
./sbin/rabbitmqctl?stop
它會和本地節點通信并指示其干凈的關閉，也可以指定關閉不同的節點，包括遠程節點，只需要傳入參數 -n ：
./sbin/rabbitmqctl?-n?rabbit@server.example.com?stop
-n node 默認 node 名稱是 rabbit@server ，如果你的主機名是 server.example.com ，那么 node 名稱就是 rabbit@server.example.com 。
關閉 RabbitMQ 應用程序
如果只想關閉應用程序，同時保持 Erlang 節點運行則可以用 stopapp： ./sbin/rabbitmqctl stopapp 這個命令在后面要講的集群模式中將會很有用。
啟動 RabbitMQ 應用程序
./sbin/rabbitmqctl?start_app
重置 RabbitMQ 節點
./sbin/rabbitmqctl?reset
該命令將清除所有的隊列。
查看已聲明的隊列
./sbin/rabbitmqctl?list_queues
查看交換器
./sbin/rabbitmqctl?list_exchanges
該命令還可以附加參數，比如列出交換器的名稱、類型、是否持久化、是否自動刪除：
./sbin/rabbitmqctl?list_exchanges?name?type?durable?auto_delete
查看綁定
./sbin/rabbitmqctl?list_bindings
Java 客戶端訪問
RabbitMQ 支持多種語言訪問，以 Java 為例看下一般使用 RabbitMQ 的步驟。
maven工程的pom文件中添加依賴
消息生產者:
package?org.study.rabbitmq;import?com.rabbitmq.client.Channel;import?com.rabbitmq.client.Connection;import?com.rabbitmq.client.ConnectionFactory;import?java.io.IOException;import?java.util.concurrent.TimeoutException;public?class?Producer?{????public?static?void?main(String[]?args)?throws?IOException,?TimeoutException?{????????//創建連接工廠????????ConnectionFactory?factory?=?new?ConnectionFactory();????????factory.setUsername("guest");????????factory.setPassword("guest");????????//設置?RabbitMQ?地址????????factory.setHost("localhost");????????//建立到代理服務器到連接????????Connection?conn?=?factory.newConnection();????????//獲得信道????????Channel?channel?=?conn.createChannel();????????//聲明交換器????????String?exchangeName?=?"hello-exchange";????????channel.exchangeDeclare(exchangeName,?"direct",?true);????????String?routingKey?=?"hola";????????//發布消息????????byte[]?messageBodyBytes?=?"quit".getBytes();????????channel.basicPublish(exchangeName,?routingKey,?null,?messageBodyBytes);????????channel.close();????????conn.close();????}}
消息消費者
package?org.study.rabbitmq;import?com.rabbitmq.client.*;import?java.io.IOException;import?java.util.concurrent.TimeoutException;public?class?Consumer?{????public?static?void?main(String[]?args)?throws?IOException,?TimeoutException?{????????ConnectionFactory?factory?=?new?ConnectionFactory();????????factory.setUsername("guest");????????factory.setPassword("guest");????????factory.setHost("localhost");????????//建立到代理服務器到連接????????Connection?conn?=?factory.newConnection();????????//獲得信道????????final?Channel?channel?=?conn.createChannel();????????//聲明交換器????????String?exchangeName?=?"hello-exchange";????????channel.exchangeDeclare(exchangeName,?"direct",?true);????????//聲明隊列????????String?queueName?=?channel.queueDeclare().getQueue();????????String?routingKey?=?"hola";????????//綁定隊列，通過鍵?hola?將隊列和交換器綁定起來????????channel.queueBind(queueName,?exchangeName,?routingKey);????????while(true)?{????????????//消費消息????????????boolean?autoAck?=?false;????????????String?consumerTag?=?"";????????????channel.basicConsume(queueName,?autoAck,?consumerTag,?new?DefaultConsumer(channel)?{????????????????@Override????????????????public?void?handleDelivery(String?consumerTag,???????????????????????????????????????????Envelope?envelope,???????????????????????????????????????????AMQP.BasicProperties?properties,???????????????????????????????????????????byte[]?body)?throws?IOException?{????????????????????String?routingKey?=?envelope.getRoutingKey();????????????????????String?contentType?=?properties.getContentType();????????????????????System.out.println("消費的路由鍵："?+?routingKey);????????????????????System.out.println("消費的內容類型："?+?contentType);????????????????????long?deliveryTag?=?envelope.getDeliveryTag();????????????????????//確認消息????????????????????channel.basicAck(deliveryTag,?false);????????????????????System.out.println("消費的消息體內容：");????????????????????String?bodyStr?=?new?String(body,?"UTF-8");????????????????????System.out.println(bodyStr);????????????????}????????????});????????}????}}
啟動 RabbitMQ 服務器:
./sbin/rabbitmq-server
運行 Consumer 先運行 Consumer ，這樣當生產者發送消息的時候能在消費者后端看到消息記錄。
運行 Producer 接著運行 Producer ,發布一條消息，在 Consumer 的控制臺能看到接收的消息：
RabbitMQ 集群
RabbitMQ 最優秀的功能之一就是內建集群，這個功能設計的目的是允許消費者和生產者在節點崩潰的情況下繼續運行，以及通過添加更多的節點來線性擴展消息通信吞吐量。RabbitMQ 內部利用 Erlang 提供的分布式通信框架 OTP 來滿足上述需求，使客戶端在失去一個 RabbitMQ 節點連接的情況下，還是能夠重新連接到集群中的任何其他節點繼續生產、消費消息。
RabbitMQ 集群中的一些概念
RabbitMQ 會始終記錄以下四種類型的內部元數據：
隊列元數據 包括隊列名稱和它們的屬性，比如是否可持久化，是否自動刪除
交換器元數據 交換器名稱、類型、屬性
綁定元數據 內部是一張表格記錄如何將消息路由到隊列
vhost 元數據
為 vhost 內部的隊列、交換器、綁定提供命名空間和安全屬性 在單一節點中，RabbitMQ 會將所有這些信息存儲在內存中，同時將標記為可持久化的隊列、交換器、綁定存儲到硬盤上。存到硬盤上可以確保隊列和交換器在節點重啟后能夠重建。而在集群模式下同樣也提供兩種選擇：存到硬盤上（獨立節點的默認設置），存在內存中。
如果在集群中創建隊列，集群只會在單個節點而不是所有節點上創建完整的隊列信息（元數據、狀態、內容）。結果是只有隊列的所有者節點知道有關隊列的所有信息，因此當集群節點崩潰時，該節點的隊列和綁定就消失了，并且任何匹配該隊列的綁定的新消息也丟失了。還好RabbitMQ 2.6.0之后提供了鏡像隊列以避免集群節點故障導致的隊列內容不可用。
RabbitMQ 集群中可以共享 user、vhost、exchange等，所有的數據和狀態都是必須在所有節點上復制的，例外就是上面所說的消息隊列。RabbitMQ 節點可以動態的加入到集群中。
當在集群中聲明隊列、交換器、綁定的時候，這些操作會直到所有集群節點都成功提交元數據變更后才返回。集群中有內存節點和磁盤節點兩種類型，內存節點雖然不寫入磁盤，但是它的執行比磁盤節點要好。內存節點可以提供出色的性能，磁盤節點能保障配置信息在節點重啟后仍然可用，那集群中如何平衡這兩者呢？
RabbitMQ 只要求集群中至少有一個磁盤節點，所有其他節點可以是內存節點，當節點加入火離開集群時，它們必須要將該變更通知到至少一個磁盤節點。如果只有一個磁盤節點，剛好又是該節點崩潰了，那么集群可以繼續路由消息，但不能創建隊列、創建交換器、創建綁定、添加用戶、更改權限、添加或刪除集群節點。換句話說集群中的唯一磁盤節點崩潰的話，集群仍然可以運行，但知道該節點恢復，否則無法更改任何東西。
RabbitMQ 集群配置和啟動 如果是在一臺機器上同時啟動多個 RabbitMQ 節點來組建集群的話，只用上面介紹的方式啟動第二、第三個節點將會因為節點名稱和端口沖突導致啟動失敗。所以在每次調用 rabbitmq-server 命令前，設置環境變量 RABBITMQNODENAME 和 RABBITMQNODEPORT 來明確指定唯一的節點名稱和端口。下面的例子端口號從5672開始，每個新啟動的節點都加1，節點也分別命名為testrabbit1、testrabbit2、testrabbit_3。
啟動第1個節點：
RABBITMQ_NODENAME=test_rabbit_1?RABBITMQ_NODE_PORT=5672?./sbin/rabbitmq-server?-detached
啟動第2個節點：
RABBITMQ_NODENAME=test_rabbit_2?RABBITMQ_NODE_PORT=5673?./sbin/rabbitmq-server?-detached
啟動第2個節點前建議將 RabbitMQ 默認激活的插件關掉，否則會存在使用了某個插件的端口號沖突，導致節點啟動不成功。
現在第2個節點和第1個節點都是獨立節點，它們并不知道其他節點的存在。集群中除第一個節點外后加入的節點需要獲取集群中的元數據，所以要先停止 Erlang 節點上運行的 RabbitMQ 應用程序，并重置該節點元數據，再加入并且獲取集群的元數據，最后重新啟動 RabbitMQ 應用程序。
停止第2個節點的應用程序：
./sbin/rabbitmqctl?-n?test_rabbit_2?stop_app
重置第2個節點元數據：
./sbin/rabbitmqctl?-n?test_rabbit_2?reset
第2節點加入第1個節點組成的集群：
./sbin/rabbitmqctl?-n?test_rabbit_2?join_cluster?test_rabbit_1@localhost
啟動第2個節點的應用程序
./sbin/rabbitmqctl?-n?test_rabbit_2?start_app
第3個節點的配置過程和第2個節點類似：
RABBITMQ_NODENAME=test_rabbit_3?RABBITMQ_NODE_PORT=5674?./sbin/rabbitmq-server?-detached./sbin/rabbitmqctl?-n?test_rabbit_3?stop_app./sbin/rabbitmqctl?-n?test_rabbit_3?reset./sbin/rabbitmqctl?-n?test_rabbit_3?join_cluster?test_rabbit_1@localhost./sbin/rabbitmqctl?-n?test_rabbit_3?start_app
RabbitMQ 集群運維
停止某個指定的節點，比如停止第2個節點：
RABBITMQ_NODENAME=test_rabbit_2?./sbin/rabbitmqctl?stop
查看節點3的集群狀態：
./sbin/rabbitmqctl?-n?test_rabbit_3?cluster_status
https://www.jianshu.com/p/79ca08116d57
-結束-
近期文章推薦：
使用Kubespray部署Kubernetes集群
談談 API 網關
一個忙碌架構師的Java后端書架-2018
限流降級神器-哨兵(sentinel)原理分析
渣學歷小公司的我，是如何靠自己在北京買房的
Docker入門與實踐
一張圖看懂JVM
詳解JVM內存管理與垃圾回收機制1 - 內存管理
詳解JVM內存管理與垃圾回收機制2 - 何為垃圾
劍指“大廠”，擺脫金九銀十的職業“瓶頸”
-關注我-
https://mp.weixin.qq.com/s?__biz=MzAxNjk4ODE4OQ==&mid=2247484485&idx=1&sn=1e76bb6d5234ebb5bfff2325e7ad312d&chksm=9bed2537ac9aac21d32ffa0858678f3a3866634d5fd578e405f1ab2e80ce5eb01e7807b1562a&scene=21#wechat_redirect
HTTP RabbitMQ
版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。
相關文章
