詳解消息隊列及RabbitMQ部署和使用

更新時間：2021年09月14日 09:26:48 作者：somenzz

消息隊列是最古老的中間件之一，從系統(tǒng)之間有通信需求開始，就自然產(chǎn)生了消息隊列。本文告訴什么是消息隊列，為什么需要消息隊列，常見的消息隊列有哪些，RabbitMQ的部署和使用

什么是消息隊列

消息隊列拆開了看，就是消息 + 隊列，消息是什么？其實就是程序之間通訊所用到的數(shù)據(jù)，消息從生產(chǎn)者那里產(chǎn)生，進入隊列后，安裝設計好的規(guī)則出隊，由消費者消費。僅此而已。

為什么需要消息隊列

消息隊列，最重要的是隊列，可以想象一下沒有隊列的場景，你去銀行辦業(yè)務的時候，大家都不排隊的場景，大家都堆在一起，個子小沒力氣的根本辦不了業(yè)務。

如果沒有消息隊列，你的系統(tǒng)將嚴重耦合，在升級維護的時候牽一發(fā)而動全身。

如果沒有消息隊列，你的系統(tǒng)的很多功能都是同步的，同步意味著前面的事件完成后，才可以進行后續(xù)的操作，前端用戶的會覺得卡頓，體驗很差。

如果沒有消息隊列，web 系統(tǒng)突然面對高并發(fā)的訪問請求，可能會崩潰。

有了消息隊列，系統(tǒng)解耦、異步通信、流量削峰、延遲通知、最終一致性保證、順序消息、流式處理等需求都可以輕松解決。

常見的消息隊列

比較常見的消息隊列產(chǎn)品主要有 ActiveMQ、RabbitMQ、ZeroMQ、Kafka、RocketMQ 等。

ActiveMQ

Apache ActiveMQ 是 Apache 軟件基金會所研發(fā)的開放源碼消息中間件；由于 ActiveMQ 是一個純Java程序，因此只需要操作系統(tǒng)支持 Java 虛擬機，ActiveMQ 便可執(zhí)行。

支持Java消息服務 (JMS) 1.1 版本
Spring Framework
集群 (Clustering)
協(xié)議支持包括：OpenWire、REST、STOMP、WS-Notification、MQTT、XMPP 以及 AMQP

RabbitMQ

RabbitMQ 實現(xiàn)了高級消息隊列協(xié)議（AMQP）的開源消息代理軟件（亦稱面向消息的中間件）。RabbitMQ 服務器是用高性能、健壯以及可伸縮性出名的 Erlang 語言編寫的，支持所有主流的操作系統(tǒng)如 Linux，Windows，MacOS。客戶端支持所有主要的編程語言。

可伸縮性：集群服務
消息持久化：從內(nèi)存持久化消息到硬盤，再從硬盤加載到內(nèi)存

ZeroMQ

ZeroMQ（也拼寫作 0MQ 或 ZMQ )是一個為可伸縮的分布式或并發(fā)應用程序設計的高性能異步消息庫。它提供一個消息隊列, 但是與面向消息的中間件不同，ZeroMQ 的運行不需要專門的消息代理（message broker）。該庫設計成常見的套接字風格的API。

ZeroMQ 是由 iMatix 公司和大量貢獻者組成的社群共同開發(fā)的。ZeroQ 通過許多第三方軟件支持大部分流行的編程語言，從 Java 和Python 到 Erlang 和 Haskell。

Kafka

Kafka 是由 Apache 軟件基金會開發(fā)的一個開源流處理平臺，由 Scala 和 Java 編寫。該項目的目標是為處理實時數(shù)據(jù)提供一個統(tǒng)一、高吞吐、低延遲的平臺。其持久化層本質(zhì)上是一個“按照分布式事務日志架構(gòu)的大規(guī)模發(fā)布/訂閱消息隊列”，這使它作為企業(yè)級基礎設施來處理流式數(shù)據(jù)非常有價值。此外，Kafka 可以通過 Kafka Connect 連接到外部系統(tǒng)（用于數(shù)據(jù)輸入/輸出），并提供了 Kafka Streams 的流式處理庫。該設計受事務日志的影響較大。

RocketMQ

RocketMQ 是一個分布式消息和流數(shù)據(jù)平臺，具有低延遲、高性能、高可靠性、萬億級容量和靈活的可擴展性。RocketMQ 是 2012 年阿里巴巴開源的第三代分布式消息中間件，2016 年 11 月 21 日，阿里巴巴向 Apache 軟件基金會捐贈了 RocketMQ；第二年 2 月 20 日，Apache 軟件基金會宣布 Apache RocketMQ 成為頂級項目。

消息隊列的選型需要根據(jù)具體應用需求而定，ZeroMQ 小而美，RabbitMQ 大而穩(wěn)，Kakfa 和 RocketMQ 快而強勁。

RabbitMQ 的部署和使用

推薦 Docker 部署，在安裝 Docker 的環(huán)境下，執(zhí)行：

docker run -d --hostname my-rabbit -p 15672:15672 -p 5672:5672 -- name rabbit-server -e RABBITMQ_DEFAULT_USER=user -e RABBITMQ_DEFAULT_PASS=password rabbitmq:3-management

現(xiàn)在在瀏覽器中打開 localhost:15672 。用戶名是 user ，密碼是 password

接下來，讓我們創(chuàng)建一個隊列，名字叫 my_app

創(chuàng)建一個 Exchange，名字叫 my_exchange

點擊 Exchange 標簽頁，點擊 my_exchange 進入詳情頁，將 my_exchange 和 my_app 綁定，路由密鑰設置為 test:

Python 編寫生產(chǎn)者

現(xiàn)在可以使用 Python 編寫生產(chǎn)者，來生產(chǎn)一條消息放入隊列，并觀察 Web 頁面的變動情況：

import pika
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost', 5672, '/', pika.PlainCredentials('user', 'password')))
channel = connection.channel() 
channel.basic_publish(exchange='my_exchange', routing_key='test', body='Test!')
connection.close()

執(zhí)行上面的代碼，即可將消息放入隊列，這里我執(zhí)行了四次，可以看到有四條消息：

消息將保留在隊列中，直到消費者把它取出，接下來我們寫一個消費消息的程序。

Python 編寫消費者

import pika
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost', 5672, '/', pika.PlainCredentials("user", "password")))
channel = connection.channel()
def callback(ch, method, properties, body):
    print(f'{body} is received')
channel.basic_consume(queue="my_app", on_message_callback=callback, auto_ack=True)
channel.start_consuming()

執(zhí)行：