Java消息隊列RabbitMQ之消息回調(diào)詳解
更新時間：2023年07月31日 09:43:39 作者：迷鹿小女子
這篇文章主要介紹了Java消息隊列RabbitMQ之消息回調(diào)詳解,消息回調(diào)，其實就是消息確認(生產(chǎn)者推送消息成功，消費者接收消息成功) , 對于程序來說，發(fā)送者沒法確認是否發(fā)送成功,需要的朋友可以參考下
消息100%的投遞
消息如何保障100%的投遞成功？ 什么是生產(chǎn)端的可靠性投遞？
- 保障消息的成功發(fā)出
- 保障MQ節(jié)點的成功接收
- 發(fā)送端收到MQ節(jié)點Broker確認應(yīng)答
- 完善的消息進行補償機制
BAT/TMD互聯(lián)網(wǎng)大廠的解決方案：
- 消息落庫，對消息狀態(tài)進行打標
- 消息的延遲投遞，做二次確認，回調(diào)檢查
冪等性概念
冪等性是什么？
- 我們可以借鑒數(shù)據(jù)庫的樂觀鎖機制
- 比如我們執(zhí)行一條更新庫存的SQL語句
- pdate t_repository set cont = cont -1,version = version + 1 where version = 1
- Elasticsearch也是嚴格遵循冪等性概念，每次數(shù)據(jù)更新，version+1
消費端-冪等性保障
在海量訂單產(chǎn)生的業(yè)務(wù)高峰期，如何避免消息的重復(fù)消費問題？
消費實現(xiàn)冪等性，就意味著，我們的消息永遠不會消費多次，即使我們收到了多條一樣的消息
業(yè)界主流的冪等性操作
唯一ID+指紋碼機制，利用數(shù)據(jù)庫主鍵去重 利用Redis的原子性去實現(xiàn)
- 唯一ID+指紋碼機制，利用數(shù)據(jù)庫主鍵去重
- Select count(1) from T_order where ID=唯一ID+指紋碼
好處：實現(xiàn)簡單
壞處：高并發(fā)下有數(shù)據(jù)庫寫入的性能瓶頸
解決方案：根據(jù)ID進行分庫分表進行算法路由
利用Redis的原子性去實現(xiàn)
使用Redis進行冪等，需要考慮的問題
第一：我們是否要進行數(shù)據(jù)落庫，如果落庫的話，關(guān)鍵解決的問題是數(shù)據(jù)庫和緩存如何做到原子性？
第二：如果不進行落庫，那么都存儲到緩存中，如何設(shè)置定時同步策略？
Confirm確認消息
理解Confirm消息確認機制 消息的確認，是指生產(chǎn)者投遞消息后，如果Broker收到消息，則會給我們生產(chǎn)者 一個應(yīng)答。
生產(chǎn)者進行接收應(yīng)答，用來確定這條消息是否正常的發(fā)送到Broker，這種方式也是消息的可靠性投遞的核心保障
如何實現(xiàn)Confirm確認消息？
第一步：在Channel上開啟確認模式：channel.confirmSelect()
第二步：在channel上添加監(jiān)聽：addConfirmListener，監(jiān)聽成功和失敗的返回結(jié)果，根據(jù)具體的結(jié)果對消息進行重新發(fā)送、或記錄日志等后續(xù)處理！
消費端代碼
package com.xieminglu.rabbitmqapi.confirm;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
public class Consumer {
public static void main(String[] args) throws Exception {
//1 創(chuàng)建ConnectionFactory
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("192.168.248.132");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/");
//2 獲取C onnection
Connection connection = connectionFactory.newConnection();
//3 通過Connection創(chuàng)建一個新的Channel
Channel channel = connection.createChannel();
String exchangeName = "test_confirm_exchange";
String routingKey = "confirm.#";
String queueName = "test_confirm_queue";
//4 聲明交換機和隊列 然后進行綁定設(shè)置, 最后制定路由Key
channel.exchangeDeclare(exchangeName, "topic", true);
channel.queueDeclare(queueName, true, false, false, null);
channel.queueBind(queueName, exchangeName, routingKey);
//5 創(chuàng)建消費者
QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
channel.basicConsume(queueName, true, queueingConsumer);
while(true){
QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
String msg = new String(delivery.getBody());
System.err.println("消費端: " + msg);
}
}
}服務(wù)提供方代碼
package com.xieminglu.rabbitmqapi.confirm;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
public class Producer {
public static void main(String[] args) throws Exception {
//1 創(chuàng)建ConnectionFactory
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("192.168.248.132");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/");
//2 獲取C onnection
Connection connection = connectionFactory.newConnection();
//3 通過Connection創(chuàng)建一個新的Channel
Channel channel = connection.createChannel();
//4 指定我們的消息投遞模式: 消息的確認模式
channel.confirmSelect();
String exchangeName = "test_confirm_exchange";
String routingKey = "confirm.save";
//5 發(fā)送一條消息
String msg = "Hello RabbitMQ Send confirm message!";
channel.basicPublish(exchangeName, routingKey, null, msg.getBytes());
//6 添加一個確認監(jiān)聽
channel.addConfirmListener(new ConfirmListener() {
@Override
public void handleNack(long deliveryTag, boolean multiple) throws IOException {
System.err.println("-------no ack!-----------");
}
@Override
public void handleAck(long deliveryTag, boolean multiple) throws IOException {
System.err.println("-------ack!-----------");
}
});
}
}Retrn返回消息
Retrn Listener用于處理一些不可路由的消息！
- 正常情況：我們的消息生產(chǎn)者，通過指定一個Exchange和RotingKey，把消息送達到某一個隊列中去，然后我們的消費者監(jiān)聽隊列，進行消費處理操作！
- 異常情況：在某些情況下，如果我們在發(fā)送消息的時候，當前的Exchange不存在或者指定的路由key路由不到，這個時候如果我們需要監(jiān)聽這種不可達的消息，就需要使用Retrn Listener！
在基礎(chǔ)API中有一個關(guān)鍵的配置項 Mandatory：如果為tre，則監(jiān)聽器會接收到路由不可達的消息，然后進行后續(xù)處理，如果為false，那么Broker端自動刪除該消息！
消費端代碼
package com.xieminglu.rabbitmqapi.returnlistener;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
public class Consumer {
public static void main(String[] args) throws Exception {
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("192.168.248.132");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/");
Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();
String exchangeName = "test_return_exchange";
String routingKey = "return.#";
String queueName = "test_return_queue";
channel.exchangeDeclare(exchangeName, "topic", true, false, null);
channel.queueDeclare(queueName, true, false, false, null);
channel.queueBind(queueName, exchangeName, routingKey);
QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
channel.basicConsume(queueName, true, queueingConsumer);
while(true){
QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
String msg = new String(delivery.getBody());
System.err.println("消費者: " + msg);
}
}
}生產(chǎn)端代碼
package com.xieminglu.rabbitmqapi.returnlistener;
import com.rabbitmq.client.*;
import java.io.IOException;
public class Producer {
public static void main(String[] args) throws Exception {
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("192.168.248.132");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/");
Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();
String exchange = "test_return_exchange";
String routingKey = "return.save";
String routingKeyError = "abc.save";
String msg = "Hello RabbitMQ Return Message";
channel.addReturnListener(new ReturnListener() {
@Override
public void handleReturn(int replyCode, String replyText, String exchange,
String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.err.println("---------handle return----------");
System.err.println("replyCode: " + replyCode);
System.err.println("replyText: " + replyText);
System.err.println("exchange: " + exchange);
System.err.println("routingKey: " + routingKey);
System.err.println("properties: " + properties);
System.err.println("body: " + new String(body));
}
});
//消息投遞成功，會被消費者所消費
channel.basicPublish(exchange, routingKey, true, null, msg.getBytes());
//消息不可達，將觸發(fā)ReturnListener
// channel.basicPublish(exchange, routingKeyError, true, null, msg.getBytes());
}
}自定義消費者
我們一般就是在代碼中編寫while循環(huán)，進行consmer.nextDelivery方法進行獲取下一條消息，然后進行消費處理！
但是我們使用自定義的Consmer更加的方便，解耦性更加的強，也是實際工作中最常用的使用方式！
自定義消費端代碼
package com.xieminglu.rabbitmqapi.consumer;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import java.io.IOException;
public class MyConsumer extends DefaultConsumer {
public MyConsumer(Channel channel) {
super(channel);
}
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.err.println("-----------consume message----------");
System.err.println("consumerTag: " + consumerTag);
System.err.println("envelope: " + envelope);
System.err.println("properties: " + properties);
System.err.println("body: " + new String(body));
}
}消費端調(diào)用
package com.xieminglu.rabbitmqapi.consumer;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class Consumer {
public static void main(String[] args) throws Exception {
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("192.168.248.132");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/");
Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();
String exchangeName = "test_consumer_exchange";
String routingKey = "consumer.#";
String queueName = "test_consumer_queue";
channel.exchangeDeclare(exchangeName, "topic", true, false, null);
channel.queueDeclare(queueName, true, false, false, null);
channel.queueBind(queueName, exchangeName, routingKey);
channel.basicConsume(queueName, true, new MyConsumer(channel));
}
}生產(chǎn)端調(diào)用
package com.xieminglu.rabbitmqapi.consumer;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class Producer {
public static void main(String[] args) throws Exception {
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("192.168.248.132");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/");
Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();
String exchange = "test_consumer_exchange";
String routingKey = "consumer.save";
String msg = "Hello RabbitMQ Consumer Message";
for(int i =0; i<5; i ++){
channel.basicPublish(exchange, routingKey, true, null, msg.getBytes());
}
}
}到此這篇關(guān)于Java消息隊列RabbitMQ之消息回調(diào)詳解的文章就介紹到這了,更多相關(guān)RabbitMQ消息回調(diào)內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
在Java堆內(nèi)存中,年輕代和老年代之間存在的對象相互引用,假設(shè)現(xiàn)在要進行一次新生代的YGC,但新生代中的對象可能被老年代所引用的,為了找到新生代中的存活對象,不得不遍歷整個老年代,這樣明顯效率很低下,那么如何快速識別并回收這種引用對象呢2024-02-02
Java實現(xiàn)生成n個不重復(fù)的隨機數(shù)
這篇文章主要為大家詳細介紹了Java實現(xiàn)生成n個不重復(fù)的隨機數(shù)，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2020-05-05
java如何實現(xiàn)基于opencv全景圖合成實例代碼
全景圖相信大家應(yīng)該都不陌生，下面這篇文章主要給大家介紹了關(guān)于java如何實現(xiàn)基于opencv全景圖合成的相關(guān)資料，文中通過示例代碼介紹的非常詳細，需要的朋友可以參考借鑒，下面隨著小編來一起學習學習吧2018-07-07
這篇文章主要介紹了Java 你知道什么是耦合、如何解(降低)耦合的操作，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-09-09
Java中存在基礎(chǔ)數(shù)據(jù)類型，但是在某些情況下，我們要對基礎(chǔ)數(shù)據(jù)類型進行對象的操作，例如，集合中只能存對象，而不能存在基礎(chǔ)數(shù)據(jù)類型，于是便出現(xiàn)了封裝類，本文將詳細給大家介紹Java封裝類和裝箱拆箱,需要的朋友可以參考下2023-05-05
關(guān)于spring中單例Bean引用原型Bean產(chǎn)生的問題及解決
這篇文章主要介紹了關(guān)于spring中單例Bean引用原型Bean產(chǎn)生的問題及解決方案，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-06-06
springboot Quartz動態(tài)修改cron表達式的方法
這篇文章主要介紹了springboot Quartz動態(tài)修改cron表達式的方法，小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2018-09-09
Java實現(xiàn)將PDF轉(zhuǎn)換為Word的示例詳解
在日常的數(shù)據(jù)處理、文檔編輯和系統(tǒng)集成工作中,將不可編輯的PDF文檔轉(zhuǎn)換為可編輯的Word文檔是一項常見且重要的需求,下面我們就來看看如何使用java實現(xiàn)這一操作吧2025-08-08
Spring Task定時任務(wù)的實現(xiàn)詳解
這篇文章主要介紹了SpringBoot定時任務(wù)功能詳細解析,這次的功能開發(fā)過程中也算是對其內(nèi)涵的進一步了解，以后遇到定時任務(wù)的處理也更清晰，更有效率了，對SpringBoot定時任務(wù)相關(guān)知識感興趣的朋友一起看看吧2022-08-08
Spring2.5在2.0的基于Schema的Bean配置的基礎(chǔ)之上，再增加了擴展XML配置的機制。通過該機制，我們可以編寫自己的Schema，并根據(jù)自定義的Schema用自定的標簽配置Bean，下面這篇文章主要介紹了關(guān)于Spring中自定義Schema如何解析生效的相關(guān)資料，需要的朋友可以參考下2018-07-07