1. 淺淺道來

1.1 什么是中間件？

IDC（互聯(lián)網(wǎng)數(shù)據(jù)中心）的定義：中間件是一種獨(dú)立的系統(tǒng)軟件服務(wù)程序，分布式應(yīng)用軟件借助這種軟件在不同的技術(shù)之間共享資源，中間件位于客戶機(jī)服務(wù)器的操作系統(tǒng)之上，管理計(jì)算資源和網(wǎng)絡(luò)通信。

首先，中間件是某一類軟件的總稱，而不是某一種具體的軟件。它是一種位于平臺(tái)（操作系統(tǒng)硬件） 和 應(yīng)用程序之間的通用服務(wù)，它屏蔽了底層操作系統(tǒng)的各種復(fù)雜性，減輕了開發(fā)人員的技術(shù)負(fù)擔(dān)，同時(shí)它的設(shè)計(jì)不針對(duì)某一具體目標(biāo)，而是提供具有普遍通用特點(diǎn)的功能模塊服務(wù)，這些服務(wù)具有標(biāo)準(zhǔn)的程序接口和協(xié)議，根據(jù)平臺(tái)的不同，也可以有不同的實(shí)現(xiàn)。

通俗的例子（僅供參考，并不算完全一致）：

我開了一家咖啡店，我身邊有 A B C 等 n 家咖啡豆的供應(yīng)商，但是我肯定要挑選價(jià)格又實(shí)惠，質(zhì)量還不錯(cuò)的豆子，但是市場(chǎng)是受到多方面因素波動(dòng)的，可能我現(xiàn)在的選擇，在一段時(shí)間后已經(jīng)不是最佳選項(xiàng)了。所以我專門找到一家市場(chǎng)中介，讓他幫我操心這一攤子事情，我只和你說清價(jià)格和質(zhì)量要求，你去找就是了，過程我一點(diǎn)也不操心。這個(gè)中介的概念，就類似中間件的

1.1.1 分布式的概念（補(bǔ)充）

這一段，來自我之前寫的 Dubbo 入門的那篇文章哈

在百度以及維基中的定義都相對(duì)專業(yè)且晦澀，大部分博客或者教程經(jīng)常會(huì)使用《分布式系統(tǒng)原理和范型》中的定義，即：“分布式系統(tǒng)是若干獨(dú)立計(jì)算機(jī)的集合，這些計(jì)算機(jī)對(duì)于用戶來說就像是單個(gè)相關(guān)系統(tǒng)”

下面我們用一些篇幅來通俗的解釋一下什么叫做分布式

1.1.1.1 什么是集中式系統(tǒng)

提到分布式，不得不提的就是 “集中式系統(tǒng)”，這個(gè)概念最好理解了，它就是將功能，程序等安裝在同一臺(tái)設(shè)備上，就由這一臺(tái)主機(jī)設(shè)備向外提供服務(wù)

舉個(gè)最簡(jiǎn)單的例子：你拿一臺(tái)PC主機(jī)，將其改裝成了一臺(tái)簡(jiǎn)單的服務(wù)器，配置好各種內(nèi)容后，你將MySQL，Web服務(wù)器，F(xiàn)TP，Nginx 等等，全部安裝在其中，打包部署項(xiàng)目后，就可以對(duì)外提供服務(wù)了，但是一旦這臺(tái)機(jī)器無論是軟件還是硬件出現(xiàn)了問題，整個(gè)系統(tǒng)都會(huì)受到嚴(yán)重的牽連錯(cuò)誤，雞蛋放在一個(gè)籃子里，要打就全打了

1.1.12 什么是分布式系統(tǒng)

既然集中式系統(tǒng)有這樣一種牽一發(fā)而動(dòng)全身的問題，那么分布式的其中一個(gè)作用，自然是來解決這樣的問題了，正如定義中所知，分布式系統(tǒng)在用戶的體驗(yàn)感官里，就像傳統(tǒng)的單系統(tǒng)一樣，一些變化都是這個(gè)系統(tǒng)本身內(nèi)部進(jìn)行的，對(duì)于用戶并沒有什么太大的感覺

例如：淘寶，京東這種大型電商平臺(tái)，它們的主機(jī)都是數(shù)以萬計(jì)的，否則根本沒法處理大量的數(shù)據(jù)和請(qǐng)求，具體其中有什么劃分，以及操作，我們下面會(huì)說到，但是對(duì)于用戶的我們，我們不需要也不想關(guān)心這些，我們?nèi)钥梢詥渭兊恼J(rèn)為，我們面對(duì)的就是 “淘寶” 這一臺(tái) “主機(jī)”

所以分布式的一個(gè)相對(duì)專業(yè)一些的說法是這樣的（進(jìn)程粒度）兩個(gè)或者多個(gè)程序，分別運(yùn)行在不同的主機(jī)進(jìn)程上，它們互相配合協(xié)調(diào)，完成共同的功能，那么這幾個(gè)程序之間構(gòu)成的系統(tǒng)就可以叫做分布式系統(tǒng)

這幾者都是相同的程序 —— 分布式這幾者都是不同的程序 —— 集群

1.2 什么是消息中間件/消息隊(duì)列（MQ）

消息中間件，顧名思義就是用來處理消息相關(guān)服務(wù)的中間件，它提供了一種系統(tǒng)之間通信交互的通道，例如發(fā)送方只需要把想傳輸?shù)男畔⒔唤o消息中間件，而發(fā)送的協(xié)議，方式，發(fā)送過程中出現(xiàn)的網(wǎng)絡(luò)，故障等等問題，都由中間件進(jìn)行處理，因此它負(fù)責(zé)保證信息的可靠傳輸。

所以消息中間件，就是一種用來接受數(shù)據(jù)，存儲(chǔ)數(shù)據(jù)，發(fā)送數(shù)據(jù)的技術(shù)，它提供了各種功能，可以實(shí)現(xiàn)消息的高可用，高可靠，也提供了很好的容錯(cuò)機(jī)制等?？梢猿绦?qū)ο到y(tǒng)資源的占用，以及傳輸效率的提升有很大幫助。

常說的 MQ 就是指消息隊(duì)列，即 Message Quene，常見的消息隊(duì)列有，經(jīng)典的 ActivieMQ，熱門的 Kafka，阿里的 RocketMQ 等等，以及這里講解的 RabbitMQ。

不同的 MQ 有著不同的特點(diǎn)，以及其更加擅長(zhǎng)的方向，倒也說不上誰好誰壞，只有誰更合適。

1.2.1 消息隊(duì)列應(yīng)用場(chǎng)景

根據(jù)業(yè)務(wù)的需要，其實(shí)它可以有多種應(yīng)用場(chǎng)景，例如解耦，削峰填谷，廣播等，我們舉兩個(gè)場(chǎng)景來梳理一下簡(jiǎn)單的過程

1.2.1.1 業(yè)務(wù)解耦

最近在考慮買幾本書看，就以買書下訂單舉例，當(dāng)我點(diǎn)擊購(gòu)買之后，可能會(huì)有這么一串業(yè)務(wù)邏輯執(zhí)行，① 減去庫存容量 ② 生成訂單 ③ 支付 ④ 更新訂單狀態(tài) ⑤ 發(fā)送購(gòu)買成功短信 ⑥ 更新商品快遞攬收狀態(tài)。在初期階段，我們完全可以讓這些業(yè)務(wù)同步執(zhí)行，但是后期為了提升效率，就可以將需要立即執(zhí)行的任務(wù)和可稍緩執(zhí)行的任務(wù)進(jìn)行分離，例如 ⑤ 發(fā)送購(gòu)買成功短信 ⑥ 更新商品快遞攬收狀態(tài)，都可以考慮異執(zhí)行。在主流程執(zhí)行結(jié)束后，這些可稍緩的業(yè)務(wù)可以通過給 MQ 發(fā)送消息，就判定已經(jīng)執(zhí)行，保證流程先結(jié)束。然后再通過拉取 MQ 消息，或者 MQ 主動(dòng)推送去異步執(zhí)行其他的業(yè)務(wù)。

1.2.1.2 削峰填谷

例如發(fā)送一條帶有已讀未讀標(biāo)識(shí)的公告信息，所以需要對(duì)每一個(gè)用戶都寫一條這樣的公告消息，例如存到 MongoDB 中，即便 MongoDB 也支撐不下來瞬時(shí)寫入百萬、千萬記錄的情況，所以可以考慮使用消息隊(duì)列。比如說我們可以在Java后端系統(tǒng)上面，用異步多線程的方法，向消息隊(duì)列MQ中發(fā)送消息，這樣Web系統(tǒng)發(fā)布公告消息的時(shí)候就不占用數(shù)據(jù)庫正常的 CRUD 操作。系統(tǒng)消息保存在消息隊(duì)列中，我們只是用它來做削峰填谷，系統(tǒng)消息最終還是要存儲(chǔ)在數(shù)據(jù)庫上面。于是我們可以這樣設(shè)計(jì)，在用戶登陸系統(tǒng)的時(shí)候，用異步線程從消息隊(duì)列MQ中，接收該用戶的系統(tǒng)消息，然后把系統(tǒng)消息存儲(chǔ)在數(shù)據(jù)庫中，最后消息隊(duì)列MQ中的該條消息自動(dòng)刪除。因?yàn)橛脩舻腻e(cuò)峰登錄，所以往數(shù)據(jù)庫中寫入消息的任務(wù)也變成了錯(cuò)峰寫入。

1.3 什么是 RabbitMQ

RabbitMQ 是一個(gè)使用 Erlang 語言編寫，且遵循 AMQP協(xié)議的開源消息隊(duì)列系統(tǒng)，支持多種客戶端（語言），用于在分布式系統(tǒng)中存儲(chǔ)消息，轉(zhuǎn)發(fā)消息，具有高可用，高可擴(kuò)性，易用性等特征。

更詳細(xì)的介紹可以直接看一下官網(wǎng)：https://www.rabbitmq.com/

總之這就是一種常見的消息隊(duì)列，它的這些特點(diǎn)，都會(huì)在后面逐條講解到，我們首先從入門下載安裝部分先說起，然后再到使用。

2. 下載與安裝

一般來說，安裝的方式有手動(dòng)安裝和 Docker 安裝，大部分場(chǎng)景下，都會(huì)使用 Docker 安裝，但是作為學(xué)習(xí)階段，如果不是特別著急，學(xué)習(xí)一下手動(dòng)安裝，也不是什么壞事。

注：云服務(wù)器和虛擬機(jī)都可以，演示的 Linux 版本為 CentOS 7.9

2.1 手動(dòng)安裝

2.1.1 下載安裝過程

注：可以在 Linux 中通過 yum 直接下載安裝，這里選擇了在自己的 Windows 主機(jī)先下載文件，然后再通過 FTP 傳到 Linux 上，直接安裝?？梢员苊馓摂M機(jī)上因?yàn)榫W(wǎng)絡(luò)而造成的一些下載問題。

首先打開官網(wǎng)的下載目錄，然后根據(jù)自己 Linux 的版本，選擇版本。

1.地址：https://www.rabbitmq.com/download.html

2.因?yàn)?RabbitMQ 是 Erlang 語言編寫的，所以還需要提供 Erlang 環(huán)境，接著去下載 Erlang。

• 地址：https://www.erlang-solutions.com/downloads
  • A：此網(wǎng)站訪問速度極慢，請(qǐng)耐心等待，或者需要掛上梯子
  • B：Erlang 版本需要與 RabbitMQ 匹配（如圖，有最大和最小版本的限制）
    • 版本查看地址：https://www.rabbitmq.com/which-erlang.html
    • 這里選擇了 RabbitMQ 3.8.14 和 Erlang 23.2.3

3.將文件上傳到 Linux 中（我這里指定位置是 /usr/local/bin/rabbitmq ，可以自己更改選擇）

• 現(xiàn)在很多 Shell 軟件都自帶內(nèi)置的 FTP 上傳，例如 FinalShell，MobaXterm 等等
• 上傳后的文件和目錄位置如下

[root@centos7 rabbitmq]# ls
esl-erlang_23.2.3-1_centos_7_amd64.rpm  rabbitmq-server-3.8.14-1.el7.noarch.rpm
[root@centos7 rabbitmq]# pwd
/usr/local/bin/rabbitmq

4.安裝 Erlang 、Socat 和 RabbitMQ

• Erlang 、Socat 都是 RabbitMQ 所依賴的

# 安裝 Erlang，安裝后執(zhí)行 erl -v 顯示版本號(hào)則代表成功
rpm -ivh esl-erlang_23.2.3-1_centos_7_amd64.rpm
# 安裝 Socat 這里沒有下載源文件，而是直接通過 yum 在線安裝，因?yàn)樗⒉淮?
yum install -y socat
# 安裝 RabbitMQ
rpm -ivh rabbitmq-server-3.8.14-1.el7.noarch.rpm

5.安裝結(jié)束，啟動(dòng)服務(wù)查看 RabbitMQ 是否可以啟動(dòng)成功

# 啟動(dòng)服務(wù)
systemctl start rabbitmq-server
# 開機(jī)自啟
systemctl enable rabbitmq-server
# 停止服務(wù)
systemctl stop rabbitmq-server
# 查看服務(wù)狀態(tài)
systemctl status rabbitmq-server.service

如圖所示，即安裝啟動(dòng)成功

2.1.2 配置 Web 界面管理

上面的安裝其實(shí)已經(jīng)結(jié)束了，但是 RabbitMQ 提供給了我們一個(gè) Web 形式的管理界面，默認(rèn)是沒有的，需要進(jìn)行安裝。

1.安裝 Web 管理插件，然后重啟服務(wù)

# 安裝命令
rabbitmq-plugins enable rabbitmq_management
# 重啟服務(wù)
systemctl restart rabbitmq-server

2.一定要開放 Linux 防火墻 的 15672 端口，否則就會(huì)無法訪問，在學(xué)習(xí)階段，你甚至可以去查詢命令把防火墻關(guān)掉

對(duì)應(yīng)服務(wù)器（阿里云，騰訊云等）就是在安全組中開放 15672 端口

訪問 Linux IP:15672 ，例如 http://192.168.122.1:15672

# 查詢 15672 是否開放，一般默認(rèn)都是 no
firewall-cmd --query-port=15672/tcp
# 開放指定端口 15672 
firewall-cmd --add-port=15672/tcp --permanent
# 重新載入
firewall-cmd --reload
# 再次查詢，結(jié)果就是 yes 了
firewall-cmd --query-port=15672/tcp

3.添加遠(yuǎn)程登錄的賬戶

• RabbitMQ 有一個(gè)默認(rèn)賬號(hào)和密碼都是 guest 但是只能在 localhost 下訪問

# 新增用戶 用戶名和密碼都是 admin
rabbitmqctl add_user admin admin

4.為遠(yuǎn)程登錄的賬戶添加權(quán)限

• administrator（超級(jí)管理員）：登錄控制臺(tái)、查看所有信息、操作用戶、操作策略
• monitoring（監(jiān)控者）： 登錄控制臺(tái)、查看所有信息
• policymaker（策略制定者）： 登錄控制臺(tái)、指定策略
• managment（普通管理員）： 登錄控制臺(tái)
  復(fù)制代碼 代碼如下:

# 設(shè)置用戶分配操作權(quán)限，admin 用戶的權(quán)限為 administrator
rabbitmqctl set_user_tags admin administrator

5.為用戶添加資源權(quán)限

• 因?yàn)?admin 已經(jīng)是超級(jí)管理員權(quán)限了，所以其實(shí)不分配資源權(quán)限也可以，會(huì)默認(rèn)去做。

# 命令格式為: set_permissions [-p <vhostpath>] <user> <conf> <write> <read>
# 這里即為 admin 用戶開啟 配置文件和讀寫的權(quán)限
rabbitmqctl set_permissions -p / admin ".*"".*"".*"

6.訪問 Linux IP:15672 ，例如 http://192.168.122.1:15672 ，輸入剛才設(shè)置好的用戶名密碼 admin

• 如圖：訪問成功

2.1.2.1 命令小結(jié)

1.添加用戶：rabbitmqctl add_user <username> <password>

2.修改密碼：rabbitmqctl change_password <username> <newpass>

3.刪除用戶：rabbitmqctl delete_user <username>

4.用戶列表：rabbitmqctl list_users

5.設(shè)置用戶角色：rabbitmqctl set_user_tags <username> <tag1,tag2>

6.刪除用戶所有角色：rabbitmqctl set_user_tags <username>

7.為用戶添加資源權(quán)限：set_permissions [-p <vhostpath>] <user> <conf> <write> <read>

使用：輸入 rabbitmqctl ，則會(huì)提示可能使用的命令，然后 使用 rabbitmqctl hepl <命令> 可以查看具體命令的使用方法和參數(shù)。

2.1.3 簡(jiǎn)單介紹 Web 界面管理

• Connections（連接）：此處用來管理與
• RabbitMQ 建立連接后的生產(chǎn)者和消費(fèi)者
• Channels（通道）：連接建立后，會(huì)形成通道，消息的投遞獲取依賴通道。
• Exchanges（交換機(jī)）：用來實(shí)現(xiàn)消息的路由
• Queues（隊(duì)列）：存放消息的隊(duì)列，消息等待被消費(fèi)，消費(fèi)后被移除隊(duì)列。
• Admin（管理）：用于對(duì)管理用戶，以及對(duì)應(yīng)權(quán)限進(jìn)行設(shè)置，如下圖所示

Tags 就是用來指定用戶的角色

• administrator（超級(jí)管理員）：登錄控制臺(tái)、查看所有信息、操作用戶、操作策略
• monitoring（監(jiān)控者）： 登錄控制臺(tái)、查看所有信息
• policymaker（策略制定者）： 登錄控制臺(tái)、指定策略
• managment（普通管理員）： 登錄控制臺(tái)

2.2 Docker 安裝

在 Docker 中安裝 RabbitMQ 不需要自己去考慮版本，環(huán)境等的各種沖突不兼容問題，是非常便捷的，我演示的這臺(tái)虛擬機(jī)是一個(gè) CentOS 7.9 裸機(jī)，所以我們從更新 yum，到安裝 Docker 和 安裝 RabbitMQ 按步驟都講一下

2.2.1 配置 yum

1.更新 yum 到最新版

# 更新 yum
yum update
# 檢查yum依賴的幾個(gè)包 yum-utils 提供 yum-config-manager 功能， 后面兩個(gè)是 devicemapper 用到的
yum install -y yum-utils device-mapper-persistent-data lvm2

2.設(shè)置 yum 源為阿里云

yum-config-manager --add-repo http://mirrors.aliyun.com/docker-ce/linux/centos/docker-ce.repo

2.2.2 安裝 docker

2.2.2.1 步驟

1.使用 yum 安裝 docker

• docker-ce 是社區(qū)版的意思，ee為企業(yè)版

yum install docker-ce -y

2.通過查看版本，檢查安裝是否成功

docker -v

sudo mkdir -p /etc/docker
sudo tee /etc/docker/daemon.json <<-'EOF'
{
  "registry-mirrors": ["https://<你的ID>.mirror.aliyuncs.com"]
}
EOF
sudo systemctl daemon-reload
sudo systemctl restart docker

• 國(guó)內(nèi)從 DockerHub 拉取鏡像有時(shí)會(huì)遇到困難，此時(shí)可以配置鏡像加速器。Docker 官方和國(guó)內(nèi)很多云服務(wù)商都提供了國(guó)內(nèi)加速器服務(wù)，例如：
  • 科大鏡像：https://docker.mirrors.ustc.edu.cn/
  • 網(wǎng)易：https://hub-mirror.c.163.com/
  • 阿里云：https://<你的ID>.mirror.aliyuncs.com
  • 七牛云加速器：https://reg-mirror.qiniu.com

當(dāng)配置某一個(gè)加速器地址之后，若發(fā)現(xiàn)拉取不到鏡像，請(qǐng)切換到另一個(gè)加速器地址。國(guó)內(nèi)各大云服務(wù)商均提供了 Docker 鏡像加速服務(wù)，建議根據(jù)運(yùn)行 Docker 的云平臺(tái)選擇對(duì)應(yīng)的鏡像加速服務(wù)。

阿里云鏡像獲取地址：https://cr.console.aliyun.com/cn-hangzhou/instances/mirrors，登陸后，左側(cè)菜單選中鏡像加速器就可以看到你的專屬地址了

2.2.2.2 Docker 常見命令

2.2.2.2.1 管理命令

• 就啟動(dòng)，停止，重啟這些簡(jiǎn)單的命令使用 service 也是可以的，systemctl 功能稍微強(qiáng)大一些

# 啟動(dòng) docker
systemctl docker start
# 停止 docker
systemctl docker stop
# 重啟 docker
systemctl docker restart
# 查看 docker 狀態(tài)
systemctl status docker
# 開機(jī)自啟
systemctl enable docker
systemctl unenable docker

2.2.2.2.2 鏡像命令

# 導(dǎo)入鏡像文件
docker load < xxx.tar.gz
# 查看安裝的鏡像
docker images
# 刪除鏡像
docker rmi 鏡像名

2.2.3 安裝 RabbitMQ （任選其一）

注：直接用 2.2.3.2 一句話安裝 會(huì)更好一些

2.2.3.1 一步一步安裝獲取

1.RabbitMQ 的鏡像

docker pull rabbitmq:management

2.創(chuàng)建并運(yùn)行容器（具體參數(shù)在 3 中介紹）

docker run -id --name 容器名 -p 15672:15672 -p 5672:5672 rabbitmq:management

2.2.3.2 一句話安裝

上面的安裝方式，就是先獲取到 RabbitMQ 鏡像后再開始安裝，這里是沒有問題的，創(chuàng)建時(shí)會(huì)有一個(gè)問題，因?yàn)槲覀円惭b management 也就是它的 web 管理，如果不做一些處理，默認(rèn)裝好的是沒有用戶的，所以還需要像前面一樣自己進(jìn)去配置，而 Docker Hub 已經(jīng)給出了我們配置的示例，即使用 -e 代表配置，使用 RABBITMQ_DEFAULT_USER 和 RABBITMQ_DEFAULT_PASS 配置用戶名和密碼

更多請(qǐng)查看 Docker Hub 官方給予例子中的 Setting default user and password 章節(jié)https://registry.hub.docker.com/_/rabbitmq/

1.執(zhí)行安裝

docker run -di --name myrabbitmq -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin -p 15672:15672 -p
 5672:5672 -p 25672:25672 -p 61613:61613 -p 1883:1883 rabbitmq:management

2.通過容器狀態(tài)，查看是否運(yùn)行成功

# 查看容器運(yùn)行狀態(tài)docker ps -a# 啟動(dòng)docker start 容器名# 停止docker stop 容器名# 退出命令行，不停止exit# 進(jìn)入到node容器（如果開啟了 -t 的情況）docker exec -it 容器名 bash

2.2.3.2.1 參數(shù)介紹

下面分別講解一下這些參數(shù)的說明：

• -i：表示運(yùn)行容器。
• -t：表示為容器保留交互的方式（命令行），即分配一個(gè)偽終端。所以常常會(huì)見到 -it 這樣的搭配。
• --name ：為容器起個(gè)名字。
• -v：表示目錄映射關(guān)系（前者是宿主機(jī)目錄，后者是映射到宿主機(jī)上的目錄），可以使用多個(gè) -v 做多個(gè)目錄或文件映射。注意：推薦做目錄映射，在宿主機(jī)上做修改，然后共享到容器上。
• -d：表示創(chuàng)建一個(gè)守護(hù)式容器在后臺(tái)運(yùn)行（這樣創(chuàng)建容器后不會(huì)自動(dòng)登錄容器，如果只加 -i -t 兩個(gè)參數(shù)，創(chuàng)建后就會(huì)自動(dòng)進(jìn)去容器），即后端掛起運(yùn)行。
• -p：表示端口映射，前者是宿主機(jī)端口，后者是容器內(nèi)的映射端口。可以使用多個(gè) -p 做多個(gè)端口映射，只有做了端口映射，才能被外界訪問。

給大家舉個(gè)例子：

# 查看容器運(yùn)行狀態(tài)
docker ps -a
# 啟動(dòng)
docker start 容器名
# 停止
docker stop 容器名
# 退出命令行，不停止
exit
# 進(jìn)入到node容器（如果開啟了 -t 的情況）
docker exec -it 容器名 bash

因?yàn)槭褂昧?-t 這個(gè)參數(shù)，所以可以分配到一個(gè)偽終端，通過 docker exec -it 容器名 bash 進(jìn)入命令行
-v 目錄映射后，進(jìn)入容器后，也會(huì)有一個(gè)一模一樣的 demo 文件夾，例如在其中可以執(zhí)行 python 程序

2.2.3.2.1 端口介紹

4369 ：erlang發(fā)現(xiàn)端口

5672：client端通信端口

15672：管理界面ui端口

25672：server間內(nèi)部通信端口

61613：不帶TLS和帶TLS的STOMP客戶端

1883：不啟用和啟用TLS的MQTT客戶端

比較關(guān)鍵的就是 5672 和 15672

更多端口詳情可以訪問官網(wǎng)文檔https://www.rabbitmq.com/networking.html

注：如果要通過遠(yuǎn)程連接，例如訪問 web 管理頁面的 15672 端口，Java 客戶端連接的 5672 端口， 一定要進(jìn)行一個(gè)開放操作，否則都連接不到。

• 以下為基于 CentOS 7.9 開放 15672 端口的例子

# 查詢 15672 是否開放，一般默認(rèn)都是 no
firewall-cmd --query-port=15672/tcp
# 開放指定端口 15672 
firewall-cmd --add-port=15672/tcp --permanent
# 重新載入
firewall-cmd --reload
# 再次查詢，結(jié)果就是 yes 了
firewall-cmd --query-port=15672/tcp

]以下是關(guān)閉防火墻的命令

systemctl disable firewalld
systemctl stop firewalld   

3. RabbitMQ 協(xié)議和模型

安裝結(jié)束后，就要進(jìn)入主題，即用 Java 或者 Springboot 代碼來實(shí)現(xiàn) RabbitMQ的幾種方式，但是想要很好的理解這幾種路由交換方式，就需要對(duì)它的協(xié)議和架構(gòu)模型有所了解。

3.1 協(xié)議

3.1.1 什么是協(xié)議？

協(xié)議，網(wǎng)絡(luò)協(xié)議的簡(jiǎn)稱，網(wǎng)絡(luò)協(xié)議是通信計(jì)算機(jī)雙方必須共同遵從的一組約定。如怎么樣建立連接、怎么樣互相識(shí)別等。只有遵守這個(gè)約定，計(jì)算機(jī)之間才能相互通信交流。它的三要素是：語法、語義、時(shí)序。

為了使數(shù)據(jù)在網(wǎng)絡(luò)上從源到達(dá)目的，網(wǎng)絡(luò)通信的參與方必須遵循相同的規(guī)則，這套規(guī)則稱為協(xié)議（protocol），它最終體現(xiàn)為在網(wǎng)絡(luò)上傳輸?shù)臄?shù)據(jù)包的格式。

3.1.1.1 網(wǎng)絡(luò)協(xié)議的三要素

1.語法：數(shù)據(jù)與控制信息的結(jié)構(gòu)和格式，以及數(shù)據(jù)出現(xiàn)的順序。

2.語義：解釋控制信息每個(gè)部分的意義，以及規(guī)定了需要發(fā)出何種控制信息以及完成的動(dòng)作做出何種響應(yīng)。

3.時(shí)序：對(duì)事件發(fā)生順序的詳細(xì)說明。

人們形象地把這三個(gè)要素描述為：做什么，怎么做，做的順序。

舉個(gè)例子 HTTP 協(xié)議

語法：HTTP 規(guī)定了請(qǐng)求報(bào)文和響應(yīng)報(bào)文的格式
語義：客戶端主動(dòng)發(fā)起請(qǐng)求稱為請(qǐng)求，服務(wù)端隨之返回?cái)?shù)據(jù)，稱為響應(yīng)
時(shí)序： 一個(gè)請(qǐng)求對(duì)應(yīng)一個(gè)響應(yīng)，而且先有請(qǐng)求后有響應(yīng)

3.1.1.1.1 面試題：為什么消息中間件不直接使用 HTTP 協(xié)議

對(duì)于一個(gè)消息中間件來說，其主要責(zé)任就是負(fù)責(zé)數(shù)據(jù)傳遞，存儲(chǔ)，分發(fā)，高性能和簡(jiǎn)潔才是我們所追求的，而 HTTP 請(qǐng)求報(bào)文頭和響應(yīng)報(bào)文頭是比較復(fù)雜的，包含了Cookie，數(shù)據(jù)的加密解密，窗臺(tái)嗎，響應(yīng)碼等附加的功能，我們并不需要這么復(fù)雜的功能。

同時(shí)大部分情況下 HTTP 大部分都是短鏈接，在實(shí)際的交互過程中，一個(gè)請(qǐng)求到響應(yīng)都很有可能會(huì)中斷，中斷以后就不會(huì)執(zhí)行持久化，就會(huì)造成請(qǐng)求的丟失。這樣就不利于消息中間件的業(yè)務(wù)場(chǎng)景，因?yàn)橄⒅虚g件可能是一個(gè)長(zhǎng)期的獲取信息的過程，出現(xiàn)問題和故障要對(duì)數(shù)據(jù)或消息執(zhí)行持久化等，目的是為了保證消息和數(shù)據(jù)的高可靠和穩(wěn)健的運(yùn)行

3.1.2 RabbitMQ 的 AMQP 協(xié)議

RabbitMQ 的使用的協(xié)議是 AMQP（advanced message queuing protocol），它在2003年時(shí)被提出，最早用于解決金融領(lǐng)不同平臺(tái)之間的消息傳遞交互問題。

AMQP 更準(zhǔn)確的說是一種 binary wire-level protocol（鏈接協(xié)議）。這是其和 JMS 的本質(zhì)差別，AMQP 不從 API 層進(jìn)行限定，而是直接定義網(wǎng)絡(luò)交換的數(shù)據(jù)格式。這使得實(shí)現(xiàn)了AMQP的 Provider（Producer） 天然性就是跨平臺(tái)的。

相比較其它消息協(xié)議，其特性為：

1.分布式事務(wù)支持

2.消息的持久化支持

3.高性能和高可靠的消息處理優(yōu)勢(shì)

3.1.3 架構(gòu)模型

想要學(xué)習(xí)后面的幾種消息具體的發(fā)送模式，這個(gè)模型圖就必須理解清楚，因?yàn)檫@幾種方式就是對(duì)這個(gè)模型不同程度的選擇和縮減

• Producer：消息的生產(chǎn)者（發(fā)送消息的程序）。
• Connection：應(yīng)用程序與Broker之間的網(wǎng)絡(luò)連接。
• Channel：信道，即信息傳輸?shù)耐ǖ?，可以建立多個(gè) Channel，每個(gè) Channel 代表一個(gè)會(huì)話任務(wù)。
  • 信道是建立在 TCP 連接內(nèi)的虛擬連接，信息的讀寫都通過信道傳輸，因?yàn)閷?duì)于操縱系統(tǒng)而言，建立和銷毀 TCP 是非常昂貴的，所以引入了信道的概念，以復(fù)用一條 TCP 連接。
• Broker(Server) ：標(biāo)識(shí)消息隊(duì)列服務(wù)器實(shí)體，例如這里就是 RabbitMQ Server。
• Virtual Host：虛擬主機(jī)，一個(gè) Broker 中可以設(shè)置多個(gè) Virtual Host，用作不同用戶的權(quán)限隔離。
  • Broker 可以理解為整個(gè)數(shù)據(jù)庫服務(wù)，而 Virtual Host 就是其中每個(gè)數(shù)據(jù)庫的感覺，不同項(xiàng)目可以對(duì)應(yīng)不同的數(shù)據(jù)庫，其中有著項(xiàng)目所屬的業(yè)務(wù)表等等。
  • 每個(gè) Virtual Host 中，可以有若干個(gè) Exchange 和 Queue。
• Exchange：交換機(jī)，用來接收生產(chǎn)者發(fā)送的消息，然后將這些消息根據(jù)路由鍵發(fā)送到隊(duì)列。
• Binding：Exchange 和 Queue 之間的虛擬連接，Binding 中可以包括多個(gè) Routing key。
• Routing key：路由規(guī)則，虛擬機(jī)用它來確認(rèn)如何路由一個(gè)特定消息。
• Queue：消息隊(duì)列，它是消息的容器，用來保存消息，每一條消息都能傳入一個(gè)或者多個(gè)隊(duì)列中，等待消費(fèi)者消費(fèi)，即取出這個(gè)消息。
• Consumer：消息的消費(fèi)者（接收消息的程序）。

4. Java 實(shí)現(xiàn) RabbitMQ

4.1 環(huán)境搭建

官網(wǎng)介紹幾種模型：https://www.rabbitmq.com/getstarted.html

截止目前為止，官網(wǎng)一共提供了 7 中模型的介紹，我們主要介紹前五種基本的模式，也有人將 Direct 和 Topic模式都?xì)w入 Routing 模式，也可以看做四大種。

4.1.1 創(chuàng)建 Java 項(xiàng)目

首先創(chuàng)建好一個(gè)不使用骨架的 Maven 項(xiàng)目，然后引入 RabbitMQ 依賴，還有單元測(cè)試依賴即可

<dependency>
   <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.10.0</version>
</dependency>
<dependency>
    <groupId>junit</groupId>
    <artifactId>junit</artifactId>
    <version>4.11</version>
</dependency>

4.1.2 創(chuàng)建虛擬主機(jī)（可選）

在這里，我們創(chuàng)建了一個(gè)新的 Virtual Hosts，用來為這個(gè)Java項(xiàng)目服務(wù)，大家還可以創(chuàng)建一個(gè)新的用戶，然后對(duì)其開啟這個(gè) Virtual Hosts 的訪問權(quán)限（即將虛擬主機(jī)與用戶綁定）。我們這里還是用 admin（我之前創(chuàng)建的一個(gè)管理員權(quán)限用戶） 來演示。

注：這部分不去做也可以，直接用 / 和 admin 用戶也行

4.1.3 創(chuàng)建連接工具類

由于我們后面要演示多種例子，而每一次獲取連接和釋放連接、關(guān)閉資源等操作代碼都是一致的，為了防止代碼冗余，優(yōu)化代碼，更易理解，提取出一個(gè)工具類，這樣大家將重心放在不同實(shí)現(xiàn)方式的對(duì)比上就行了。

• RabbitMqUtil 工具類

public class RabbitMqUtil {
    /**
     * 主機(jī)名 即 Linux IP地址
     */
    private static String host = "";
    /**
     *  端口號(hào) 客戶端訪問默認(rèn)都是 5672
     */
    private static int port = 0;
    /**
     * 虛擬主機(jī) 可以設(shè)置為默認(rèn)的 / 或者自己創(chuàng)建出指定的虛擬主機(jī)
     */
    private static String virtualHost = "";
    /**
     * 用戶名
     */
    private static String username = "";
    /**
     * 密碼
     */
    private static String password = "";
    // 使用靜態(tài)代碼塊為Properties對(duì)象賦值
    static {
        try {
            //實(shí)例化對(duì)象
            Properties properties = new Properties();
            //獲取properties文件的流對(duì)象
            InputStream in = RabbitMqUtil.class.getClassLoader().getResourceAsStream("rabbitmq.properties");
            properties.load(in);
            // 分別獲取 value
            host = properties.getProperty("host");
            port = Integer.parseInt(properties.getProperty("port"));
            virtualHost = properties.getProperty("virtualHost");
            username = properties.getProperty("username");
            password = properties.getProperty("password");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    /**
     * 獲取連接
     *
     * @return 連接
     */
    public static Connection getConnection() {
        try {
            // 創(chuàng)建連接工廠
            ConnectionFactory connectionFactory = new ConnectionFactory();
            // 設(shè)置連接 rabbitmq 主機(jī)
            connectionFactory.setHost(host);
            // 設(shè)置端口號(hào)
            connectionFactory.setPort(port);
            // 設(shè)置連接的虛擬主機(jī)（數(shù)據(jù)庫的感覺）
            connectionFactory.setVirtualHost(virtualHost);
            // 設(shè)置訪問虛擬主機(jī)的用戶名和密碼
            connectionFactory.setUsername(username);
            connectionFactory.setPassword(password);
            // 返回一個(gè)新連接
            return connectionFactory.newConnection();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }
    /**
     * 關(guān)閉通道和釋放連接
     *
     * @param channel    channel
     * @param connection connection
     */
    public static void close(Channel channel, Connection connection) {
        try {
            if (channel != null) {
                channel.close();
            }
            if (connection != null) {
                connection.close();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

• properties

host=192.168.122.1
port=5672
virtualHost=/rabbitmq_maven_01
username=admin
password=adminv

4.2 五種實(shí)現(xiàn)方式

說明：

• 隊(duì)列名，消息等等字符串內(nèi)容，更推薦定義成變量傳入，我文中都是直接寫在參數(shù)中的，這種魔法值的寫法，并不是很優(yōu)美。
• 生產(chǎn)者中使用了 Junit 單元測(cè)試，但是消費(fèi)者中卻在 main 函數(shù)中編寫，這是因?yàn)?，我們希望消費(fèi)者處于一個(gè)持續(xù)運(yùn)行等待的狀態(tài)，如果使用 Junit 會(huì)導(dǎo)致，程序在執(zhí)行一次后結(jié)束掉.
  • 除了在 main 函數(shù)中編寫，還可以考慮使用 sleep 等待或者 while(true) 讓程序不要直接終止掉。

4.2.1 簡(jiǎn)單隊(duì)列模式（Hello Word）

• Producer：消息的生產(chǎn)者（發(fā)送消息的程序）。
• Queue：消息隊(duì)列，理解為一個(gè)容器，生產(chǎn)者向它發(fā)送消息，它把消息存儲(chǔ)，等待消費(fèi)者消費(fèi)。
• Consumer：消息的消費(fèi)者（接收消息的程序）。

4.2.1.1 如何理解

由圖所示，簡(jiǎn)單隊(duì)列模式，一個(gè)生產(chǎn)者，經(jīng)過一個(gè)隊(duì)列，對(duì)應(yīng)一個(gè)消費(fèi)者。可以看做是點(diǎn)對(duì)點(diǎn)的一種傳輸方式，相較與 3.1.3 中的模型圖，最主要的特點(diǎn)就是看不到 Exchange（交換機(jī)） 和 routekey（路由鍵） ，正是因?yàn)檫@種模式簡(jiǎn)單，所以并不會(huì)涉及到復(fù)雜的條件分發(fā)等等，因此也不需要用戶去顯式的考慮交換機(jī)和路由鍵的問題。

• 但是要注意，這種模式并不是生產(chǎn)者直接對(duì)接隊(duì)列，而是用了默認(rèn)的交換機(jī)，默認(rèn)的交換機(jī)會(huì)把消息發(fā)送到和 routekey 名稱相同的隊(duì)列中去，這也是我們?cè)诤竺娲a中在 routekey 位置填寫了隊(duì)列名稱的原因

4.2.1.2 代碼實(shí)現(xiàn)

4.2.1.2.1 生產(chǎn)者代碼

public class Producer {
    @Test
    public void sendMessage() throws IOException, TimeoutException {
        // 通過工具類獲取連接
        Connection connection = RabbitMqUtil.getConnection();
        // 獲取連接通道
        Channel channel = connection.createChannel();
        // 通道綁定消息隊(duì)列
        channel.queueDeclare("queue1",false,false,false,null);
        // 發(fā)布消息
        channel.basicPublish("","queue1",null,"This is rabbitmq message 001 !".getBytes());
        // 通過工具關(guān)閉channel和釋放連接
        RabbitMqUtil.close(channel,connection);
    }
}public class Producer {
    @Test
    public void sendMessage() throws IOException, TimeoutException {
        // 通過工具類獲取連接
        Connection connection = RabbitMqUtil.getConnection();
        // 獲取連接通道
        Channel channel = connection.createChannel();
        // 通道綁定消息隊(duì)列
        channel.queueDeclare("queue1",false,false,false,null);
        // 發(fā)布消息
        channel.basicPublish("","queue1",null,"This is rabbitmq message 001 !".getBytes());
        // 通過工具關(guān)閉channel和釋放連接
        RabbitMqUtil.close(channel,connection);
    }
}

1.通過工具類獲取連接

2.獲取連接通道：根據(jù) 3.1.3 的模型圖可知，生產(chǎn)者需要在獲取到連接后，再獲取信道，才能去訪問后面的交換機(jī)隊(duì)列等。

3.通道綁定消息隊(duì)列：綁定隊(duì)列前，應(yīng)該綁定交換機(jī)，但是此模式中隱蔽了交換機(jī)的概念，背后使用了默認(rèn)的交換機(jī)，所以直接綁定隊(duì)列。

• queueDeclare 方法解釋
  • 參數(shù)1：queue（隊(duì)列名稱），如果隊(duì)列不存在，則自動(dòng)創(chuàng)建。
  • 參數(shù)2：durable（隊(duì)列是否持久化），持久化可以保證服務(wù)器重啟后此隊(duì)列仍然存在。
  • 參數(shù)3：exclusive（排他隊(duì)列）即是否獨(dú)占隊(duì)列，如果此項(xiàng)為 true，該隊(duì)列僅對(duì)首次申明它的連接可見，并在連接斷開時(shí)自動(dòng)刪除。
  • 參數(shù)4：autoDelete（自動(dòng)刪除），最后一個(gè)消費(fèi)者將消息消費(fèi)完畢后，自動(dòng)刪除隊(duì)列。
  • 參數(shù)5：arguments（攜帶附加屬性）。

4.發(fā)布消息：此處可以指定消息隊(duì)列的發(fā)送方法，以及內(nèi)容等，因?yàn)榇四Ｊ奖容^簡(jiǎn)單，所以沒有涉及到全部參數(shù)，后面的模式會(huì)有詳細(xì)的講解

• basicPublish 方法解釋
  • 參數(shù)1：exchange（交換機(jī)名稱）。
  • 參數(shù)2：routingKey（路由key），此處填寫隊(duì)列名，可理解為把消息發(fā)送到和 routekey 名稱相同的隊(duì)列中去。
  • 參數(shù)3：props（消息的控制狀態(tài)），可以在此處控制消息的持久化。參數(shù)為：MessageProperties.PERSISTENT_TEXT_PLAIN參數(shù)4：body（消息主體），類型是一個(gè)字節(jié)數(shù)組，要轉(zhuǎn)一下類型。

5.通過工具關(guān)閉channel和釋放連接：先關(guān)閉通道，再釋放連接。

4.2.1.2.2 消費(fèi)者代碼

public class Consumer {
    public static void main(String[] args) throws IOException, TimeoutException{
        // 通過工具類獲取連接
        Connection connection = RabbitMqUtil.getConnection();
        // 獲取連接通道
        Channel channel = connection.createChannel();
        // 通道綁定消息隊(duì)列
        channel.queueDeclare("queue1", false, false, false, null);
        // 消費(fèi)消息
        channel.basicConsume("queue1", true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("new String(body): " + new String(body));
            }
        });
    }
}

1.通過工具類獲取連接

2.獲取連接通道

3.通道綁定消息隊(duì)列

4.消費(fèi)消息：此處用來指定消費(fèi)哪個(gè)隊(duì)列的消息，以及一些機(jī)制和回調(diào)

• basicConsume 方法解釋
  • 參數(shù)1：queue（隊(duì)列名稱），即消費(fèi)哪個(gè)隊(duì)列的消息 。
  • 參數(shù)2：autoAck（自動(dòng)應(yīng)答）開始消息的自動(dòng)確認(rèn)機(jī)制，只要消費(fèi)了就從隊(duì)列刪除消息。
  • 參數(shù)3：callback（消費(fèi)時(shí)的回調(diào)接口），callback 的類型是 Consumer 這里使用了 DefaultConsumer 就是 Consumer 的一個(gè)實(shí)現(xiàn)類。其中重寫 handleDelivery 方法，就可以獲取到消費(fèi)的數(shù)據(jù)內(nèi)容了，這里主要使用了其中的 body，即查看消息主體，其他三個(gè)參數(shù)暫時(shí)還沒用到，有興趣可以先打印輸出一下，能先有個(gè)大概的了解。4.2.2 工作隊(duì)列模式（Work Queue）

• Producer：消息的生產(chǎn)者（發(fā)送消息的程序）。
• Queue：消息隊(duì)列，理解為一個(gè)容器，生產(chǎn)者向它發(fā)送消息，它把消息存儲(chǔ)，等待消費(fèi)者消費(fèi)。
• Consumer：消息的消費(fèi)者（接收消息的程序）。
  • 此處我們假設(shè) Consumer1、Consumer2、Consumer3 分別為完成任務(wù)速度不一樣快的消費(fèi)者，這會(huì)引出此模式的一個(gè)重點(diǎn)問題。

4.2.2.1 如何理解

工作模式由圖可以看出，就是在簡(jiǎn)單隊(duì)列模式的基礎(chǔ)上，增加了多個(gè)消費(fèi)者，也就是讓多個(gè)消費(fèi)者綁定同一個(gè)隊(duì)列，共同去消費(fèi)，這樣能解決簡(jiǎn)單隊(duì)列模式中，如果生產(chǎn)速速遠(yuǎn)大于消費(fèi)速度，而導(dǎo)致的消息堆積現(xiàn)象。

• 因?yàn)橄⒈幌M(fèi)后就會(huì)消失，所以不必?fù)?dān)心任務(wù)會(huì)重復(fù)執(zhí)行。

4.2.2.2 代碼實(shí)現(xiàn)

注：工作隊(duì)列模式有兩種

輪詢模式：每個(gè)消費(fèi)者均分消息公平分發(fā)模式（能者多勞）：按能力分發(fā)，處理速度快的分發(fā)的多，處理速度慢的分發(fā)的少

我們首先演示的是輪詢模式，根據(jù)它的缺點(diǎn)，又能引出公平分發(fā)模式

下面只描述與上面有差異的部分，在簡(jiǎn)單模式中，這些基本的方法都有介紹過

4.2.2.2.1 輪詢模式-生產(chǎn)者代碼

public class Producer {
    @Test
    public void sendMessage() throws IOException, TimeoutException {
        // 通過工具類獲取連接
        Connection connection = RabbitMqUtil.getConnection();
        // 獲取連接通道
        Channel channel = connection.createChannel();
        // 通道綁定消息隊(duì)列
        channel.queueDeclare("work", true, false, false, null);
        for (int i = 1; i <= 20; i++) {
            // 發(fā)布消息
            channel.basicPublish("", "work", null, (i + "號(hào)消息").getBytes());
        }
        // 通過工具關(guān)閉channel和釋放連接
        RabbitMqUtil.close(channel, connection);
    }
}

流程和簡(jiǎn)單隊(duì)列模式基本一致，有一些小小的改動(dòng)，生產(chǎn)者中主要就是加了層循環(huán)，因?yàn)橛卸鄠€(gè)消費(fèi)者，所以多發(fā)送一些消息，可以看出一些特點(diǎn)和問題。

4.2.2.2.2 輪詢模式-消費(fèi)者代碼

消費(fèi)者 1

public class Consumer1 {
    public static void main(String[] args) throws IOException {
        // 通過工具類獲取連接
        Connection connection = RabbitMqUtil.getConnection();
        // 獲取連接通道
        final Channel channel = connection.createChannel();
        // 通道綁定消息隊(duì)列
        channel.queueDeclare("work", true, false, false, null);
        // 消費(fèi)消息
        channel.basicConsume("work", true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("消費(fèi)者1號(hào)：消費(fèi)-" + new String(body));
            }
        });
    }
}

消費(fèi)者 2

public class Consumer2 {
    public static void main(String[] args) throws IOException {
        // 通過工具類獲取連接
        Connection connection = RabbitMqUtil.getConnection();
        // 獲取連接通道
        final Channel channel = connection.createChannel();
        // 通道綁定消息隊(duì)列
        channel.queueDeclare("work", true, false, false, null);
        // 消費(fèi)消息
        channel.basicConsume("work", true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消費(fèi)者2號(hào)：消費(fèi)-" + new String(body));
            }
        });
    }

上述兩個(gè)消費(fèi)者都在 basicConsume中開啟了自動(dòng) Ack 應(yīng)答，這一點(diǎn)下面會(huì)詳述，同時(shí)在消費(fèi)者 1 中，增加了 sleep 2s 的語句，模擬消費(fèi)者1處理消息速度慢，而消費(fèi)者2處理消息速度快的場(chǎng)景。

運(yùn)行結(jié)果：

Consumer1

消費(fèi)者1號(hào)：消費(fèi)-1號(hào)消息

消費(fèi)者1號(hào)：消費(fèi)-3號(hào)消息

消費(fèi)者1號(hào)：消費(fèi)-5號(hào)消息

消費(fèi)者1號(hào)：消費(fèi)-7號(hào)消息

消費(fèi)者1號(hào)：消費(fèi)-9號(hào)消息

消費(fèi)者1號(hào)：消費(fèi)-11號(hào)消息

消費(fèi)者1號(hào)：消費(fèi)-13號(hào)消息

消費(fèi)者1號(hào)：消費(fèi)-15號(hào)消息

消費(fèi)者1號(hào)：消費(fèi)-17號(hào)消息

消費(fèi)者1號(hào)：消費(fèi)-19號(hào)消息

Consumer2

消費(fèi)者2號(hào)：消費(fèi)-2號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-4號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-6號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-8號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-10號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-12號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-14號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-16號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-18號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-20號(hào)消息

觀察執(zhí)行過程：發(fā)現(xiàn)兩個(gè)消費(fèi)者雖然每個(gè)人最后都各自處理了一半的消息，而且是按照一人一條分配的，但是消費(fèi)者2號(hào)處理速度快，一下子就全部處理完了，但是消費(fèi)者1號(hào)，每一次處理都需要 2s 所以，只能緩慢的處理，而消費(fèi)者2號(hào)就處于一個(gè)空閑浪費(fèi)的情況了。

如何切換為公平分發(fā)模式呢？

這就和 basicConsume 中的第二個(gè)參數(shù)，開啟自動(dòng)確認(rèn)消費(fèi)有關(guān)了，它默認(rèn)是 true，也就代表只要一旦拿到隊(duì)列中分發(fā)給這個(gè)消費(fèi)者的消息，我就會(huì)自動(dòng)返回一個(gè)確認(rèn)消費(fèi)的標(biāo)識(shí)，隊(duì)列收到后就會(huì)自動(dòng)刪除掉隊(duì)列中的消息。

• 但是這其中有一個(gè)很重要的問題，這種方式就是將風(fēng)險(xiǎn)交給了消費(fèi)者，例如消費(fèi)者收到了自己需要處理的 10 條消息，剛消費(fèi)了 4 個(gè)，消費(fèi)者宕機(jī)，掛掉了，后面的 6 個(gè)消息就丟失了。

如果想要修改為按能力分配的方式，有兩個(gè)要點(diǎn)

1.設(shè)置通道一次只能消費(fèi)一個(gè)消息

2.關(guān)閉消息的自動(dòng)確認(rèn)，手動(dòng)確認(rèn)消息

4.2.2.2.3 公平分發(fā)模式-生產(chǎn)者代碼

public class Producer {
    @Test
    public void sendMessage() throws IOException, TimeoutException {
        // 通過工具類獲取連接
        Connection connection = RabbitMqUtil.getConnection();
        // 獲取連接通道
        Channel channel = connection.createChannel();
        // 一次只發(fā)送一條消息
        channel.basicQos(1);
        // 通道綁定消息隊(duì)列
        channel.queueDeclare("work", true, false, false, null);
        for (int i = 1; i <= 20; i++) {
            // 發(fā)布消息
            channel.basicPublish("", "work", null, (i + "號(hào)消息").getBytes());
        }
        // 通過工具關(guān)閉channel和釋放連接
        RabbitMqUtil.close(channel, connection);
    }

4.2.2.2.4 公平分發(fā)模式-消費(fèi)者代碼

消費(fèi)者1

public class Consumer1 {
    public static void main(String[] args) throws IOException {
        // 通過工具類獲取連接
        Connection connection = RabbitMqUtil.getConnection();
        // 獲取連接通道
        final Channel channel = connection.createChannel();
        // 一次只接受一條未確認(rèn)的消息
        channel.basicQos(1);
        // 通道綁定消息隊(duì)列
        channel.queueDeclare("work", true, false, false, null);
        // 消費(fèi)消息
        channel.basicConsume("work", false, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("消費(fèi)者1號(hào)：消費(fèi)-" + new String(body));
                // 返回 deliveryTag 代表隊(duì)列可以刪除此消息了
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        });
    }
}

消費(fèi)者2

public class Consumer2 {
    public static void main(String[] args) throws IOException {
        // 通過工具類獲取連接
        Connection connection = RabbitMqUtil.getConnection();
        // 獲取連接通道
        final Channel channel = connection.createChannel();
        //步驟一:一次只接受一條未確認(rèn)的消息
        channel.basicQos(1);
        // 通道綁定消息隊(duì)列
        channel.queueDeclare("work", true, false, false, null);
        // 消費(fèi)消息
        channel.basicConsume("work", false, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消費(fèi)者2號(hào)：消費(fèi)-" + new String(body));
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        });
    }
public class Consumer2 {
    public static void main(String[] args) throws IOException {
        // 通過工具類獲取連接
        Connection connection = RabbitMqUtil.getConnection();
        // 獲取連接通道
        final Channel channel = connection.createChannel();
        //步驟一:一次只接受一條未確認(rèn)的消息
        channel.basicQos(1);
        // 通道綁定消息隊(duì)列
        channel.queueDeclare("work", true, false, false, null);
        // 消費(fèi)消息
        channel.basicConsume("work", false, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消費(fèi)者2號(hào)：消費(fèi)-" + new String(body));
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        });
    }

運(yùn)行結(jié)果：

Consumer1

消費(fèi)者1號(hào)：消費(fèi)-1號(hào)消息

Consumer2

消費(fèi)者2號(hào)：消費(fèi)-2號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-3號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-4號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-5號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-6號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-7號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-8號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-9號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-10號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-11號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-12號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-13號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-14號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-15號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-16號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-17號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-18號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-19號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-20號(hào)消息

4.2.3 發(fā)布與訂閱模式（Fanout 廣播）

• Producer：消息的生產(chǎn)者（發(fā)送消息的程序）。
• Exchange ：交換機(jī)，負(fù)責(zé)發(fā)送消息給指定隊(duì)列。
• Queue：消息隊(duì)列，理解為一個(gè)容器，生產(chǎn)者向它發(fā)送消息，它把消息存儲(chǔ)，等待消費(fèi)者消費(fèi)。
• Consumer：消息的消費(fèi)者（接收消息的程序）。

4.2.3.1 如何理解

Fanout 直譯為 “扇出” 但是大家更多的會(huì)把它叫做廣播或者發(fā)布與訂閱，它是一種沒有路由key的模式，生產(chǎn)者將消息發(fā)送給交換機(jī)，交換機(jī)會(huì)把所有消息復(fù)制同步到所有與它綁定過的隊(duì)列上，而每個(gè)隊(duì)列只能有一個(gè)消費(fèi)者拿到這條消息，如果在一個(gè)消費(fèi)者連接中，創(chuàng)建多個(gè)通道，則會(huì)出現(xiàn)爭(zhēng)搶消息的結(jié)果。

4.2.3.2 代碼實(shí)現(xiàn)

注：下面只描述與上面有差異的部分，在簡(jiǎn)單模式中，這些基本的方法都有介紹過

4.2.3.2.1 生產(chǎn)者代碼

public class Producer {
    @Test
    public void sendMessage() throws IOException, TimeoutException {
        // 通過工具類獲取連接
        Connection connection = RabbitMqUtil.getConnection();
        // 獲取連接通道
        final Channel channel = connection.createChannel();
        // 聲明交換機(jī)
        channel.exchangeDeclare("order", "fanout");
        for (int i = 1; i <= 20; i++) {
            // 發(fā)布消息
            channel.basicPublish("order", "", null, "fanout!".getBytes());
        }
        // 通過工具關(guān)閉channel和釋放連接
        RabbitMqUtil.close(channel, connection);
    }
}

1.聲明交換機(jī)

• exchangeDeclare 方法解釋
  • 參數(shù)1：exchange（交換機(jī)名稱），如果交換機(jī)不存在，則自動(dòng)創(chuàng)建
  • 參數(shù)2：type（類型），此處選擇 fanout 模式

2.發(fā)布消息：在 basicPublish 方法的第一個(gè)參數(shù)中輸入上述定義好的交換機(jī)的名字，第二個(gè)參數(shù)，路由鍵為空

• 循環(huán) 20 條是為了演示消費(fèi)者

4.2.3.2.2 消費(fèi)者代碼

public class Consumer1 {
    public static void main(String[] args) throws IOException {
        // 通過工具類獲取連接
        Connection connection = RabbitMqUtil.getConnection();
        Channel channel = connection.createChannel();
        // 聲明交換機(jī)
        channel.exchangeDeclare("order", "fanout");
        // 創(chuàng)建臨時(shí)隊(duì)列
        String queue = channel.queueDeclare().getQueue();
        // 綁定臨時(shí)隊(duì)列和交換機(jī)
        channel.queueBind(queue, "order", "");
        // 消費(fèi)消息
        channel.basicConsume(queue, true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消費(fèi)者1號(hào)：消費(fèi)-" + new String(body));
            }
        });
    }
}

1.聲明交換機(jī)

2.創(chuàng)建臨時(shí)隊(duì)列

3..綁定臨時(shí)隊(duì)列和交換機(jī)

• queueBind 方法解釋
  • 參數(shù)1：queue（臨時(shí)隊(duì)列）
  • 參數(shù)2：exchange（交換機(jī)）
  • 參數(shù)3：routingKey（路由key）
• 消費(fèi)者2：演示了一個(gè)連接中，多個(gè)通道的情況

public class Consumer2 {
    public static void main(String[] args) throws IOException {
       // 通過工具類獲取連接
        Connection connection = RabbitMqUtil.getConnection();
        // 獲取連接通道
        Channel channel = connection.createChannel();
        Channel channel2 = connection.createChannel();
        // 聲明交換機(jī)
        channel.exchangeDeclare("order", "fanout");
        channel2.exchangeDeclare("order", "fanout");
        // 創(chuàng)建臨時(shí)隊(duì)列
        String queue = channel.queueDeclare().getQueue();
        System.out.println(queue);
        // 綁定臨時(shí)隊(duì)列和交換機(jī)
        channel.queueBind(queue, "order", "");
        channel2.queueBind(queue, "order", "");
        // 消費(fèi)消息
        channel.basicConsume(queue, true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消費(fèi)者2號(hào)：消費(fèi)-" + new String(body));
            }
        });
        // 消費(fèi)消息
        channel2.basicConsume(queue, true, new DefaultConsumer(channel2) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消費(fèi)者2-2號(hào)：消費(fèi)-" + new String(body));
            }
        });
    }
}

運(yùn)行結(jié)果：

消費(fèi)者2號(hào)：消費(fèi)-2號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-3號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-4號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-5號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-6號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-7號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-8號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-9號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-10號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-11號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-12號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-13號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-14號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-15號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-16號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-17號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-18號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-19號(hào)消息

消費(fèi)者2號(hào)：消費(fèi)-20號(hào)消息

4.2.3.2.3 為什么消費(fèi)者中也聲明交換機(jī)？

從上面的代碼中可以看出，在 Producer 和 Conusmer 中我們都分別聲明了交換機(jī)，但是消費(fèi)者由圖可知，并不會(huì)與交換機(jī)有直接的接觸，為什么消費(fèi)者中也聲明交換機(jī)呢？

這是為了保證 Producer 或者 Producer 執(zhí)行的時(shí)候，永遠(yuǎn)不會(huì)因?yàn)榻粨Q機(jī)還沒被聲明而出錯(cuò)，例如你只在 Producer 聲明了交換機(jī)，那么你就必須先啟動(dòng) Producer ，如果直接執(zhí)行 Conusmer 此時(shí)交換機(jī)就還不存在，就會(huì)報(bào)錯(cuò)。而全部寫入聲明，則可以保證不論先啟動(dòng)誰，都會(huì)聲明到交換機(jī)。

4.2.4 路由模式（ Routing / Direct）

• Producer：消息的生產(chǎn)者（發(fā)送消息的程序）。
• Exchange ：交換機(jī)，負(fù)責(zé)發(fā)送消息給指定隊(duì)列。
• routingKey：路由key，即上圖的 key1，key2 等，相當(dāng)于在交換機(jī)和隊(duì)列之間又加了一層限制
• Queue：消息隊(duì)列，理解為一個(gè)容器，生產(chǎn)者向它發(fā)送消息，它把消息存儲(chǔ)，等待消費(fèi)者消費(fèi)。
• Consumer：消息的消費(fèi)者（接收消息的程序）。

4.2.4.1 如何理解

路由模式的交換機(jī)類型是 direct，與 fanout 模式相比，多了路由 key 這個(gè)概念。生產(chǎn)者發(fā)送攜帶指定 routingKey（路由key） 的消息到交換機(jī)，交換機(jī)拿著此 routingKey 去找到綁定了這個(gè) routingKey 的隊(duì)列，然后發(fā)送到此隊(duì)列，一個(gè)隊(duì)列可以綁定多個(gè) routingKey 。

4.2.4.2 代碼實(shí)現(xiàn)

4.2.4.2.1 生產(chǎn)者代碼

public class Producer {
    @Test
    public void sendMessage() throws IOException, TimeoutException {
        // 通過工具類獲取連接
        Connection connection = RabbitMqUtil.getConnection();
        // 獲取連接通道
        Channel channel = connection.createChannel();
        // 聲明交換機(jī)
        channel.exchangeDeclare("order_direct", "direct");
        // 指定 routingKey 
        String key = "info";
        // 發(fā)布消息
        channel.basicPublish("order_direct", key, null, ("發(fā)送給指定路由" + key + "的消息").getBytes());
        // 通過工具關(guān)閉channel和釋放連接
        RabbitMqUtil.close(channel, connection);
    }
}

1.指定 routingKey ，即在 basicPublish 方法 的第二個(gè)參數(shù)中，指定 key 的值

4.2.4.2.2 消費(fèi)者代碼

• 消費(fèi)者 1

public class Consumer1 {
    public static void main(String[] args) throws IOException {
        // 通過工具類獲取連接
        Connection connection = RabbitMqUtil.getConnection();
        Channel channel = connection.createChannel();
        // 聲明交換機(jī)
        channel.exchangeDeclare("order_direct", "direct");
        // 獲取臨時(shí)隊(duì)列
        String queue = channel.queueDeclare().getQueue();
        // 綁定臨時(shí)隊(duì)列和交換機(jī)
        channel.queueBind(queue, "order_direct", "info");
        channel.queueBind(queue, "order_direct", "error");
        channel.queueBind(queue, "order_direct", "warn");
        // 消費(fèi)消息
        channel.basicConsume(queue, true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消費(fèi)者1：消費(fèi)-" + new String(body));
            }
        });
    }
}

1.只是在綁定隊(duì)列和交換機(jī)的時(shí)候，增加了 key 這個(gè)值

• 消費(fèi)者2

public class Consumer2 {
    public static void main(String[] args) throws IOException {
         // 通過工具類獲取連接
        Connection connection = RabbitMqUtil.getConnection();
        Channel channel = connection.createChannel();
        // 聲明交換機(jī)
        channel.exchangeDeclare("order_direct", "direct");
        // 獲取臨時(shí)隊(duì)列
        String queue = channel.queueDeclare().getQueue();
        // 綁定臨時(shí)隊(duì)列和交換機(jī)
        channel.queueBind(queue, "order_direct", "error");
        // 消費(fèi)消息
        channel.basicConsume(queue, true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消費(fèi)者2：消費(fèi)-" + new String(body));
            }
        });
    }
}

運(yùn)行結(jié)果：只有消費(fèi)者 1 收到了消息

[code]消費(fèi)者1：消費(fèi)-發(fā)送給指定路由info的消息