Docker啟動RabbitMQ實現(xiàn)生產者與消費者的詳細過程
更新時間：2023年02月23日 10:08:26 作者：zoeil
這篇文章主要介紹了Docker啟動RabbitMQ，實現(xiàn)生產者與消費者,通過Docker拉取鏡像并啟動RabbitMQ，本文結合實例代碼給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下
一、Docker拉取鏡像并啟動RabbitMQ
拉取鏡像
docker pull rabbitmq:3.8.8-management
查看鏡像
docker images rabbitmq
啟動鏡像
docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3.8.8-management
Linux虛擬機記得開放5672端口或者關閉防火墻，在window通過 主機ip:15672 訪問rabbitmq控制臺
用戶名密碼默認為guest
二、Hello World
（一）依賴導入
<!--指定 jdk 編譯版本-->
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>8</source>
<target>8</target>
</configuration>
</plugin>
</plugins>
</build>
<dependencies>
<!--rabbitmq 依賴客戶端-->
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.8.0</version>
</dependency>
<!--操作文件流的一個依賴-->
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
<version>2.6</version>
</dependency>
</dependencies>（二）消息生產者
工作原理
- Broker：接收和分發(fā)消息的應用，RabbitMQ Server 就是 Message Broker
- Connection：publisher／consumer 和 broker 之間的 TCP 連接
- Channel：如果每一次訪問 RabbitMQ 都建立一個 Connection，在消息量大的時候建立 TCP Connection 的開銷將是巨大的，效率也較低。Channel 是在 connection 內部建立的邏輯連接，如果應用程序支持多線程，通常每個 thread 創(chuàng)建單獨的 channel 進行通訊，AMQP method 包含了 channel id 幫助客戶端和 message broker 識別 channel，所以 channel 之間是完全隔離的。Channel 作為輕量級的 Connection 極大減少了操作系統(tǒng)建立 TCP connection 的開銷
- Exchange：message 到達 broker 的第一站，根據(jù)分發(fā)規(guī)則，匹配查詢表中的 routing key，分發(fā)消息到 queue 中去。常用的類型有：direct (point-to-point), topic (publish-subscribe) and fanout (multicast)
- Queue：消息最終被送到這里等待 consumer 取走
我們需要先獲取連接(Connection)，然后通過連接獲取信道(Channel)，這里我們演示簡單例子，可以直接跳過交換機(Exchange)發(fā)送隊列(Queue)
public class Producer {
private final static String QUEUE_NAME = "hello";
public static void main(String[] args) throws Exception {
//創(chuàng)建一個連接工廠
ConnectionFactory factory = new ConnectionFactory();
// 設置主機ip
factory.setHost("182.92.234.71");
// 設置用戶名
factory.setUsername("guest");
// 設置密碼
factory.setPassword("guest");
//channel 實現(xiàn)了自動 close 接口 自動關閉 不需要顯示關閉
Connection connection = factory.newConnection();
// 獲取信道
Channel channel = connection.createChannel();
/*
* 生成一個隊列
* queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,
Map<String, Object> arguments)
* 1.隊列名稱
* 2.隊列里面的消息是否持久化 默認消息存儲在內存中
* 3.該隊列是否只供一個消費者進行消費 是否進行共享 true 可以多個消費者消費
* 4.是否自動刪除 最后一個消費者端開連接以后 該隊列是否自動刪除 true 自動刪除
* 5.其他參數(shù)
**/
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
String message = "hello rabbitmq";
/*
* 發(fā)送一個消息
* basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)
* 1.發(fā)送到哪個交換機
* 2.路由的key是哪個
* 3.其他的參數(shù)信息
* 4.發(fā)送消息的消息體
*
**/
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println("發(fā)送成功");
}
}
（三）消息消費者
public class Consumer {
private static final String QUEUE_NAME = "hello";
public static void main(String[] args) throws Exception {
//創(chuàng)建一個連接工廠
ConnectionFactory factory = new ConnectionFactory();
// 設置主機ip
factory.setHost("182.92.234.71");
// 設置用戶名
factory.setUsername("guest");
// 設置密碼
factory.setPassword("guest");
//channel 實現(xiàn)了自動 close 接口 自動關閉 不需要顯示關閉
Connection connection = factory.newConnection();
// 獲取信道
Channel channel = connection.createChannel();
// 推送的消息如何進行消費的回調接口
DeliverCallback deliverCallback = (consumerTag, message) -> {
System.out.println(new String(message.getBody()));
};
// 取消消費的一個回調接口，如在消費的時候隊列被刪除了
CancelCallback cancelCallback = (consumerTag) -> {
System.out.println("消息消費被中斷");
};
/*
* 消費者消費消息
* basicConsume(String queue, boolean autoAck,
* DeliverCallback deliverCallback, CancelCallback cancelCallback)
* 1.消費哪個隊列
* 2.消費成功之后是否要自動應答 true 代表自動應答 false 手動應答
* 3.消費者未成功消費的回調
**/
channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);
}
}
三、實現(xiàn)輪訓分發(fā)消息
（一）抽取工具類
可以發(fā)現(xiàn)，上面獲取連接工廠，然后獲取連接，再獲取信道的步驟是一致的，我們可以抽取成一個工具類來調用，并使用單例模式-餓漢式完成信道的初始化
public class RabbitMqUtils {
private static Channel channel;
static {
ConnectionFactory factory = new ConnectionFactory();
// 設置ip地址
factory.setHost("192.168.23.100");
// 設置用戶名
factory.setUsername("guest");
// 設置密碼
factory.setPassword("guest");
try {
// 創(chuàng)建連接
Connection connection = factory.newConnection();
// 獲取信道
channel = connection.createChannel();
} catch (Exception e) {
System.out.println("創(chuàng)建信道失敗，錯誤信息：" + e.getMessage());
}
}
public static Channel getChannel() {
return channel;
}
}（二）啟動兩個工作線程
相當于前面的消費者，我們只需要寫一個類，通過ideal實現(xiàn)多線程啟動即可模擬兩個線程
public class Worker01 {
private final static String QUEUE_NAME = "hello";
public static void main(String[] args) throws Exception {
Channel channel = RabbitMqUtils.getChannel();
DeliverCallback deliverCallback = ( consumerTag, message) -> {
System.out.println("接受到消息：" + new String(message.getBody()));
};
CancelCallback cancelCallback = (cunsumerTag) -> {
System.out.println("消費者取消消費接口回調邏輯");
};
// 啟動兩次，第一次為C1， 第二次為C2
System.out.println("C2消費者等待消費消息");
channel.basicConsume(QUEUE_NAME, true, deliverCallback,cancelCallback);
}
}（三）啟動發(fā)送線程
public class Test01 {
private final static String QUEUE_NAME = "hello";
public static void main(String[] args) throws IOException {
Channel channel = RabbitMqUtils.getChannel();
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 通過控制臺輸入充當消息，使輪訓演示更明顯
Scanner scanner = new Scanner(System.in);
while(scanner.hasNext()) {
String message = scanner.next();
channel.basicPublish("", QUEUE_NAME,null, message.getBytes() );
System.out.println("消息發(fā)送完成：" + message);
}
}
}結果
四、實現(xiàn)手動應答
（一）消息應答概念
消費者完成一個任務可能需要一段時間，如果其中一個消費者處理一個長的任務并僅只完成 了部分突然它掛掉了，會發(fā)生什么情況。RabbitMQ 一旦向消費者傳遞了一條消息，便立即將該消 息標記為刪除。在這種情況下，突然有個消費者掛掉了，我們將丟失正在處理的消息。以及后續(xù) 發(fā)送給該消費這的消息，因為它無法接收到。 為了保證消息在發(fā)送過程中不丟失，rabbitmq 引入消息應答機制，消息應答就是: 消費者在接 收到消息并且處理該消息之后，告訴 rabbitmq 它已經處理了，rabbitmq 可以把該消息刪除了。
自動應答：消費者發(fā)送后立即被認為已經傳送成功。這種模式需要在高吞吐量和數(shù)據(jù)傳輸安全性方面做權衡,因為這種模式如果消息在接收到之前，消費者那邊出現(xiàn)連接或者 channel 關閉，那么消息就丟失了。
當然另一方面這種模式消費者那邊可以傳遞過載的消息， 沒有對傳遞的消息數(shù)量進行限制 ， 當然這樣有可能使得消費者這邊由于接收太多還來不及處理的消息，導致這些消息的積壓，最終 使得內存耗盡，最終這些消費者線程被操作系統(tǒng)殺死，所以這種模式僅適用在消費者可以高效并 以某種速率能夠處理這些消息的情況下使用 。
手動應答：消費者接受到消息并順利完成業(yè)務后再調用方法進行確認，rabbitmq 才可以把該消息刪除
（二）消息應答的方法
- Channel.basicAck(用于肯定確認)
- RabbitMQ 已知道該消息并且成功的處理消息，可以將其丟棄了
- Channel.basicNack(用于否定確認)
- Channel.basicReject(用于否定確認)
- 與 Channel.basicNack 相比少一個參數(shù)Multiple
- multiple 的 true 和 false 代表不同意思
true 代表批量應答 channel 上未應答的消息
比如說 channel 上有傳送 tag 的消息 5,6,7,8 當前 tag 是 8 那么此時
5-8 的這些還未應答的消息都會被確認收到消息應答
false 同上面相比
只會應答 tag=8 的消息 5,6,7 這三個消息依然不會被確認收到消息應答
- 不處理該消息了直接拒絕，可以將其丟棄了
（三）消息自動重新入隊
如果消費者由于某些原因失去連接(其通道已關閉，連接已關閉或 TCP 連接丟失)，導致消息未發(fā)送 ACK 確認，RabbitMQ 將了解到消息未完全處理，并將對其重新排隊。如果此時其他消費者可以處理，它將很快將其重新分發(fā)給另一個消費者。這樣，即使某個消費者偶爾死亡，也可以確 保不會丟失任何消息。
（四）消息手動應答代碼
1、生產者
public class Test01 {
private final static String QUEUE_NAME = "ack";
public static void main(String[] args) throws IOException {
Channel channel = RabbitMqUtils.getChannel();
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
Scanner scanner = new Scanner(System.in);
while(scanner.hasNext()) {
String message = scanner.next();
channel.basicPublish("", QUEUE_NAME,null, message.getBytes() );
System.out.println("消息發(fā)送完成：" + message);
}
}
}2、睡眠工具類模擬業(yè)務執(zhí)行
public class SleepUtils {
public static void sleep(int second) {
try {
Thread.sleep(1000 * second);
} catch (InterruptedException _ignored) {
Thread.currentThread().interrupt();
}
}
}3、消費者
public class Worker01 {
private final static String QUEUE_NAME = "ack";
public static void main(String[] args) throws Exception {
System.out.println("C1，業(yè)務時間短");
Channel channel = RabbitMqUtils.getChannel();
DeliverCallback deliverCallback = ( consumerTag, message) -> {
SleepUtils.sleep(1); // 模擬業(yè)務執(zhí)行1秒
System.out.println("接受到消息：" + new String(message.getBody()));
/*
* 1、消息標識
* 2、是否啟動批量確認，false：否。
* 啟用批量有可能造成消息丟失，比如未消費的消息提前被確然刪除，后面業(yè)務消費該消息
* 時出現(xiàn)異常會導致該消息的丟失
*/
channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
};
CancelCallback cancelCallback = (cunsumerTag) -> {
System.out.println("消費者取消消費接口回調邏輯");
};
boolean autoAck = false;
channel.basicConsume(QUEUE_NAME, autoAck, deliverCallback,cancelCallback);
}
}
==============================================================================
public class Worker02 {
private final static String QUEUE_NAME = "ack";
public static void main(String[] args) throws Exception {
System.out.println("C2，業(yè)務時間長");
Channel channel = RabbitMqUtils.getChannel();
DeliverCallback deliverCallback = ( consumerTag, message) -> {
SleepUtils.sleep(15); // 模擬業(yè)務執(zhí)行15秒
System.out.println("接受到消息：" + new String(message.getBody()));
/*
* 1、消息標識
* 2、是否啟動批量確認，false：否。
* 啟用批量有可能造成消息丟失，比如未消費的消息提前被確然刪除，后面業(yè)務消費該消息
* 時出現(xiàn)異常會導致該消息的丟失
*/
channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
};
CancelCallback cancelCallback = (cunsumerTag) -> {
System.out.println("消費者取消消費接口回調邏輯");
};
boolean autoAck = false;
channel.basicConsume(QUEUE_NAME, autoAck, deliverCallback,cancelCallback);
}
}
worker01業(yè)務時間短，worker02業(yè)務時間長，我們提前終止worker02模擬出異常，可以看到消息dd會被放回隊列由worker01接收處理。
注意：這里需要先啟動生產者聲明隊列ack，不然啟動消費者會報錯
最后一個案例我們可以看到消息輪訓+消息自動重新入隊+手動應答。
到此這篇關于Docker啟動RabbitMQ，實現(xiàn)生產者與消費者的文章就介紹到這了,更多相關Docker啟動RabbitMQ內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
詳解如何使用Docker快速部署ELK環(huán)境(最新5.5.1版本)
這篇文章主要介紹了詳解如何使用Docker快速部署ELK環(huán)境(最新5.5.1版本),小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2018-08-08
Redmine?是一個開源的，基于Web的項目管理和缺陷跟蹤工具，本文主要介紹了Docker?redmine項目管理工具的使用，具有一定的參考價值，感興趣的可以了解一下2022-01-01
這篇文章主要為大家介紹了docker容器的創(chuàng)建及常用操作命令詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步早日升職加薪2022-04-04
docker?build運行報錯source:?not?found解決分析
這篇文章主要為大家介紹了docker?build運行報錯source:?not?found解決分析，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪2023-09-09
這篇文章主要介紹了docker運行nginx鏡像的實現(xiàn),并將配置文件和目錄掛載到宿主機上,以實現(xiàn)方便統(tǒng)一的管理配置信息,感興趣的可以了解一下2023-10-10
rocketmq-streams的ILeaseService使用示例詳解
這篇文章主要為大家介紹了rocketmq-streams的ILeaseService使用示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪2023-07-07
docker/docker desktop for window環(huán)境下創(chuàng)建本地鏡像倉庫方式
這篇文章主要介紹了docker/docker desktop for window環(huán)境下創(chuàng)建本地鏡像倉庫方式,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2025-03-03
這篇文章主要介紹了Docker安裝logstash的詳細過程,本文給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2021-09-09
文章介紹了如何在Docker中清理volumes以釋放磁盤空間,包括查看所有卷、查找未使用的卷、清理未使用的卷、手動刪除特定卷、檢查卷的使用情況、清理卷中的數(shù)據(jù)、預防措施以及遷移Docker數(shù)據(jù)目錄等步驟,感興趣的朋友一起看看吧2025-03-03
這篇文章主要介紹了docker離線安裝并導入鏡像過程,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2023-12-12