# 設(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（連接）：此處用來(lái)管理與
- RabbitMQ 建立連接后的生產(chǎn)者和消費(fèi)者
- Channels（通道）：連接建立后，會(huì)形成通道，消息的投遞獲取依賴通道。
- Exchanges（交換機(jī)）：用來(lái)實(shí)現(xiàn)消息的路由
- Queues（隊(duì)列）：存放消息的隊(duì)列，消息等待被消費(fèi)，消費(fèi)后被移除隊(duì)列。
- Admin（管理）：用于對(duì)管理用戶，以及對(duì)應(yīng)權(quán)限進(jìn)行設(shè)置，如下圖所示
Tags 就是用來(lái)指定用戶的角色
- 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è)版
2.通過查看版本，檢查安裝是否成功
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ù)的說(shuō)明：
-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 管理頁(yè)面的 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 代碼來(lái)實(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ī)之間才能相互通信交流。它的三要素是：語(yǔ)法、語(yǔ)義、時(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.語(yǔ)法：數(shù)據(jù)與控制信息的結(jié)構(gòu)和格式，以及數(shù)據(jù)出現(xiàn)的順序。
2.語(yǔ)義：解釋控制信息每個(gè)部分的意義，以及規(guī)定了需要發(fā)出何種控制信息以及完成的動(dòng)作做出何種響應(yīng)。
3.時(shí)序：對(duì)事件發(fā)生順序的詳細(xì)說(shuō)明。
人們形象地把這三個(gè)要素描述為：做什么，怎么做，做的順序。
舉個(gè)例子 HTTP 協(xié)議
語(yǔ)法：HTTP 規(guī)定了請(qǐng)求報(bào)文和響應(yīng)報(bào)文的格式
語(yǔ)義：客戶端主動(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è)消息中間件來(lái)說(shuō)，其主要責(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)確的說(shuō)是一種 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ù)庫(kù)服務(wù)，而 Virtual Host 就是其中每個(gè)數(shù)據(jù)庫(kù)的感覺，不同項(xiàng)目可以對(duì)應(yīng)不同的數(shù)據(jù)庫(kù)，其中有著項(xiàng)目所屬的業(yè)務(wù)表等等。
- 每個(gè) Virtual Host 中，可以有若干個(gè) Exchange 和 Queue。
Exchange：交換機(jī)，用來(lái)接收生產(chǎn)者發(fā)送的消息，然后將這些消息根據(jù)路由鍵發(fā)送到隊(duì)列。Binding：Exchange 和 Queue 之間的虛擬連接，Binding 中可以包括多個(gè) Routing key。Routing key：路由規(guī)則，虛擬機(jī)用它來(lái)確認(rèn)如何路由一個(gè)特定消息。Queue：消息隊(duì)列，它是消息的容器，用來(lái)保存消息，每一條消息都能傳入一個(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，用來(lái)為這個(gè)Java項(xiàng)目服務(wù)，大家還可以創(chuàng)建一個(gè)新的用戶，然后對(duì)其開啟這個(gè) Virtual Hosts 的訪問權(quán)限（即將虛擬主機(jī)與用戶綁定）。我們這里還是用 admin（我之前創(chuàng)建的一個(gè)管理員權(quán)限用戶） 來(lái)演示。
注：這部分不去做也可以，直接用 / 和 admin 用戶也行
4.1.3 創(chuàng)建連接工具類
由于我們后面要演示多種例子，而每一次獲取連接和釋放連接、關(guān)閉資源等操作代碼都是一致的，為了防止代碼冗余，優(yōu)化代碼，更易理解，提取出一個(gè)工具類，這樣大家將重心放在不同實(shí)現(xiàn)方式的對(duì)比上就行了。
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ù)庫(kù)的感覺）
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)方式
說(shuō)明：
- 隊(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)者?？梢钥醋鍪屈c(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)消息：此處用來(lá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 的語(yǔ)句，模擬消費(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)誰(shuí)，都會(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)者代碼
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è)值
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的消息
通配符匹配模式的交換機(jī)類型為 topic，因?yàn)樗c Direct 模式很相似，所以大家有時(shí)候也會(huì)把 Direct 模式和 Topic 共同歸入路由模式下，它們的區(qū)別就是，Direct 模式的 routingKey 是一個(gè)指定的值，而 Topic 模式的 routingKey 可以使用通配符， 而且一般都是由一個(gè)或多個(gè)單詞組成，多個(gè)單詞之間以”.”分割，例如： ideal.insert。
SpringBoot 提供 Spring For RabbitMQ 的啟動(dòng)器，同時(shí)提供了一系列注解以及 RabbitTemplate 供我們使用，能夠極大的簡(jiǎn)化開發(fā) RabbitMQ 的步驟，下面分別演示了【5.1 基于純注解】 以及【 5.2 基于注解 + 配置類】 的寫法，其使用方式大同小異，只是聲明和綁定隊(duì)列交換機(jī)等的位置不同。一般認(rèn)為后者更好維護(hù)管理，任選其一即可。
1.首先創(chuàng)建 SprinBoot 項(xiàng)目，然后選擇 RabbitMQ 的啟動(dòng)器，以及單元測(cè)試等基本啟動(dòng)器
注：此方式?jīng)]有創(chuàng)建配置類來(lái)管理隊(duì)列以及交換機(jī)的聲明和綁定等，而是全部通過注解的方式直接在消費(fèi)者中寫入
通過 RabbitTemplate 的 convertAndSend 方法用來(lái)發(fā)送消息，他有多種重載方式，今天分別會(huì)用到 2 個(gè) 和 3 個(gè)參數(shù)的
2.監(jiān)聽 RabbitMQ，在 @RabbitListener 注解中，可以實(shí)現(xiàn)，隊(duì)列的聲明，以及后面交換機(jī)與隊(duì)列的綁定等
3.在方法上添加 @RabbitHandler 注解，就能夠?qū)崿F(xiàn)自動(dòng)回調(diào)，這樣我們就能拿到生產(chǎn)者中的消息了
到現(xiàn)在已經(jīng)實(shí)現(xiàn)了修改配置文件的方式實(shí)現(xiàn)按能力分配，補(bǔ)充幾個(gè)配置的內(nèi)容，我們上面只用了一部分，其他的方便大家參考，yml 和 properties 大家自己選擇即可
其實(shí)這種方式，就是將交換機(jī)，隊(duì)列的聲明和綁定都在配置類中進(jìn)行，一個(gè)是消費(fèi)者中的注解變的簡(jiǎn)潔了，再有就是統(tǒng)一管理，更加條理，而且生產(chǎn)者和消費(fèi)者引用的時(shí)候也更加方便，日后修改的時(shí)候，也不需要對(duì)每一處都修改。
2.定義常量：將交換機(jī)名，隊(duì)列名，路由key 等都可以創(chuàng)建為常量，調(diào)用，管理和修改都非常方便，還可以創(chuàng)建出一個(gè)專門的 RabbitMQ 的常量類。
5.綁定交換機(jī)和隊(duì)列：利用 BindingBuilder 的 bind 方法綁定隊(duì)列，to 綁定到指定交換機(jī)，with 傳入路由key
