RabbitMQ 3.9( 基礎 )
1、認識MQ
1.1、什麼是MQ？
- MQ全稱：message queue 即 消息隊列
- 這個隊列遵循的原則：FIFO 即 先進先出
- 隊列裡面存的就是message
1.2、為什麼要用MQ？
1.2.1、流量削峰
- 這種情況，要是訪問 1020次 / s呢？這種肯定會讓支付系統宕機了，因為太大了嘛，受不了，所以：流量削峰
- 這樣就讓message排著隊了，然後使用FIFO先進先出，這樣支付系統就可以承受得了了
1.2.2、應用解耦
- 上面這種，只要支付系統或庫存系統其中一個挂彩了，那麼訂單系統也要挂彩，因此：解耦唄
- 而採用了MQ之後，支付系統和庫存系統有一個出問題，那麼它的處理記憶體是在MQ中的，此時訂單系統就不會有影響，可以正常完成，等到故障恢復了，訂單系統再處理對應的事情，這就提高了系統的可用性
1.2.3、非同步處理
- 如上圖，訂單系統要調用支付系統的API，而訂單系統並不知道支付系統處理對應的業務需要多久，要解決可以採用訂單系統隔一定時間又去訪問支付系統，看處理完沒有，而使用MQ更容易解決。
1.3、RabbitMQ的原理？
- 圖中的東西後續會慢慢見到
- Broker實體：接收和分發消息的應用 / RabbitMQ Server / Message Broker
- 而上圖中RabbitMQ的四個核心就是：Producer生產者、exchange交換機、queue隊列、Consumer消費者
- Producer生產者：就是負責推送消息的程式
- Exchange交換機：接收來自生產者的消息，並且把消息放到隊列中
- queue隊列：就是一個數據結構，本質就是一個很大的消息緩衝區，許多生產者可以把消息推送到一個隊列，許多消費者可以從一個隊列中獲得數據
- Consumer消費者：就是接收消息的程式
- 注意：生產者、消息中間件MQ、消費者大多時候並不是在同一台機器上的，所以：生產者有時可以是消費者；而消費者有時也可以是生產者
- Connection鏈接：就是讓Producer生產者、Broker實體、Consumer消費者之間建立TCP鏈接
- Virtual host虛擬機：處於多租戶和安全因素考慮而設計的，當多個不同的用戶使用同一個 RabbitMQ server 提供的服務時，可以劃分出多個 vhost，每個用戶在自己的 vhost 創建 exchange／queue 等
- Channel信道：就是發消息的通道，它是在Connection內部建立的邏輯連接
- Routes路由策略 / binding綁定：交換機以什麼樣的策略將消息發布到Queue。也就是exchange交換機 和 queue隊列之間的聯繫，即 二者之間的虛擬連接，它裡面可以包含routing key 路由鍵
1.4、RabbitMQ的通訊方式
- 這個玩意兒在官網中有圖，地址：//www.rabbitmq.com/getstarted.html 學完之後這張圖最好放在自己腦海里，平時開發玩的就是這些，下面的工作模式在後續會慢慢接觸
- 另外：下面放的是七種，實質上第六種RPC用得很少
- 1、hello word – 簡單模式
- 2、work queues – 工作模式
- 3、publish / subscribe – 發布訂閱模式
- 4、Routing – 路由模式
- 5、Topics – 主題模式
- 6、RPC模式 – 不用了解也行
- 7、publisher confirms – 發布確認模式
2、安裝RabbitMQ
- 以下的方式自行選擇一種即可
2.1、在Centos 7下安裝
- 查看自己的Linux版本
uname -a
2.1.1、使用rpm紅帽軟體
準備工作
- 1、下載Erlang，因為：RabbitMQ是Erlang語言寫的，Erlang下載地址【 ps：這是官網 】：//www.erlang.org/downloads，選擇自己要的版本即可
- 另外：RabbitMQ和Erlang的版本對應關係鏈接地址 //www.rabbitmq.com/which-erlang.html
- 當然：上面這種是下載gz壓縮包，配置挺煩的，可以直接去github中下載rpm文件，地址：//github.com/rabbitmq/erlang-rpm/releases ， 選擇自己需要的版本即可，注意一個問題：要看是基於什麼Linux的版本
- 要是github下載慢的話，都有自己的文明上網加速方式，要是沒有的話，可以進入 //github.com/fhefh2015/Fast-GitHub 下載好了然後集成到自己瀏覽器的擴展程式中即可，而如果進入github很慢的話，可以選擇去gitee中搜索一個叫做：
dev-sidecar的東西安裝，這樣以後進入github就很快了，還有另外的很多方式，不介紹了。
- 2、執行
rpm -ivh erlang文件命令- i 就是 install的意思
- vh 就是顯示安裝進度條
- 注意：需要保證自己的Linux中有rpm命令，沒有的話，執行
yum install rpm指令即可安裝rpm
- 3、安裝RabbitMQ需要的依賴環境
yum install socat -y
- 4、下載RabbitMQ的rpm文件，github地址：//github.com/rabbitmq/rabbitmq-server/releases ， 選擇自己要的版本即可
- 5、安裝RabbitMQ
-
6、啟動RabbitMQ服務
啟動服務 sbin/service rabbitmq-server start 停止服務 /sbin/service rabbitmq-server stop 查看啟動狀態 /sbin/service rabbitmq-server status 開啟開機自動 chkconfig rabbitmq-server on
- 查看啟動狀態
- 這表示正在啟動，需要等一會兒，看到下面的樣子就表示啟動成功
- 7、安裝web管理插件
1、停止RabbitMQ服務
service rabbitmq-server stop // 使用上面的命令 /sbin/service rabbitmq-server stop也行
2、安裝插件
rabbitmq-plugins enable rabbitmq_management
3、開啟RabbitMQ服務
service rabbitmq-server start
- 要是訪問不了，看看自己的防火牆關沒關啊
# 查看防火牆狀態
systemctl status firewalld
# 關閉防火牆
systemctl stop firewalld
# 一勞永逸 禁用防火牆
systemctl enable firewalld
- 同時查看自己的伺服器有沒有開放15672埠，不同的東西有不同的處理方式，如我的雲伺服器直接在伺服器網址中添加規則即可，其他的方式自行百度
2.1.2、使用Docker安裝
-
需要保證自己的Linux中有Docker容器，教程鏈接：//www.cnblogs.com/xiegongzi/p/15621992.html
-
使用下面的兩種方式都不需要進行web管理插件的安裝和erlang的安裝
- 1、查看自己的docker容器中是否已有了rabbitmq這個名字的鏡像
docker images
- 刪除鏡像
docker rmi 鏡像ID // 如上例的 dockerrmi 16c 即可刪除鏡像
- 2、拉取RabbitMQ鏡像 並 啟動Docker容器
docker run -it --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3.9-management
- 3、查看Docker容器是否啟動
docker ps
- 4、再次在瀏覽器進行訪問就可以吃雞了，不需要再安裝插件啊，剛剛上一步拉鏡像和啟動時已經安裝好了
2.1.3、使用Docker-compose安裝
- 採用了第二種方式的話，記得把已經啟動的Docker容器關了，以下是附加的一些Docker的基操
# 拉取鏡像
docker pull 鏡像名稱
# 查看全部鏡像
docker images
# 刪除鏡像
docker rmi 鏡像ID
# 將本地的鏡像導出
docker save -o 導出的路徑 鏡像id
# 載入本地的鏡像文件
docker load -i 鏡像文件
# 修改鏡像名稱
docker tag 鏡像id 新鏡像名稱:版本
# 簡單運行操作
docker run 鏡像ID | 鏡像名稱
# 跟參數的運行
docker run -d -p 宿主機埠:容器埠 --name 容器名稱 鏡像ID | 鏡像名稱
# 如：docker run -d -p 8081:8080 --name tomcat b8
# -d：代表後台運行容器
# -p 宿主機埠:容器埠：為了映射當前Linux的埠和容器的埠
# --name 容器名稱：指定容器的名稱
# 查看運行的容器
docker ps [-qa]
# -a：查看全部的容器，包括沒有運行
# -q：只查看容器的標識
# 查看日誌
docker logs -f 容器id
# -f：可以滾動查看日誌的最後幾行
# 進入容器內部
docker exec -it 容器id bash
# 退出容器：exit
# 將宿主機的文件複製到容器內部的指定目錄
docker cp 文件名稱 容器id:容器內部路徑
docker cp index.html 982:/usr/local/tomcat/webapps/ROOT
=====================================================================
# 重新啟動容器
docker restart 容器id
# 啟動停止運行的容器
docker start 容器id
# 停止指定的容器（刪除容器前，需要先停止容器）
docker stop 容器id
# 停止全部容器
docker stop $(docker ps -qa)
# 刪除指定容器
docker rm 容器id
# 刪除全部容器
docker rm $(docker ps -qa)
- 1、創建一個文件夾，這些我很早之前就玩過了，所以建好了的
# 創建文件夾
mkdir 文件夾名
- 2、進入文件夾，創建docker-compose.yml文件，注意：文件名必須是這個
# 創建文件
touch docker-compose.yml
- 3、編輯docker-compose.yml文件
# 編輯文件
vim docker-compose.yml
- 裡面編寫的內容如下，編寫好保存即可。注意：別用tab縮進啊，會出問題的，另外：每句的後面別有空格，嚴格遵循yml格式的
version: "3.1"
services:
rabbitmq:
# 鏡像
image: rabbitmq:3.9-management
# 自啟
restart: always
# Docker容器名
container_name: rabbitmq
# 埠號，docker容器內部埠 映射 外部埠
ports:
- 5672:5672
- 15672:15672
# 數據卷映射 把容器裡面的東西映射到容器外面去 容易操作，否則每次都要進入容器
volumes:
- ./data:/opt/install/rabbitMQ-docker/
- 4、在docker-compose.yml所在路徑執行如下命令，注意：一定要在此文件路徑中才行，因為默認是在當前文件夾下找尋docker-compose文件
# 啟動
docker-compose up -d
# -d 後台啟動
=========================================================
# 附加內容：docker-compose的一些命令操作
# 1. 基於docker-compose.yml啟動管理的容器
docker-compose up -d
# 2. 關閉並刪除容器
docker-compose down
# 3. 開啟|關閉|重啟已經存在的由docker-compose維護的容器
docker-compose start|stop|restart
# 4. 查看由docker-compose管理的容器
docker-compose ps
# 5. 查看日誌
docker-compose logs -f
# 有興趣的也可以去了解docker-file自定義鏡像
- 去瀏覽器訪問一樣的吃雞
- 上面就是RabbitMQ的基操做完了，不過默認帳號是guest遊客狀態，很多事情還做不了呢，所以還得做一些操作
2.1.4、解決不能登入web管理介面的問題
2.1.4.1、使用rpm紅帽軟體安裝的RabbitMQ
- 這種方式直接使用guest進行登錄是不得吃的
- 這是因為guest是遊客身份，不能進入，需要添加新用戶
查看當前用戶 / 角色有哪些
rabbitmqctl list_users
刪除用戶
rabbitmqctl delete_user 用戶名
添加用戶
rabbitmqctl add_user 用戶名 密碼
設置用戶角色
rabbitmqctl set_user_tags 用戶名 administrator
設置用戶許可權【 ps：guest角色就是沒有這一步 】
rabbitmqctl set_permissions -p "/" 用戶名 ".*" ".*" ".*"
# 設置用戶許可權指令解釋
set_permissions [-p <vhostpath>] <user> <conf> <write> <read>
- 現在使用admin去瀏覽器登錄就可以了
2.1.4.2、使用docker 或 docker-compose安裝的RabbitMQ
- 這兩種方式直接使用guest就可以進行登錄，後續的操作就是一樣的了
3、開始玩RabbitMQ
- 創建Maven項目 並導入如下依賴
<dependencies>
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.9.0</version>
</dependency>
</dependencies>
- 回到前面的RabbitMQ原理圖
3.1、Hello word 簡單模式
- 對照原理圖來玩，官網中有Hello word的模式圖
- 即：一個生產者Producer、一個默認交換機Exchange、一個隊列queue、一個消費者Consumer
生產者
- 就是下圖前面部分
package cn.zixieqing;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Producer {
private static final String HOST = "ip"; // 放RabbitMQ服務的伺服器ip
private static final int PORT = 5672; // 伺服器中RabbitMQ的埠號，在瀏覽器用的15672是通過5672映射出來的15672
private static final String USER_NAME = "admin";
private static final String PASSWORD = "admin";
private static final String QUEUE_NAME = "hello word";
public static void main(String[] args) throws IOException, TimeoutException {
// 1、獲取鏈接工廠
ConnectionFactory factory = new ConnectionFactory();
// 2、設置鏈接資訊
factory.setHost(HOST);
factory.setPort(PORT);
factory.setUsername(USER_NAME);
factory.setPassword(PASSWORD);
/*
當然：這裡還可以設置vhost虛擬機 - 前提是自己在web管理介面中添加了vhost
factory.setVirtualHost();
*/
// 3、獲取鏈接Connection
Connection connection = factory.newConnection();
// 4、創建channel信道 - 它才是去和交換機 / 隊列打交道的
Channel channel = connection.createChannel();
// 5、準備一個隊列queue
// 這裡理論上是去和exchange打交道，但是：這裡是hello word簡單模式，所以直接使用默認的exchange即可
/*
下面這是參數的完整意思，源碼中偷懶了，沒有見名知意
queueDeclare( queueName,isPersist,isShare,isAutoDelete,properties )
參數1、隊列名字
參數2、是否持久化( 保存到磁碟 ），默認是在記憶體中的
參數3、是否共享，即：是否只供一個消費者消費，是否讓多個消費者共享這個隊列中的資訊
參數4、是否自動刪除，即：最後一個消費者獲取資訊之後，這個隊列是否自動刪除
參數5、其他配置項，這涉及到後面的知識，目前選擇null
*/
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
System.out.println("正在發送資訊！！！");
// 6、推送資訊到隊列中
// 準備發送的資訊內容
String message = "it is hello word";
/*
basicPublish( exchangeName,queueName,properties,message )
參數1、交互機名字 - 目前使用了默認的
參數2、指定路由規則 - 目前使用隊列名字
參數3、指定傳遞的消息所攜帶的properties
參數4、推送的具體消息 - byte類型的
*/
channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
// 7、釋放資源 - 倒著關閉即可
if ( null != channel ) channel.close();
if ( null != connection ) connection.close();
System.out.println("消息發送完畢");
}
}
- 運行之後，去瀏覽器管理介面進行查看
消費者
public class Consumer {
private static final String HOST = "ip"; // 自己的伺服器ip
private static final int PORT = 5672;
private static final String USER_NAME = "admin";
private static final String PASSWORD = "admin";
private static final String QUEUE_NAME = "hello word";
public static void main(String[] args) throws IOException, TimeoutException {
// 1、創建鏈接工廠
ConnectionFactory factory = new ConnectionFactory();
// 2、設置鏈接資訊
factory.setHost(HOST);
factory.setPort(PORT);
factory.setUsername(USER_NAME);
factory.setPassword(PASSWORD);
// 3、創建鏈接對象
Connection connection = factory.newConnection();
// 4、創建信道channel
Channel channel = connection.createChannel();
// 5、從指定隊列中獲取消息
/*
basicConsume( queueName,isAutoAnswer,deliverCallback,cancelCallback )
參數1、隊列名
參數2、是否自動應答,為true時，消費者接收到消息後，會立即告訴RabbitMQ
參數3、消費者如何消費消息的回調
參數4、消費者取消消費的回調
*/
System.out.println("開始接收消息！！！");
DeliverCallback deliverCallback = (consumerTag, message) -> {
System.out.println("接收到了消息：" + new String(message.getBody(), StandardCharsets.UTF_8) );
};
CancelCallback cancelCallback = consumerTag -> System.out.println("消費者取消了消費資訊行為");
channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);
// 6、釋放資源 - 但是這裡不能直接關閉啊，否則：看不到接收的結果的，可以選擇不關，也可以選擇加一句程式碼System.in.read();
// channel.close();
// connection.close();
}
}
3.2、work queue工作隊列模式
- 流程圖就是官網中的
- 一個生產者批量生產消息
- 一個默認交換機
- 一個隊列
- 多個消費者
- 換言之：就是有大量的任務 / 密集型任務有待處理（ 生產者生產的消息 ），此時我們就將這些任務推到隊列中去，然後使用多個工作執行緒（ 消費者 ）來進行處理，否則：一堆任務直接就跑來了，那消費者不得亂套了，因此：這種就需要讓這種模式具有如下的特點：
- 1、消息是有序排好的（ 也就是在隊列中 ）
- 2、工作執行緒 / 消費者不能同時接收同一個消息，換言之：生產者推送的任務必須是輪詢分發的，即：工作執行緒1接收第一個，工作執行緒2接收第二個；工作執行緒1再接收第三個，工作執行緒2接收第四個
抽取RabbitMQ鏈接的工具類
package cn.zixieqing.util;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class MQUtil {
private static final String HOST = "自己的ip";
private static final int PORT = 5672;
private static final String USER_NAME = "admin";
private static final String PASSWORD = "admin";
public static Channel getChannel(String vHost ) throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost(HOST);
factory.setPort(PORT);
factory.setUsername(USER_NAME);
factory.setPassword(PASSWORD);
if ( !vHost.isEmpty() ) factory.setVirtualHost(vHost);
return factory.newConnection().createChannel();
}
}
生產者
- 和hello word沒什麼兩樣
package cn.zixieqing.workqueue;
import cn.zixieqing.util.MQUtil;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.util.Scanner;
import java.util.concurrent.TimeoutException;
public class WorkProducer {
private static final String QUEUE_NAME = "work queue";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = MQUtil.getChannel("");
// 1、聲明隊列
/*
下面這是參數的完整意思，源碼中偷懶了，沒有見名知意
queueDeclare( queueName,isPersist,isShare,isAutoDelete,properties )
參數1、隊列名字
參數2、是否持久化( 保存到磁碟 ），默認是在記憶體中的
參數3、是否共享，即：是否只供一個消費者消費，是否讓多個消費者共享這個隊列中的資訊
參數4、是否自動刪除，即：最後一個消費者獲取資訊之後，這個隊列是否自動刪除
參數5、其他配置項，這涉及到後面的知識，目前選擇null
*/
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 2、準備消息
System.out.println("請輸入要推送的資訊，按回車確認：");
Scanner input = new Scanner(System.in);
// 3、推送資訊到隊列中
while (input.hasNext()) {
/*
basicPublish( exchangeName,routing key,properties,message )
參數1、交互機名字 - 目前是使用了默認的
參數2、指定路由規則 - 目前使用隊列名字
參數3、指定傳遞的消息所攜帶的properties
參數4、推送的具體消息 - byte類型的
*/
String message = input.next();
channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
System.out.println("消息====>" + message + "====>推送完畢！");
}
}
}
消費者
- 消費者01
package cn.zixieqing.workqueue;
import cn.zixieqing.util.MQUtil;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;
public class WorkConsumer {
private static final String QUEUE_NAME = "work queue";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = MQUtil.getChannel("");
DeliverCallback deliverCallback = (consumerTag, message) -> {
System.out.println("接收到了消息====>" + new String(message.getBody(), StandardCharsets.UTF_8));
};
CancelCallback cancelCallback = consumerTag -> {
System.out.println( consumerTag + "消費者中斷了接收消息====>" );
};
System.out.println("消費者01正在接收消息......");
channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);
}
}
- 消費者02
package cn.zixieqing.workqueue;
import cn.zixieqing.util.MQUtil;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;
public class WorkConsumer {
private static final String QUEUE_NAME = "work queue";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = MQUtil.getChannel("");
DeliverCallback deliverCallback = (consumerTag, message) -> {
System.out.println("接收到了消息====>" + new String(message.getBody(), StandardCharsets.UTF_8));
};
CancelCallback cancelCallback = consumerTag -> {
System.out.println( consumerTag + "消費者中斷了接收消息====>" );
};
System.out.println("消費者02正在接收消息......");
channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);
}
}
3.3、消息應答機制
- 消費者在接收到消息並且處理該消息之後，告訴 rabbitmq 它已經處理了，rabbitmq 可以把該消息刪除了
- 目的就是為了保證數據的安全，如果沒有這個機制的話，那麼就會造成下面的情況
- 消費者接收隊列中的消息時，沒接收完，出現異常了，然後此時MQ以為消費者已經把消息接收並處理了（ MQ並沒有接收到消息有沒有被消費者處理完畢 ），然後MQ就把隊列 / 消息給刪了，後續消費者異常恢復之後再次接收消息，就會出現：接收不到了
3.3.1、消息應答機制的分類
- 這個東西已經見過了
/*
basicConsume( queueName,isAutoAnswer,deliverCallback,cancelCallback )
參數1、隊列名
參數2、是否自動應答,為true時，消費者接收到消息後，會立即告訴RabbitMQ
參數3、消費者如何消費消息的回調
參數4、消費者取消消費的回調
*/
channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);
3.3.1.1、自動應答
- 指的是：消息發送後立即被認為已經傳送成功
- 需要具備的條件：
- 1、發送的消息很多，就是高吞吐量的那種
- 2、發送的消息在傳輸方面是安全的
- 優點：處理效率快，很高效
3.3.1.2、手動應答
-
就是我們自己去設定，好處是可以批量應答並且減少網路擁堵
-
調用的API如下：
-
Channel.basicACK( long, boolean ); // 用於肯定確認，即：MQ已知道該消息 並且 該消息已經成功被處理了，所以MQ可以將其丟棄了 Channel.basicNack( long, boolena, boolean ); // 用於否定確認 Channel.basicReject( long, boolea ); // 用於否定確認 與Channel.basicNack( long, boolena, boolean )相比，少了一個參數，這個參數名字叫做：multiple
-
-
multiple參數說明，它為true和false有著截然不同的意義【 ps：建議弄成false，雖然是挨個去處理，從而應答，效率慢，但是：數據安全，否則：很大可能造成數據丟失 】
- true 代表批量應答MQ，channel 上未應答 / 消費者未被處理完畢的消息
- false 只會處理隊列放到channel信道中當前正在處理的消息告知MQ是否確認應答 / 消費者處理完畢了
3.3.1.3、消息重新入隊原理
- 指的是：如果消費者由於某些原因失去連接(其通道已關閉，連接已關閉或 TCP 連接丟失)，導致消息未發送 ACK 確認，RabbitMQ 將了解到消息未完全處理，並將對其重新排隊。如果此時其他消費者可以處理，它將很快將其重新分發給另一個消費者。這樣，即使某個消費者偶爾死亡，也可以確保不會丟失任何消息
- 如下圖：消息1原本是C1這個消費者來接收的，但是C1失去鏈接了，而C2消費者並沒有斷開鏈接，所以：最後MQ將消息重新入隊queue，然後讓C2來處理消息1
3.3.1.4、手動應答的程式碼演示
生產者
package cn.zixieqing.ACK;
import cn.zixieqing.util.MQUtil;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Scanner;
import java.util.concurrent.TimeoutException;
public class AckProducer {
private static final String QUEUE_NAME = "ack queue";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = MQUtil.getChannel("");
// 聲明隊列
/*
下面這是參數的完整意思，源碼中偷懶了，沒有見名知意
queueDeclare( queueName,isPersist,isShare,isAutoDelete,properties )
參數1、隊列名字
參數2、是否持久化( 保存到磁碟 ），默認是在記憶體中的
參數3、是否共享，即：是否只供一個消費者消費，是否讓多個消費者共享這個隊列中的資訊
參數4、是否自動刪除，即：最後一個消費者獲取資訊之後，這個隊列是否自動刪除
參數5、其他配置項，這涉及到後面的知識，目前選擇null
*/
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
System.out.println("請輸入要推送的消息：");
Scanner input = new Scanner(System.in);
while (input.hasNext()) {
/*
basicPublish( exchangeName,routing key,properties,message )
參數1、交互機名字 - 使用了默認的
參數2、指定路由規則，使用隊列名字
參數3、指定傳遞的消息所攜帶的properties
參數4、推送的具體消息 - byte類型的
*/
String message = input.next();
channel.basicPublish("",QUEUE_NAME,null,message.getBytes(StandardCharsets.UTF_8));
System.out.println("消息====>" + message + "推送完畢");
}
}
}
消費者01
package cn.zixieqing.ACK;
import cn.zixieqing.util.MQUtil;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.concurrent.TimeoutException;
public class AckConsumer {
private static final String QUEUE_NAME = "ack queue";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = MQUtil.getChannel("");
DeliverCallback deliverCallback = (consumerTag, message) -> {
try {
Thread.sleep(5*1000);
System.out.println("接收到了消息=====>" + new String( message.getBody(), StandardCharsets.UTF_8 ));
// 添加手動應答
/*
basicAck( long, boolean )
參數1、消息的標識tag，這個標識就相當於是消息的ID
參數2、是否批量應答multiple
*/
channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
} catch (InterruptedException e) {
e.printStackTrace();
}
};
System.out.println("消費者01正在接收消息，需要5秒處理完");
channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {
System.out.println("觸發消費者取消消費消息行為的回調");
System.out.println(Arrays.toString(consumerTag.getBytes(StandardCharsets.UTF_8)));
});
}
}
消費者02
package cn.zixieqing.ACK;
import cn.zixieqing.util.MQUtil;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.concurrent.TimeoutException;
public class AckConsumer {
private static final String QUEUE_NAME = "ack queue";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = MQUtil.getChannel("");
DeliverCallback deliverCallback = (consumerTag, message) -> {
try {
Thread.sleep(10*1000);
System.out.println("接收到了消息=====>" + new String( message.getBody(), StandardCharsets.UTF_8 ));
// 添加手動應答
/*
basicAck( long, boolean )
參數1、消息的標識tag，這個標識就相當於是消息的ID
參數2、是否批量應答multiple
*/
channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
} catch (InterruptedException e) {
e.printStackTrace();
}
};
System.out.println("消費者02正在接收消息，需要10秒處理完");
channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {
System.out.println("觸發消費者取消消費消息行為的回調");
System.out.println(Arrays.toString(consumerTag.getBytes(StandardCharsets.UTF_8)));
});
}
}
3.4、RabbitMQ的持久化 durable
3.4.1、隊列持久化
- 這個玩意兒的配置吧，早就見過了，在生產者消息發送時，有一個聲明隊列的過程，那裡面就有一個是否持久化的配置
/*
下面這是參數的完整意思，源碼中偷懶了，沒有見名知意
queueDeclare( queueName,isPersist,isShare,isAutoDelete,properties )
參數1、隊列名字
參數2、是否持久化( 保存到磁碟 ），默認是在記憶體中的
參數3、是否共享，即：是否只供一個消費者消費，是否讓多個消費者共享這個隊列中的資訊
參數4、是否自動刪除，即：最後一個消費者獲取資訊之後，這個隊列是否自動刪除
參數5、其他配置項，這涉及到後面的知識，目前選擇null
*/
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
- 而如果沒有持久化，那麼RabbitMQ服務由於其他什麼原因導致挂彩的時候，那麼重啟之後，這個沒有持久化的隊列就灰飛煙滅了【 ps：注意和裡面的消息還沒關係啊，不是說隊列持久化了，那麼消息就持久化了 】
- 在這個隊列持久化配置中，它的默認值就是false，所以要改成true時，需要注意一個點：選擇隊列持久化，那麼必須保證當前這個隊列是新的，即：RabbitMQ中沒有當前隊列，否則：需要進到web管理介面把已有的同名隊列刪了，然後重新配置當前隊列持久化選項為true，不然：報錯
- 那麼：當我把持久化選項改為true，並 重新發送消息時
inequivalent arg 'durable' for queue 'queue durable' in vhost '/': received 'true' but current is 'false'- 告知你：vhost虛擬機中已經有了這個叫做durable的隊列，要接收的選項值是true，但是它當前的值是false，所以報錯了唄
- 解決方式就是去web管理介面，把已有的durable隊列刪了，重新執行
- 再次執行就可以吃雞了，同時去web管理介面會發現它狀態變了，多了一個D標識
- 有了這個玩意兒之後，那麼就算RabbitMQ出問題了，後續恢復之後，那麼這個隊列也不會丟失
3.4.2、消息持久化
-
注意：這裡說的消息持久化不是說配置之後消息就一定不會丟失，而是：把消息標記為持久化，然後RabbitMQ盡量讓其持久化到磁碟
-
但是：也會有意外，比如：RabbitMQ在將消息持久化到磁碟時，這是有一個時間間隔的，數據還沒完全刷寫到磁碟呢，RabbitMQ萬一出問題了，那麼消息 / 數據還是會丟失的，所以：消息持久化配置是一個弱持久化，但是：對於簡單隊列模式完全足夠了，強持久化的實現方式在後續的publisher / confirm發布確認模式中
-
至於配置極其地簡單，在前面都已經見過這個配置項，就是生產者發消息時做文章，就是下面的第三個參數，把它改為
MessageProperties.PERSISTENT_TEXT_PLAIN即可
/*
basicPublish( exchangeName,routing key,properties,message )
參數1、交互機名字 - 使用了默認的
參數2、指定路由規則，使用隊列名字
參數3、指定傳遞的消息所攜帶的properties
參數4、推送的具體消息 - byte類型的
*/
channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
// 改成消息持久化
channel.basicPublish("",QUEUE_NAME,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());
- MessageProperties類的源碼如下：
public class MessageProperties {
public static final BasicProperties MINIMAL_BASIC = new BasicProperties((String)null, (String)null, (Map)null, (Integer)null, (Integer)null, (String)null, (String)null, (String)null, (String)null, (Date)null, (String)null, (String)null, (String)null, (String)null);
public static final BasicProperties MINIMAL_PERSISTENT_BASIC = new BasicProperties((String)null, (String)null, (Map)null, 2, (Integer)null, (String)null, (String)null, (String)null, (String)null, (Date)null, (String)null, (String)null, (String)null, (String)null);
public static final BasicProperties BASIC = new BasicProperties("application/octet-stream", (String)null, (Map)null, 1, 0, (String)null, (String)null, (String)null, (String)null, (Date)null, (String)null, (String)null, (String)null, (String)null);
public static final BasicProperties PERSISTENT_BASIC = new BasicProperties("application/octet-stream", (String)null, (Map)null, 2, 0, (String)null, (String)null, (String)null, (String)null, (Date)null, (String)null, (String)null, (String)null, (String)null);
public static final BasicProperties TEXT_PLAIN = new BasicProperties("text/plain", (String)null, (Map)null, 1, 0, (String)null, (String)null, (String)null, (String)null, (Date)null, (String)null, (String)null, (String)null, (String)null);
public static final BasicProperties PERSISTENT_TEXT_PLAIN = new BasicProperties("text/plain", (String)null, (Map)null, 2, 0, (String)null, (String)null, (String)null, (String)null, (Date)null, (String)null, (String)null, (String)null, (String)null);
public MessageProperties() {
}
}
- 上面用到了BasicProperties類型，它的屬性如下：
public static class BasicProperties extends AMQBasicProperties {
// 消息內容的類型
private String contentType;
// 消息內容的編碼格式
private String contentEncoding;
// 消息的header
private Map<String, Object> headers;
// 消息是否持久化，1：否，2：是
private Integer deliveryMode;
// 消息的優先順序
private Integer priority;
// 關聯ID
private String correlationId;
// :用於指定回復的隊列的名稱
private String replyTo;
// 消息的失效時間
private String expiration;
// 消息ID
private String messageId;
// 消息的發送時間
private Date timestamp;
// 類型
private String type;
// 用戶ID
private String userId;
// 應用程式ID
private String appId;
// 集群ID
private String clusterId;
}
3.5、不公平分發 和 預取值
不公平分發
- 這個東西是在消費者那一方進行設置的
- RabbitMQ默認是公平分發，即：輪詢分發
- 輪詢分發有缺點：如前面消費者01（ 設5秒的那個 ）和 消費者02 （ 設10秒的那個 ），這種情況如果採用輪詢分發，那麼：01要快一點，而02要慢一點，所以01很快處理完了，然後處於空閑狀態，而02還在拚命奮鬥中，最後的結果就是02不停干，而01悠悠閑閑的，浪費了時間，所以：應該壓榨一下01，讓它不能停
- 設置方式：在消費者接收消息之前進行
channel.basicQos（ int prefetchCount ）設置
// 不公平分發，就是在這裡接收消息之前做處理
/*
basicQos( int prefetchCount )
為0、輪詢分發 也是RabbitMQ的默認值
為1、不公平分發
*/
channel.basicQos(1);
channel.basicConsume("qos queue", true, deliverCallback, consumerTag -> {
System.out.println("消費者中斷了接收消息行為觸發的回調");
});
預取值
- 指的是：多個消費者在消費消息時，讓每一個消費者預計消費多少條消息
- 而要設置這種效果，和前面不公平分發的設置是一樣的，只是把裡面的參數改一下即可
// 預取值，也是在這裡接收消息之前做處理，和不公平分發調的是同一個API
/*
basicQos( int prefetchCount ) 為0、輪詢分發 也是RabbitMQ的默認值;為1、不公平分發
而當這裡的數字變成其他的，如：上圖中上面的那個消費者要消費20條消息，那麼把下面的數字改成對應的即可
注意點：這是要設置哪個消費者的預取值，那就是在哪個消費者程式碼中進行設定啊
*/
channel.basicQos(10); // 這樣就表示這個程式碼所在的消費者需要消費10條消息了
channel.basicConsume("qos queue", true, deliverCallback, consumerTag -> {
System.out.println("消費者中斷了接收消息行為觸發的回調");
});
3.6、publisher / confirms 發布確認模式
3.6.1、發布確認模式的原理
- 這個玩意兒的目的就是為了持久化
- 在上面的過程中，想要讓數據持久化，那麼需要具備以下的條件
- 1、隊列持久化
- 2、消息持久化
- 3、發布確認
- 而所謂的發布確認指的就是：數據在刷寫到磁碟時，成功了，那麼MQ就回復生產者一下，數據確認刷寫到磁碟了，否則：只具備前面的二者的話，那也有可能出問題，如：數據推到了隊列中，但是還沒來得及刷寫到磁碟呢，結果RabbitMQ宕機了，那數據也有可能會丟失,所以：現在持久化的過程就是如下的樣子：
開啟發布確認
- 在發送消息之前（ 即：調basicPublish（） 之前 ）調一個API就可以了
channel.confirmSelect(); // 沒有參數
3.6.2、發布確認的分類
3.6.2.1、單個確認發布
- 一句話：一手交錢一手交貨，即 生產者發布一條消息，RabbitMQ就要回復確認狀態，否則不再發放消息，因此：這種模式是同步發布確認的方式，缺點：很慢，優點：能夠實時地了解到那條消息出異常 / 哪些消息都發布成功了
public static void main(String[] args) throws InterruptedException, TimeoutException, IOException {
// 單個確認發布
singleConfirm(); // 單個確認發布發送這些消息花費4797ms
}
public static void singleConfirm() throws IOException, TimeoutException, InterruptedException {
Channel channel = MQUtil.getChannel("");
// 開啟確認發布
channel.confirmSelect();
// 聲明隊列 並 讓隊列持久化
channel.queueDeclare("singleConfirm", true, false, false, null);
long begin = System.currentTimeMillis();
for (int i = 1; i <= 100; i++) {
// 發送消息 並 讓消息持久化
channel.basicPublish("","singleConfirm", MessageProperties.PERSISTENT_TEXT_PLAIN,String.valueOf(i).getBytes() );
// 發布一個 確認一個 channel.waitForConfirms()
if ( channel.waitForConfirms() )
System.out.println("消息".concat( String.valueOf(i) ).concat( "發送成功") );
}
long end = System.currentTimeMillis();
System.out.println("單個確認發布發送這些消息花費".concat( String.valueOf( end-begin ) ).concat("ms") );
}
3.6.2.2、批量確認發布
- 一句話：只要結果，是怎麼一個批量管不著，只需要把一堆消息發布之後，回復一個結果即可，這種發布也是同步的
- 優點：效率相比單個發布要高
- 缺點：如果因為什麼系統故障而導致發布消息出現問題，那麼就會導致是批量發了一些消息，然後再回復的，中間有哪個消息出問題了鬼知道
public static void main(String[] args) throws InterruptedException, TimeoutException, IOException {
// 單個確認發布
// singleConfirm(); // 單個確認發布發送這些消息花費4797ms
// 批量發布
batchConfirm(); // 批量發布發送的消息共耗時：456ms
}
public static void batchConfirm() throws IOException, TimeoutException, InterruptedException {
Channel channel = MQUtil.getChannel("");
// 開啟確認發布
channel.confirmSelect();
// 聲明隊列 並 讓隊列持久化
channel.queueDeclare("batchConfirm", true, false, false, null);
long begin = System.currentTimeMillis();
for (int i = 1; i <= 100; i++) {
// 發送消息 並 讓消息持久化
channel.basicPublish("","batchConfirm", MessageProperties.PERSISTENT_TEXT_PLAIN,String.valueOf(i).getBytes() );
// 批量發布 並 回復批量發布的結果 - 發了10條之後再確認
if (i % 10 == 0) {
channel.waitForConfirms();
System.out.println("消息" + ( i-10 ) + "====>" + i + "的消息發布成功");
}
}
// 為了以防還有另外的消息未被確認，再次確認一下
channel.waitForConfirms();
long end = System.currentTimeMillis();
System.out.println("批量發布發送的消息共耗時：" + (end - begin) + "ms");
}
3.6.2.3、非同步確認發布 – 必須會的一種
- 由上圖可知：所謂的非同步確認發布就是：
- 1、生產者只管發消息就行，不用管消息有沒有成功
- 2、發布的消息是存在一個map集合中的，其key就是消息的標識tag / id，value就是消息內容
- 3、如果消息成功發布了，那麼實體broker會有一個ackCallback()回調函數來進行處理【 ps：裡面的處理邏輯是需要我們進行設計的 】
- 4、如果消息未成功發布，那麼實體broker會調用一個nackCallback()回調函數來進行處理【 ps：裡面的處理邏輯是需要我們進行設計的 】
- 5、而需要非同步處理，就是因為生產者只管發就行了，因此：一輪的消息肯定是很快就發布過去了，就可以做下一輪的事情了，至於上一輪的結果是怎麼樣的，那就需要等到兩個callback回調執行完了之後給結果，而想要能夠調取到兩個callback回調，那麼：就需要對發送的資訊進行監聽 / 對信道進行監聽
- 而上述牽扯到一個map集合，那麼這個集合需要具備如下的條件：
- 1、首先此集合應是一個安全且有序的，同時還支援高並發
- 2、其次能夠將序列號( key ) 和 消息( value )輕鬆地進行關聯
程式碼實現
public static void main(String[] args) throws InterruptedException, TimeoutException, IOException {
// 單個確認發布
// singleConfirm(); // 單個確認發布發送這些消息花費4797ms
// 批量發布
// batchConfirm(); // 批量發布發送的消息共耗時：456ms
asyncConfirm(); // 非同步發布確認耗時：10ms
}
// 非同步發布確認
public static void asyncConfirm() throws IOException, TimeoutException {
Channel channel = MQUtil.getChannel("");
channel.confirmSelect();
channel.queueDeclare("async confirm", true, false, false, null);
// 1、準備符合條件的map
ConcurrentSkipListMap<Long, Object> messagePoolMap = new ConcurrentSkipListMap<>();
// 3、對信道channel進行監聽
// 成功確認發布回調
ConfirmCallback ackCallback = (messageTag, multiple) -> {
System.out.println("確認發布了消息=====>" + messagePoolMap.headMap(messageTag) );
// 4、把確認發布的消息刪掉，減少記憶體開銷
// 判斷是否是批量刪除
if ( multiple ){
// 通過消息標識tag 把 確認發布的消息取出
messagePoolMap.headMap(messageTag).clear();
/**
* 上面這句程式碼拆分寫法
* ConcurrentNavigableMap<Long, Object> confirmed = messagePoolMap.headMap(messageTag);
* confirmed.clear();
*/
}else {
messagePoolMap.remove(messageTag);
}
};
// 沒成功發布確認回調
ConfirmCallback nackCallback = (messageTag, multiple) -> {
System.out.println("未確認的消息是：" + messagePoolMap.get(messageTag) );
};
// 進行channel監聽 這是非同步的
/**
* channel.addConfirmListener(ConfirmCallback var1, ConfirmCallback var2)
* 參數1、消息成功發布的回調函數 ackCallback()
* 參數2、消息未成功發布的回調函數 nackCallback()
*/
channel.addConfirmListener( ackCallback,nackCallback );
long begin = System.currentTimeMillis();
for (int i = 1; i <= 100; i++) {
// 2、將要發布的全部資訊保存到map中去
/*
channel.getNextPublishSeqNo() 獲取下一次將要發送的消息標識tag
*/
messagePoolMap.put(channel.getNextPublishSeqNo(),String.valueOf(i) );
// 生產者只管發布就行
channel.basicPublish("","async confirm",MessageProperties.PERSISTENT_TEXT_PLAIN,String.valueOf(i).getBytes());
System.out.println("消息=====>" + i + "發送完畢");
}
long end = System.currentTimeMillis();
System.out.println("非同步發布確認耗時：" + ( end-begin ) + "ms" );
}
3.7、交換機
- 正如前面一開始就畫的原理圖，交換機的作用就是為了接收生產者發送的消息 並 將消息發送到隊列中去
- 注意點：前面一直玩的那些模式，雖然沒有寫交換機，但並不是說RabbitMQ就沒用交換機【 ps：使用的是””空串，也就是使用了RabbitMQ的默認交換機 】，生產者發送的消息只能發到交換機中，從而由交換機來把消息發給隊列
3.7.1、交換機exchange的分類
- 直接( direct ) / 默認
- 主題( topic )
- 標題 ( heanders ) – 這個已經很少用了
- 扇出( fancut ) / 發布訂閱模式
臨時隊列
- 所謂的臨時隊列指的就是：自動幫我們生成隊列名 並且 當生產者和隊列斷開之後，這個隊列會被自動刪除
- 所以這麼一說：前面玩過的一種就屬於臨時隊列，即：將下面的第四個參數改成true即可【 ps：當然讓隊列名隨機生成就完全匹配了 】
/*
下面這是參數的完整意思，源碼中偷懶了，沒有見名知意
queueDeclare( queueName,isPersist,isShare,isAutoDelete,properties )
參數1、隊列名字
參數2、是否持久化( 保存到磁碟 ），默認是在記憶體中的
參數3、是否共享，即：是否只供一個消費者消費，是否讓多個消費者共享這個隊列中的資訊
參數4、是否自動刪除，即：最後一個消費者獲取資訊之後，這個隊列是否自動刪除
參數5、其他配置項，這涉及到後面的知識，目前選擇null
*/
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
- 而如果要更簡單的生成臨時隊列，那麼調用如下的API即可
String queueName = channel.queueDeclare().getQueue();
- 這樣幫我們生成的隊列效果就和
channel.queueDeclare(QUEUE_NAME, false, false, true, null);是一樣的了
3.7.2、fanout扇出 / 發布訂閱模式
- 這玩意兒吧，好比群發，一人發，很多人收到消息，就是原理圖的另一種樣子，生產者發布的一個消息，可以供多個消費者進行消費
- 實現方式就是讓一個交換機binding綁定多個隊列
生產者
package cn.zixieqing.fanout;
import cn.zixieqing.util.MQUtil;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Scanner;
import java.util.concurrent.TimeoutException;
public class FanoutProducer {
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = MQUtil.getChannel("");
/**
* 定義交換機
* 參數1、交換機名字
* 參數2、交換機類型
*/
channel.exchangeDeclare("fanoutExchange", BuiltinExchangeType.FANOUT);
System.out.println("請輸入要發送的內容：");
Scanner input = new Scanner(System.in);
while (input.hasNext()){
String message = input.next();
channel.basicPublish("fanoutExchange","", null,message.getBytes(StandardCharsets.UTF_8));
System.out.println("消息=====>" + message + "發送完畢");
}
}
}
消費者01
package cn.zixieqing.fanout;
import cn.zixieqing.util.MQUtil;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;
public class FanoutConsumer01 {
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = MQUtil.getChannel("");
// 綁定隊列
/**
* 參數1、隊列名字
* 參數2、交換機名字
* 參數3、用於綁定的routing key / binding key
*/
String queueName = channel.queueDeclare().getQueue();
channel.queueBind(queueName, "fanoutExchange", "");
System.out.println("01消費者正在接收消息........");
channel.basicConsume(queueName,true,(consumerTag,message)->{
// 這裡面接收到消息之後就可以用來做其他事情了，如：存到磁碟
System.out.println("接收到了消息====>" + new String( message.getBody(), StandardCharsets.UTF_8));
},consumerTage->{});
}
}
消費者02
package cn.zixieqing.fanout;
import cn.zixieqing.util.MQUtil;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;
public class FanoutConsumer02 {
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = MQUtil.getChannel("");
// 綁定隊列
/**
* 參數1、隊列名字
* 參數2、交換機名字
* 參數3、用於綁定的routing key / binding key
*/
String queueName = channel.queueDeclare().getQueue();
channel.queueBind(queueName, "fanoutExchange", "");
System.out.println("02消費者正在接收消息........");
channel.basicConsume(queueName,true,(consumerTag,message)->{
// 這裡面接收到消息之後就可以用來做其他事情了，如：存到磁碟
System.out.println("接收到了消息====>" + new String( message.getBody(), StandardCharsets.UTF_8));
},consumerTage->{});
}
}
3.7.3、direct交換機 / routing路由模式
- 這個玩意兒吧就是發布訂閱模式，也就是fanout類型交換機的變樣板，即：多了一個routing key的配置而已，也就是說：生產者和消費者傳輸消息就通過routing key進行關聯起來，因此：現在就變成了生產者想把消息發給誰就發給誰
生產者
package cn.zixieqing.direct;
import cn.zixieqing.util.MQUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Scanner;
import java.util.concurrent.TimeoutException;
public class DirectProducer {
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = MQUtil.getChannel("");
channel.exchangeDeclare("directExchange", BuiltinExchangeType.DIRECT);
System.out.println("請輸入要發送的消息：");
Scanner input = new Scanner(System.in);
while (input.hasNext()){
String message = input.next();
/**
* 對第二個參數routing key做文章
* 假如這裡的routing key為zixieqing 那麼：就意味著消費者只能是綁定了zixieqing的隊列才可以進行接收這裡發的消息內容
*/
channel.basicPublish("directExchange","zixieqing",null,message.getBytes(StandardCharsets.UTF_8));
System.out.println("消息=====>" + message + "====>發送完畢");
}
}
}
消費者01
package cn.zixieqing.direct;
import cn.zixieqing.util.MQUtil;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;
public class DirectConsumer01 {
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = MQUtil.getChannel("");
channel.queueDeclare("direct", false, false, false, null);
/**
* 隊列綁定
* 參數1、隊列名
* 參數2、交換機名字
* 參數3、routing key 這裡的routing key 就需要和生產者中的一樣了，這樣才可以通過這個routing key去對應的隊列中取消息
*/
channel.queueBind("direct", "directExchange", "zixieqing");
System.out.println("01消費者正在接收消息.......");
channel.basicConsume("direct",true,(consumerTag,message)->{
System.out.println("01消費者接收到了消息====>" + new String( message.getBody(), StandardCharsets.UTF_8));
},consumerTag->{});
}
}
- 上面這種，生產者的消息肯定能夠被01消費者給消費，因為：他們的交換機名字、隊列名字和routing key的值都是相同的
- 而此時再加一個消費者，讓它的routing key值和消費者中的不同
package cn.zixieqing.direct;
import cn.zixieqing.util.MQUtil;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;
public class DirectConsumer02 {
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = MQUtil.getChannel("");
channel.queueDeclare("direct", false, false, false, null);
/**
* 隊列綁定
* 參數1、隊列名
* 參數2、交換機名字
* 參數3、routing key 這裡的routing key 就需要和生產者中的一樣了，這樣才可以通過這個routing key去對應的隊列中取消息
*/
// 搞點事情：這裡的routing key的值zixieqing和生產者的不同
channel.queueBind("direct", "directExchange", "xiegongzi");
System.out.println("02消費者正在接收消息.......");
channel.basicConsume("direct",true,(consumerTag,message)->{
System.out.println("02消費者接收到了消息====>" + new String( message.getBody(), StandardCharsets.UTF_8));
},consumerTag->{});
}
}
3.7.4、topic交換機 / topic主題模式 – 使用最廣的一個
- 前面玩的fanout扇出類型的交換機 / 發布訂閱模式是一個生產者發布，多個消費者共享消息，和qq群類似；而direct直接交換機 / 路由模式是消費者只能消費和消費者相同routing key的消息
- 而上述這兩種還有局限性，如：現在生產者的routing key為zi.xie.qing，而一個消費者只消費含xie的消息，一個消費者只消費含qing的消息，另一個消費者只消費第一個為zi的零個或無數個單詞的消息，甚至還有一個消費者只消費最後一個單詞為qing，前面有三個單詞的routing key的消息呢？
- 這樣一看，發布訂閱模式和路由模式都不能解決，更別說前面玩的簡單模式、工作隊列模式、發布確認模式了，這些和目前的這個需求更不搭了，因此：就來了這個topic主題模式
topic中routing key的要求
- 只要交換機類型是topic類型的，那麼其routing key就不能亂寫，要求：routing key只能是一個單詞列表，多個單詞之間採用點隔開，如：cn.zixieqing.rabbit
- 單詞列表的長度不能超過255個位元組
- 在routing key的規則列表中有兩個替換符可以用
- 1、
*代表一個單詞 - 2、
#代表零活無數個單詞
- 1、
- 假如有如下的一個綁定關係圖
- Q1綁定的是：中間帶 orange 帶 3 個單詞的字元串(.orange.)
- Q2綁定的是：
- 最後一個單詞是 rabbit 的 3 個單詞(..rabbit)
- 第一個單詞是 lazy 的多個單詞(lazy.#)
- 熟悉一下這種綁定關係（ 左為一些routes路由規則，右為能匹配到上圖綁定關係的結果 ）
quick.orange.rabbit 被隊列 Q1Q2 接收到
lazy.orange.elephant 被隊列 Q1Q2 接收到
quick.orange.fox 被隊列 Q1 接收到
lazy.brown.fox 被隊列 Q2 接收到
lazy.pink.rabbit 雖然滿足兩個綁定，但只被隊列 Q2 接收一次
quick.brown.fox 不滿足任何綁定關係，不會被任何隊列接收到，會被丟棄
quick.orange.male.rabbit 是四個單詞，不滿足任何綁定關係，會被丟棄
lazy.orange.male.rabbit 雖是四個單詞，但匹配 Q2，因：符合lazy.#這個規則
- 當隊列綁定關係是下列這種情況時需要引起注意
- 當一個隊列綁定鍵是#，那麼這個隊列將接收所有數據，就有點像 fanout 了
- 如果隊列綁定鍵當中沒有#和*出現，那麼該隊列綁定類型就是 direct 了
把上面的綁定關係和測試轉換成程式碼玩一波
生產者
package cn.zixieqing.topic;
import cn.zixieqing.util.MQUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;
public class TopicProducer {
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = MQUtil.getChannel("");
channel.exchangeDeclare("topicExchange", BuiltinExchangeType.TOPIC);
/**
* 準備大量的routing key 和 message
*/
HashMap<String, String> routesAndMessageMap = new HashMap<>();
routesAndMessageMap.put("quick.orange.rabbit", "被隊列 Q1Q2 接收到");
routesAndMessageMap.put("lazy.orange.elephant", "被隊列 Q1Q2 接收到");
routesAndMessageMap.put("quick.orange.fox", "被隊列 Q1 接收到");
routesAndMessageMap.put("lazy.brown.fox", "被隊列 Q2 接收到");
routesAndMessageMap.put("lazy.pink.rabbit", "雖然滿足兩個綁定，但只被隊列 Q2 接收一次");
routesAndMessageMap.put("quick.brown.fox", "不滿足任何綁定關係，不會被任何隊列接收到，會被丟棄");
routesAndMessageMap.put("quick.orange.male.rabbit", "是四個單詞，不滿足任何綁定關係，會被丟棄");
routesAndMessageMap.put("lazy.orange.male.rabbit ", "雖是四個單詞，但匹配 Q2，因：符合lazy.#這個規則");
System.out.println("生產者正在發送消息.......");
for (Map.Entry<String, String> routesAndMessageEntry : routesAndMessageMap.entrySet()) {
String routingKey = routesAndMessageEntry.getKey();
String message = routesAndMessageEntry.getValue();
channel.basicPublish("topicExchange",routingKey,null,message.getBytes(StandardCharsets.UTF_8));
System.out.println("消息====>" + message + "===>發送完畢");
}
}
}
消費者01
package cn.zixieqing.topic;
import cn.zixieqing.util.MQUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;
public class TopicConsumer01 {
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = MQUtil.getChannel("");
channel.exchangeDeclare("topicExchange", BuiltinExchangeType.TOPIC);
channel.queueDeclare("Q1", false, false, false, null);
channel.queueBind("Q1", "topicExchange", "*.orange.*");
System.out.println("消費者01正在接收消息......");
channel.basicConsume("Q1",true,(consumerTage,message)->{
System.out.println("01消費者接收到了消息====>" + new String( message.getBody(), StandardCharsets.UTF_8));
System.out.println("此條消息的交換機名為：" + message.getEnvelope().getExchange() + "，路由鍵為：" + message.getEnvelope().getRoutingKey());
},consumerTag->{});
}
}
消費者02
package cn.zixieqing.topic;
import cn.zixieqing.util.MQUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;
public class TopicConsumer02 {
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = MQUtil.getChannel("");
channel.exchangeDeclare("topicExchange", BuiltinExchangeType.TOPIC);
channel.queueDeclare("Q2", false, false, false, null);
channel.queueBind("Q2", "topicExchange", "*.*.rabbit");
channel.queueBind("Q2", "topicExchange", "lazy.#");
System.out.println("消費者02正在接收消息......");
channel.basicConsume("Q2",true,(consumerTage,message)->{
System.out.println("02消費者接收到了消息====>" + new String( message.getBody(), StandardCharsets.UTF_8));
System.out.println("此條消息的交換機名為：" + message.getEnvelope().getExchange() + "，路由鍵為：" + message.getEnvelope().getRoutingKey());
},consumerTag->{});
}
}
3.8.2、隊列超過最大長度
3.8.2.1、隊列超過所限制的最大個數
- 意思就是：某一個隊列要求只能放N個消息，但是放了N+1個消息，這就超過隊列的最大個數了
生產者
- 就是一個正常的生產者發送消息而已
package cn.zixieqing.dead_letter_queue.queuelength.queuenumber;
import cn.zixieqing.util.MQUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;
public class Producer {
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = MQUtil.getChannel("");
channel.exchangeDeclare("messageNumber_normal_exchange", BuiltinExchangeType.DIRECT);
for (int i = 1; i < 11; i++) {
String message = "生產者發送了消息" + i;
channel.basicPublish("messageNumber_normal_exchange","zi",null,
message.getBytes(StandardCharsets.UTF_8) );
System.out.println("消息====>" + message + "====>發送完畢");
}
}
}
01消費者
package cn.zixieqing.dead_letter_queue.queuelength.queuenumber;
import cn.zixieqing.util.MQUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.concurrent.TimeoutException;
public class Consumer01 {
/**
* 正常交換機名稱
*/
public static final String NORMAL_EXCHANGE = "messageNumber_normal_exchange";
/**
* 正常隊列名稱
*/
public static final String NORMAL_QUEUE = "messageNumber_queue";
/**
* 死信交換機名稱
*/
public static final String DEAD_EXCHANGE = "messageNumber_dead_exchange";
/**
* 死信隊列名稱
*/
public static final String DEAD_QUEUE = "messageNumber_dead_queue";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = MQUtil.getChannel("");
// 聲明正常交換機、死信交換機
channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
// 聲明死信隊列
channel.queueDeclare(DEAD_QUEUE, false, false, false, null);
// 死信交換機和死信隊列進行綁定
channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "xie");
// 聲明正常隊列 並 考慮達到條件時和死信交換機進行聯繫
HashMap<String, Object> params = new HashMap<>();
// 死信交換機
params.put("x-dead-letter-exchange", DEAD_EXCHANGE);
// 死信路由鍵
params.put("x-dead-letter-routing-key", "xie");
// 達到隊列能接受的最大個數限制就多了如下的配置
params.put("x-max-length", 6);
channel.queueDeclare(NORMAL_QUEUE, false, false, false, params);
// 正常隊列和正常交換機進行綁定
channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "zi");
System.out.println("01消費者正在接收消息......");
channel.basicConsume(NORMAL_QUEUE,true,(consumeTag,message)->{
System.out.println("01消費者接收到了消息：" + new String( message.getBody(), StandardCharsets.UTF_8));
},consumeTag->{});
}
}
- 啟動01消費者，然後關掉（ 模仿異常 ），最後啟動生產者，那麼：生產者發送了10個消息，由於01消費者這邊做了配置，所以有6個消息是在正常隊列中，餘下的4個消息就會進入死信隊列
3.8.2.2、超過隊列能接受消息的最大位元組長度
- 和前面一種相比，在01消費者方做另一個配置即可
params.put("x-max-length-bytes", 255);
注意：關於兩種情況同時使用的問題
- 如配置的如下兩個
params.put("x-max-length", 6);
params.put("x-max-length-bytes", 255);
- 那麼先達到哪個上限設置就執行哪個
3.8.3、消息被拒收
- 注意點：必須開啟手動應答
// 第二個參數改成false
channel.basicConsume(NORMAL_QUEUE,false,(consumeTag,message)->{},consumeTag->{});
生產者
package cn.zixieqing.dead_letter_queue.reack;
import cn.zixieqing.util.MQUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;
public class Producer {
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = MQUtil.getChannel("");
channel.exchangeDeclare("reack_normal_exchange", BuiltinExchangeType.DIRECT);
for (int i = 1; i < 11; i++) {
String message = "生產者發送的消息" + i;
channel.basicPublish("reack_normal_exchange","zixieqing",null,message.getBytes(StandardCharsets.UTF_8));
System.out.println("消息===>" + message + "===>發送完畢");
}
}
}
消費者
package cn.zixieqing.dead_letter_queue.reack;
import cn.zixieqing.util.MQUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.concurrent.TimeoutException;
public class Consumer01 {
public static final String NORMAL_EXCHANGE = "reack_normal_exchange";
public static final String DEAD_EXCHANGE = "reack_dead_exchange";
public static final String DEAD_QUEUE = "reack_dead_queue";
public static final String NORMAL_QUEUE = "reack_normal_queue";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = MQUtil.getChannel("");
// 聲明正常交換機、死信交換機
channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
// 聲明死信隊列
channel.queueDeclare(DEAD_QUEUE, false, false, false, null);
// 死信隊列綁定死信交換機
channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "xie");
// 聲明正常隊列
HashMap<String, Object> params = new HashMap<>();
params.put("x-dead-letter-exchange", DEAD_EXCHANGE);
params.put("x-dead-letter-routing-key", "xie");
channel.queueDeclare(NORMAL_QUEUE, false, false, false, params);
channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "zixieqing");
System.out.println("01消費者正在接收消息.....");
// 1、注意：需要開啟手動應答（ 第二個參數為false ）
channel.basicConsume(NORMAL_QUEUE,false,(consumeTag,message)->{
String msg = new String(message.getBody(), StandardCharsets.UTF_8);
// 如果發送的消息為：生產者發送的消息5 則：拒收
if ( "生產者發送的消息5".equals( msg ) ) {
System.out.println("此消息====>" + msg + "===>是拒收的");
// 2、做拒收處理 - 注意：第二個參數設為false，表示不再重新入正常隊列的隊，這樣消息才可以進入死信隊列
channel.basicReject( message.getEnvelope().getDeliveryTag(),false);
}else {
System.out.println("01消費者接收到了消息=====>" + msg);
}
},consumeTag->{});
}
}
