快捷導(dǎo)航

詳解消息隊(duì)列及RabbitMQ部署和使用

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

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

什么是消息隊(duì)列

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

為什么需要消息隊(duì)列

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

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

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

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

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

常見的消息隊(duì)列

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

ActiveMQ

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

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

RabbitMQ

RabbitMQ 實(shí)現(xiàn)了高級消息隊(duì)列協(xié)議（AMQP）的開源消息代理軟件（亦稱面向消息的中間件）。RabbitMQ 服務(wù)器是用高性能、健壯以及可伸縮性出名的 Erlang 語言編寫的，支持所有主流的操作系統(tǒng)如 Linux，Windows，MacOS?？蛻舳酥С炙兄饕木幊陶Z言。

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

ZeroMQ

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

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

Kafka

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

RocketMQ

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

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

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)建一個隊(duì)列，名字叫 my_app

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

點(diǎn)擊 Exchange 標(biāo)簽頁，點(diǎn)擊 my_exchange 進(jìn)入詳情頁，將 my_exchange 和 my_app 綁定，路由密鑰設(shè)置為 test:

Python 編寫生產(chǎn)者

現(xiàn)在可以使用 Python 編寫生產(chǎn)者，來生產(chǎn)一條消息放入隊(duì)列，并觀察 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í)行上面的代碼，即可將消息放入隊(duì)列，這里我執(zhí)行了四次，可以看到有四條消息：

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

Python 編寫消費(fèi)者

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í)行：