Java整合RabbitMQ實(shí)現(xiàn)五種常見消費(fèi)模型
引言
RabbitMQ是一個(gè)流行的消息隊(duì)列中間件，它確保了不同應(yīng)用程序之間的可靠消息傳遞。由于其高性能、輕量級(jí)和靈活性，RabbitMQ在許多應(yīng)用程序中被廣泛使用，例如異步任務(wù)處理、負(fù)載均衡、事件通知 等。在RabbitMQ中，消息的生產(chǎn)和消費(fèi)是通過一系列的消費(fèi)模型來管理的。每個(gè)消費(fèi)模型都有不同的特點(diǎn)和應(yīng)用場(chǎng)景，可以幫助開發(fā)人員構(gòu)建高效的消息傳遞系統(tǒng)。本文將深入介紹RabbitMQ的五種常見消費(fèi)模型，包括簡單隊(duì)列模型、工作隊(duì)列模型、發(fā)布/訂閱模型、路由模型和主題模型，刪除線格式 并探討它們各自的優(yōu)缺點(diǎn)和適用場(chǎng)景。希望此文能幫助你更好地理解RabbitMQ消費(fèi)模型并在實(shí)踐中達(dá)到更好的效果。
1. 簡單隊(duì)列模型（Simple Queue Model）
簡單隊(duì)列模型是最基礎(chǔ)的RabbitMQ模型。它包括單個(gè)生產(chǎn)者和單個(gè)消費(fèi)者。生產(chǎn)者將消息發(fā)送到一個(gè)隊(duì)列中，然后消費(fèi)者從隊(duì)列中讀取消息并處理。這種模式不適用于多個(gè)消費(fèi)者或消息廣播，因?yàn)橐坏┫⒈灰粋€(gè)消費(fèi)者接收，它就會(huì)從隊(duì)列中刪除。
優(yōu)缺點(diǎn)及適用場(chǎng)景
- 優(yōu)點(diǎn)：
實(shí)現(xiàn)簡單，易于理解和部署。
可以提供一些基本的可靠性保證，例如消息確認(rèn)和持久化。
- 缺點(diǎn)：
不支持并發(fā)消費(fèi)。
不支持多個(gè)消費(fèi)者共同消費(fèi)一個(gè)隊(duì)列。
- 適用場(chǎng)景：
單生產(chǎn)者和單消費(fèi)者之間的點(diǎn)對(duì)點(diǎn)通信。
系統(tǒng)中只有一個(gè)進(jìn)程或線程可以處理消息。
例如，一個(gè)后端服務(wù)向另一個(gè)后端服務(wù)發(fā)送消息，或者一個(gè)客戶端將任務(wù)發(fā)送給服務(wù)器
代碼示例
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
// 聲明隊(duì)列
String queueName = "simpleQueue";
channel.queueDeclare(queueName, false, false, false, null);
// 發(fā)送消息
String message = "Hello, RabbitMQ!";
channel.basicPublish("", queueName, null, message.getBytes("UTF-8"));
System.out.println("Sent message: " + message);
// 接收消息
boolean autoAck = true;
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println("Received message: " + message);
}
};
channel.basicConsume(queueName, autoAck, consumer);
2. 工作隊(duì)列模型（Work Queue Model）
工作隊(duì)列模型允許多個(gè)消費(fèi)者協(xié)同地從一個(gè)隊(duì)列中接收、處理和分發(fā)消息。在這種模型中，消息被平均分配給不同的消費(fèi)者。當(dāng)一個(gè)消費(fèi)者正在處理一個(gè)消息時(shí)，它不能接收新的消息。這確保了公平的分布和消費(fèi)，同時(shí)在不同的消費(fèi)者之間進(jìn)行負(fù)載均衡。
優(yōu)缺點(diǎn)及適用場(chǎng)景
- 優(yōu)點(diǎn)：
支持多個(gè)消費(fèi)者處理同一個(gè)隊(duì)列中的消息。
消費(fèi)負(fù)載均衡，每個(gè)消費(fèi)者最多處理一條消息。
通過設(shè)置并發(fā)數(shù)，可以實(shí)現(xiàn)更高的消息吞吐量。
- 缺點(diǎn)：
沒有消息路由的動(dòng)態(tài)性。
如果有消息時(shí)，所有的消費(fèi)者都會(huì)在接收到該消息后進(jìn)行同樣的處理，無法根據(jù)具體情況進(jìn)行消息的劃分，而且消息被平均分配，不能根據(jù)消息的重要性和緊急性進(jìn)行處理。
- 適用場(chǎng)景：
需要在多個(gè)工人之間分配任務(wù)的應(yīng)用程序，例如異步任務(wù)處理或負(fù)載均衡。
代碼示例
// 創(chuàng)建連接和通道
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
// 聲明隊(duì)列
String queueName = "workQueue";
channel.queueDeclare(queueName, false, false, false, null);
// 發(fā)送消息
String message = "Hello, RabbitMQ!";
channel.basicPublish("", queueName, null, message.getBytes("UTF-8"));
System.out.println("Sent message: " + message);
// 設(shè)置每個(gè)消費(fèi)者在接收到確認(rèn)前只能處理一條消息
channel.basicQos(1);
// 接收消息
boolean autoAck = false;
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println("Received message: " + message);
// 手動(dòng)發(fā)送消息確認(rèn)
channel.basicAck(envelope.getDeliveryTag(), false);
}
};
channel.basicConsume(queueName, autoAck, consumer);
3. 發(fā)布/訂閱模型（Publish/Subscribe Model）
發(fā)布/訂閱模型允許一個(gè)生產(chǎn)者向多個(gè)消費(fèi)者廣播一條消息。在這種模型中，生產(chǎn)者將消息發(fā)送到一個(gè)交換機(jī)中，然后這個(gè)交換機(jī)將消息路由到所有與之綁定的隊(duì)列。每個(gè)隊(duì)列對(duì)應(yīng)一個(gè)消費(fèi)者，可以獨(dú)立地處理這個(gè)隊(duì)列中的消息。
優(yōu)缺點(diǎn)及適用場(chǎng)景
- 優(yōu)點(diǎn)：
支持廣播式消息發(fā)布和訂閱。
與其他應(yīng)用程序解耦，生產(chǎn)者和消費(fèi)者不需要知道對(duì)方的存在和細(xì)節(jié)。
- 缺點(diǎn)：
不支持消息路由的動(dòng)態(tài)性。
沒有消息過濾機(jī)制，每個(gè)訂閱者都會(huì)收到所有的消息。
- 適用場(chǎng)景：
需要將消息通知多個(gè)消費(fèi)者的應(yīng)用程序，例如事件通知或新聞發(fā)布。
代碼示例
// 創(chuàng)建連接和通道
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
// 聲明交換器
String exchangeName = "publishSubscribeExchange";
channel.exchangeDeclare(exchangeName, "fanout");
// 創(chuàng)建隨機(jī)隊(duì)列并綁定到交換器
String queueName = channel.queueDeclare().getQueue();
channel.queueBind(queueName, exchangeName, "");
// 發(fā)送消息
String message = "Hello, RabbitMQ!";
channel.basicPublish(exchangeName, "", null, message.getBytes("UTF-8"));
System.out.println("Sent message: " + message);
// 接收消息
boolean autoAck = true;
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println("Received message: " + message);
}
};
channel.basicConsume(queueName, autoAck, consumer);
4. 路由模型（Routing Model）
路由模型允許生產(chǎn)者根據(jù)路由鍵將消息發(fā)送到指定的隊(duì)列中。在這種模型中，交換機(jī)會(huì)將消息路由到與它所綁定的隊(duì)列匹配的路由鍵的隊(duì)列中。消費(fèi)者可以從這些隊(duì)列中接收和處理消息。
優(yōu)缺點(diǎn)及適用場(chǎng)景
- 優(yōu)點(diǎn)：
支持基于路由鍵的動(dòng)態(tài)消息路由。
可以根據(jù)消息的類型、內(nèi)容和優(yōu)先級(jí)選擇發(fā)送給哪個(gè)隊(duì)列，支持消息的定向投遞。
- 缺點(diǎn)：
需要提前配置好交換機(jī)和隊(duì)列之間的綁定關(guān)系。
支持的路由邏輯有限，只能通過路由鍵進(jìn)行匹配。
- 適用場(chǎng)景：
需要根據(jù)某些特定屬性或條件將消息路由到相應(yīng)隊(duì)列的應(yīng)用程序，例如日志記錄或按優(yōu)先級(jí)處理任務(wù)。
代碼示例
// 創(chuàng)建連接和通道
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
// 聲明交換器
String exchangeName = "routingExchange";
channel.exchangeDeclare(exchangeName, "direct");
// 綁定隊(duì)列到交換器，并指定路由鍵
String queueName = "routingQueue";
String routingKey = "important";
channel.queueDeclare(queueName, false, false, false, null);
channel.queueBind(queueName, exchangeName, routingKey);
// 發(fā)送消息
String message = "Important message!";
channel.basicPublish(exchangeName, routingKey, null, message.getBytes("UTF-8"));
System.out.println("Sent message: " + message);
// 接收消息
boolean autoAck = true;
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println("Received message: " + message);
}
};
channel.basicConsume(queueName, autoAck, consumer);
5. 主題模型（Topic Model）
主題模型是路由模型的擴(kuò)展，它可以實(shí)現(xiàn)更靈活的消息路由和分發(fā)。在這種模型中，生產(chǎn)者可以使用通配符匹配來匹配路由鍵。交換機(jī)會(huì)將消息路由到與它所綁定的隊(duì)列匹配的路由鍵的隊(duì)列中。消費(fèi)者可以從這些隊(duì)列中接收和處理消息。
優(yōu)缺點(diǎn)及適用場(chǎng)景
- 優(yōu)點(diǎn)：
支持更靈活、更具體的消息路由和過濾。
可以使用通配符匹配路由鍵，實(shí)現(xiàn)更復(fù)雜的消息匹配和分發(fā)。
- 缺點(diǎn)：
高度配置化和復(fù)雜化，需要額外配置主題模式下的應(yīng)用程序邏輯。
在一些場(chǎng)景下，通配符匹配路由鍵可能會(huì)導(dǎo)致性能問題。
- 適用場(chǎng)景：
需要根據(jù)消息內(nèi)容的模式將消息路由到不同隊(duì)列的應(yīng)用程序，例如按標(biāo)簽或關(guān)鍵字分發(fā)和處理不同的任務(wù)。
代碼示例
// 創(chuàng)建連接和通道
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
// 聲明交換器
String exchangeName = "topicExchange";
channel.exchangeDeclare(exchangeName, "topic");
// 綁定隊(duì)列到交換器，并指定匹配模式
String queueName = "topicQueue";
String routingKeyPattern = "com.example.*";
channel.queueDeclare(queueName, false, false, false, null);
channel.queueBind(queueName, exchangeName, routingKeyPattern);
// 發(fā)送消息
String routingKey = "com.example.news";
String message = "Important news!";
channel.basicPublish(exchangeName, routingKey, null, message.getBytes("UTF-8"));
System.out.println("Sent message: " + message);
// 接收消息
boolean autoAck = true;
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println("Received message: " + message);
}
};
channel.basicConsume(queueName, autoAck, consumer);
總的來說，不同的消息隊(duì)列模型適用于不同的場(chǎng)景和需求。簡單隊(duì)列模型適合于點(diǎn)對(duì)點(diǎn)通信；工作隊(duì)列模型適用于任務(wù)分配和負(fù)載均衡；發(fā)布/訂閱模型適用于消息廣播和解耦；路由模型適用于動(dòng)態(tài)消息路由和選擇性投遞；主題模型適用于靈活的消息路由和過濾。根據(jù)具體的業(yè)務(wù)需求和系統(tǒng)架構(gòu)，合理選擇適用的消息隊(duì)列模型可以提高系統(tǒng)的可擴(kuò)展性、可靠性和性能。
到此這篇關(guān)于Java整合RabbitMQ實(shí)現(xiàn)五種常見消費(fèi)模型的文章就介紹到這了,更多相關(guān)Java RabbitMQ消費(fèi)模型內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
Jedis是Redis官方推薦的用于訪問Java客戶端，下面這篇文章主要給大家介紹了關(guān)于Java客戶端利用Jedis操作redis緩存的相關(guān)資料，文中給出了詳細(xì)的示例代碼，需要的朋友可以參考借鑒，下面來一起看看吧。2017-07-07
這篇文章主要為大家詳細(xì)介紹了java使用GUI實(shí)現(xiàn)貪吃蛇游戲，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2022-04-04
sharding-jdbc 兼容 MybatisPlus動(dòng)態(tài)數(shù)據(jù)源的配置方法
這篇文章主要介紹了sharding-jdbc 兼容 MybatisPlus動(dòng)態(tài)數(shù)據(jù)源的配置方法,本文通過實(shí)例代碼給大家介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友參考下吧2024-07-07
Java虛擬機(jī)JVM之server模式與client模式的區(qū)別
這篇文章主要介紹了Java虛擬機(jī)JVM的client模式和Server模式兩者的區(qū)別和聯(lián)系2017-12-12
Java實(shí)現(xiàn)讀取及生成Excel文件的方法
這篇文章主要介紹了Java實(shí)現(xiàn)讀取及生成Excel文件的方法,結(jié)合實(shí)例形式分析了java通過引入第三方j(luò)ar包poi-3.0.1-FINAL-20070705.jar實(shí)現(xiàn)針對(duì)Excel文件的讀取及生成功能,需要的朋友可以參考下2017-12-12
Java的泛型擦除和運(yùn)行時(shí)泛型信息獲取方式
Java泛型在編譯時(shí)會(huì)發(fā)生類型擦除,即泛型參數(shù)被替換為它們的限定類型（如Object）,這使得ArrayList<Integer>和ArrayList<String>在運(yùn)行時(shí)類型相同,盡管如此,我們可以通過定義類或匿名內(nèi)部類的方式在運(yùn)行時(shí)獲取泛型信息2024-09-09
這篇文章主要介紹了SpringBoot注解@CrossOrigin使用詳解,@CrossOrigin是用來處理跨域請(qǐng)求的注解2023-12-12
跨域,指的是瀏覽器不能執(zhí)行其他網(wǎng)站的腳本,它是由瀏覽器的同源策略造成的,是瀏覽器對(duì)JavaScript施加的安全限制,需要的朋友可以參考下
這篇文章主要介紹了java使用smslib連接短信貓發(fā)關(guān)短信代碼,需要的朋友可以參考下2014-02-02
Java設(shè)計(jì)模式之組合模式（Composite模式）介紹
這篇文章主要介紹了Java設(shè)計(jì)模式之組合模式（Composite模式）介紹,Composite定義：將對(duì)象以樹形結(jié)構(gòu)組織起來,以達(dá)成“部分－整體” 的層次結(jié)構(gòu),使得客戶端對(duì)單個(gè)對(duì)象和組合對(duì)象的使用具有一致性,需要的朋友可以參考下2015-03-03
Java裝飾器設(shè)計(jì)模式_動(dòng)力節(jié)點(diǎn)Java學(xué)院整理
這篇文章主要介紹了Java裝飾器設(shè)計(jì)模式的相關(guān)資料，非常不錯(cuò)，具有參考借鑒價(jià)值,需要的朋友可以參考下2017-05-05