# RabbitMQ概述和運行機制 # RabbitMQ概述 **RabbitMQ概述**：RabbitMQ是使用最廣泛的開源消息代理。RabbitMQ輕量級，易于在集群內部和云平臺中部署。它支持多種消息傳遞協議。 它可以滿足企業高規模，高可用性的要求。RabbitMQ使用Erlang語言開發的。 **MQ概述**：全稱為Message Queue, 消息隊列（MQ）是一種應用程序對應用程序的通信方法。應用程序通過讀寫隊列的消息（針對應用程序的數據）來通信，而無需專用連接來鏈接它們。 **MQ運行機制**： P表示生產者，C表示消費者，紅色部分為消息隊列  ## MQ實戰場景： 1.我們在雙11的時候，當我們凌晨大量的秒殺和搶購商品，然后去結算的時候，就會發現，界面會提醒我們，讓我們稍等，以及一些友好的圖片文字提醒。而不是像前幾年的時代，動不動就頁面卡死，報錯等來呈現給用戶。在這個業務場景中，我們就可以采用隊列的機制來處理，因為同時結算就只能達到這么多。   2.在我們平時的超市中購物也是一樣，當我們在結算的時候，并不會一窩蜂一樣涌入收銀臺，而是排隊結算。這也是隊列機制。一個接著一個的處理，不能插隊。  ***** **RabbitMQ 是一個由 Erlang 語言開發的 AMQP 的開源實現。** RabbitMQ是AMQP服務器的一種。 **AMQP簡介**：AMQP，即Advanced Message Queuing Protocol，**高級消息隊列協議**，是應用層協議的一個開放標準，為面向消息的中間件設計。消息中間件主要用于組件之間的解耦，消息的發送者無需知道消息使用者的存在，反之亦然。 AMQP的主要特征是面向消息、隊列、路由（包括點對點和發布/訂閱）、可靠性、安全。 它是應用層協議的一個開放標準，為面向消息的中間件設計，基于此協議的客戶端與消息中間件可傳遞消息，**并不受產品、開發語言等條件的限制。** AMQP 里主要要說兩個組件：Exchange 和 Queue （在 AMQP 1.0 里還會有變動），如下圖所示，綠色的 X 就是 Exchange ，紅色的是 Queue ，這兩者都在 Server 端，又稱作 Broker ，這部分是 RabbitMQ 實現的，而藍色的則是客戶端，通常有 Producer（生產者） 和 Consumer（消費者） 兩種類型：  Publisher 消息的生產者，也是一個向交換器發布消息的客戶端應用程序。 Exchange 交換器，用來接收生產者發送的消息并將這些消息路由給服務器中的隊列。 Queue 消息隊列，用來保存消息直到發送給消費者。它是消息的容器，也是消息的終點。一個消息可投入一個或多個隊列。消息一直在隊列里面，等待消費者連接到這個隊列將其取走。 Consumer 消息的消費者，表示一個從消息隊列中取得消息的客戶端應用程序。 Broker 表示消息隊列服務器實體。 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 安裝 RabbitMQ是用erlang語言編寫的，所以我們先安裝erlang語言環境 配置erlang語言環境 ``` vim /etc/yum.repos.d/rabbitmq-erlang.repo # 在rabbitmq-erlang.repo 文件中加入下面的代碼 [rabbitmq-erlang] name=rabbitmq-erlang baseurl=https://dl.bintray.com/rabbitmq/rpm/erlang/20/el/7 gpgcheck=1 gpgkey=https://dl.bintray.com/rabbitmq/Keys/rabbitmq-release-signing-key.asc enabled=1 # 執行導入key rpm --import https://dl.bintray.com/rabbitmq/Keys/rabbitmq-release-signing-key.asc # 安裝erlang yum install erlang -y #安裝erlang ``` **安裝rabbitmq服務** 下載rabbitmq 地址：http://www.rabbitmq.com/download.html ``` wget https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.7.8/rabbitmq-server-3.7.8-1.el7.noarch.rpm yum install rabbitmq-server-3.7.8-1.el7.noarch.rpm ``` 啟用RabbitMQ的web插件 ，方便后期管理界面 ``` rabbitmq-plugins enable rabbitmq_management ``` ``` The following plugins have been configured: rabbitmq_management rabbitmq_management_agent rabbitmq_web_dispatch Applying plugin configuration to rabbit@localhost... The following plugins have been enabled: rabbitmq_management rabbitmq_management_agent rabbitmq_web_dispatch set 3 plugins. Offline change; changes will take effect at broker restart. ``` 設置開機啟動 ``` systemctl enable rabbitmq-server.service ``` 啟動服務 ``` systemctl start rabbitmq-server ``` rabbitmq配置文件位置 ``` ls /var/lib/rabbitmq/mnesia ``` 訪問控制臺 默認用戶名和密碼： guest/guest 。guest用戶僅允許從在服務器以localhost或127.0.0.1作為ip登錄 如果遠程登錄，如：http://192.168.1.63:15672/, 則會提示錯誤，登錄不了。 為RabbitMQ創建用戶并賦權。 ``` rabbitmqctl add_user root 123456 #添加用戶 rabbitmqctl set_user_tags root administrator #設置用戶權限為administrator ``` 到此，已經搭建成功。 # RabbitMQ使用方法 **RabbitMQ查看相關的命令** ``` rabbitmqctl list_connections #用于查看當前的連接 rabbitmqctl list_queues #會列出所有隊列名稱，后邊可能還會帶著這個隊列當前消息數 rabbitmqctl status #查看當前隊列信息 ``` ## RabbitMQ的vhost管理 當我們在創建用戶時，會指定用戶能訪問一個虛擬機，并且該用戶只能訪問該虛擬機下的隊列和交換機，如果沒有指定，默認的是”/”;一個rabbitmq服務器上可以運行多個vhost，以便于適用不同的業務需要，這樣做既可以滿足權限配置的要求，也可以避免不同業務之間隊列、交換機的命名沖突問題，因為不同vhost之間是隔離的。 添加yang-web和yang-bbs兩個虛擬機來管理網站和論壇的隊列 ``` rabbitmqctl add_vhost yang-web rabbitmqctl add_vhost yang-bbs ``` 查看創建的虛擬主機 網頁查看 刪除bbs虛擬機 ``` rabbitmqctl delete_vhost yang-bbs ``` 查看虛擬機列表 ``` rabbitmqctl list_vhosts ``` **“/”是rabbitmq默認的虛擬機，之前默認連接的都是它** # RabbitMQ管理用戶、角色和權限管理 1、用戶管理語法 添加用戶：rabbitmqctl add_user {username} {password} 刪除用戶：rabbitmqctl delete_user {username} 修改密碼：rabbitmqctl change_password {username} {newpassword} 2、角色權限分配 設置用戶角色語法：rabbitmqctl set_user_tags {username} {tag} RabbitMQ的tag用戶角色分類：none、management、policymaker、monitoring、administrator tag常用角色為：administrator, monitoring, management ### RabbitMQ各類角色描述： （1）、none角色權限 ：不能訪問 management plugin （2）、management角色權限： 列出自己可以通過AMQP登入的virtual hosts 查看自己的virtual hosts中的queues, exchanges 和 bindings 查看和關閉自己的channels 和 connections 查看有關自己的virtual hosts的“全局”的統計信息，包含其他用戶在這些virtual hosts中的活動。 （3）、policymaker角色權限 #policymaker ['p?l?s?me?k?(r)] 決策者 擁有management的所有權限，還擁有查看、創建和刪除自己的virtual hosts所屬的policies（策略）和parameters（[p??r?m?t?(r)] 參數 ） （4）、monitoring 角色權限 擁有management的所有權限，還擁有： 列出所有virtual hosts，包括他們不能登錄的virtual hosts 查看其他用戶的connections和channels 查看節點級別的數據如clustering和memory使用情況 查看真正的關于所有virtual hosts的全局的統計信息 （5）、administrator角色權限 擁有policymaker和monitoring的所有權限，還擁有： 創建和刪除virtual hosts 查看、創建和刪除users 查看創建和刪除permissions 關閉其他用戶的connections TODO # 使用python調用rabbitmq服務器 ``` pip install pika #安裝pika模塊。python用pika模塊調用rabbitmq。 ``` **注：** rabbitmq本質是一個生產者和消費者的模型結構。生產者->rabbitmq->消費者，即生產者產生消息，給到rabbitmq存儲，消費者從rabbitmq中讀取數據。 創建生產者代碼send.py ``` import pika #啟用對庫進行阻塞，同步操作以進行簡單的使用 # 連接隊列到服務器 #這里可以連接遠程IP，請記得打開遠程端口 connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost')) #聲明一個管道 channel = connection.channel() # 創建隊列。有就不管，沒有就自動創建 #聲明一個隊列，生產者和消費者都要聲明一個相同的隊列，用來防止萬一某一方掛了，另一方能正常運行 channel.queue_declare(queue='hello') # 使用默認的交換機發送信息。exchange為空就使用默認的 channel.basic_publish(exchange='', routing_key='hello', # queue名字 #路由鍵，寫明將消息發往哪個隊列，本例是將消息發往隊列hello body='Hello World!') # 消息內容 print(" [x] Sent 'Hello World!'") connection.close() 注：declare [d??kle?(r)] 聲明 ； consuming [k?n?sju:m??] 消費；publish [?p?bl??] 頒布 ``` 創建消耗者代碼receive.py ``` import pika # 連接服務器 connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost')) channel = connection.channel() # rabbitmq消費端仍然使用此方法創建隊列。這樣做的意思是：若沒有就創建，和發送端道理一樣。目的是為了保證隊列一定會有 channel.queue_declare(queue='hello') # 收到消息后的回調 def callback(ch, method, properties, body): print(" [x] Received %r" % body) channel.basic_consume(callback, #回調函數。執行結束后立即執行另外一個函數返回給發送端是否執行完畢。 queue='hello', no_ack=True) #不會告知服務端我是否收到消息。一般注釋。#如果注釋掉，對方沒有收到消息的話不會將消息丟失，始終在隊列里等待下次發送。 print(' [*] Waiting for messages. To exit press CTRL+C') channel.start_consuming() ``` ## 開始測試隊列 python send.py #此命令執行兩次，產生兩個名字叫hello的消息 rabbitmqctl list_queues #查看消息隊列為2 在web界面查看消息隊列，發現在2個消息隊列等待處理： python receive.py #消費或處理這2個消息 rabbitmqctl list_queues #查看隊列，已經為0 在web界面，查看隊列，也為0了 消息被消費后，隊列就相應的移除 # 交換機 消息實際上投遞到的是交換機，具體路由到那個隊列由交換機根據路由鍵(routing key)完成。 當你發消息到代理服務器時，即便路由鍵是空的，RabbitMQ也會將其和使用的路由鍵進行匹配。如果路由的消息不匹配任何綁定模式，消息將會進入黑洞。 交換機在隊列與消息中間起到了中間層的作用，有了交換機我們可以實現更靈活的功能，RabbitMQ中有三種常用的交換機類型： direct: 如果路由鍵匹配，消息就投遞到對應的隊列 fanout：投遞消息給所有綁定在當前交換機上面的隊列 topic：允許實現有趣的消息通信場景，使得5不同源頭的消息能夠達到同一個隊列。topic隊列名稱有兩個特殊的關鍵字。 * 可以替換一個單詞 可以替換所有的單詞 ： 可以理解，direct為1v1, fanout為1v所有，topic比較靈活，可以1v任意。  # 虛擬主機 每一個虛擬主機(vhost)相當于mini版的RabbitMQ服務器，擁有自己的隊列，交換機和綁定，權限… 這使得一個RabbitMQ服務眾多的應用程序，而不會互相沖突。 rabbitMQ默認的虛擬主機為: “/” ，一般我們在創建Rabbit的用戶時會再給用戶分配一個虛擬主機。 # 消息投遞策略 默認情況下RabbitMQ的隊列和交換機在RabbitMQ服務器重啟之后會消失，原因在于隊列和交換機的durable屬性，該屬性默認情況下為false. 能從AMQP服務器崩潰中恢復的消息稱為持久化消息，如果想要從崩潰中恢復那么消息必須 * 投遞模式設置2，來標記消息為持久化 * 發送到持久化的交換機 * 到到持久化的隊列 缺點：消息寫入磁盤性能差很多。除非特別關鍵的消息會使