RabbitMQ 发布确认模式
RabbitMQ 发布确认模式
一、原理
RabbitMQ 的发布确认模式（Publisher Confirms）是一种机制，用于确保消息在被 RabbitMQ 服务器成功接收后，发布者能够获得确认。这一机制在高可用性和可靠性场景下尤为重要，能够有效防止消息丢失。
在发布消息时，RabbitMQ 会向发布者返回确认信号，指示该消息已成功处理。这一过程可以大大提高消息传递的可靠性，尤其是在系统出现故障时。通过实现发布确认模式，发布者可以确保所有发送的消息都已被服务器接收并存储，从而维护数据一致性。
二、三种方式比较
1. 单个确认
单个确认模式是指每发送一条消息，发布者都会等待 RabbitMQ 发送确认信号。这种方式简单易懂，但在高负载场景下，性能表现可能较差，因为每条消息都需要等待确认，造成延迟。
优点：
- 实现简单，易于理解和使用。
- 适合消息量较小的场景。
缺点：
- 性能较低，确认过程中的延迟会影响整体吞吐量。
public static void publishMessageIndividually() throws Exception {
Channel channel = RabbitMqUtils.getChannel();
String queue_name = UUID.randomUUID().toString();
channel.queueDeclare(queue_name,false,false,false,null);
//开始时间
long begin = System.currentTimeMillis();
channel.confirmSelect();
for (int i = 0; i < message_count; i++) {
String message = i+"";
channel.basicPublish("",queue_name,null,message.getBytes());
//单个消息马上确认
boolean flag = channel.waitForConfirms();
if(flag == true){
System.out.println("消息发送成功");
}
}
//结束时间
long end = System.currentTimeMillis();
System.out.println("发布"+message_count+"条用时："+(end - begin)+"mills");
}
2. 批量确认
批量确认模式允许发布者在发送多条消息后，一次性等待 RabbitMQ 的确认信号。通过将多条消息组合在一起进行确认，可以显著提高消息发布的效率。
优点：
- 显著提高吞吐量，减少网络延迟。
- 适合需要高性能的场景。
缺点：
-
需要管理确认逻辑，稍微复杂。
-
一旦发生错误，可能需要重新发送整个批次的消息。
//批量发布确认
public static void publishMessageBatch() throws Exception {
Channel channel = RabbitMqUtils.getChannel();
String queue_name = UUID.randomUUID().toString();
channel.queueDeclare(queue_name,false,false,false,null);
channel.confirmSelect();
//开始时间
long begin = System.currentTimeMillis();
//批量确认的长度
int basicSize = 100;
for (int i = 1; i <= message_count; i++) {
String message = i+"";
channel.basicPublish("",queue_name,null,message.getBytes());
if(i%basicSize == 0){
System.out.println("消息发送成功");
channel.confirmSelect();
}
}
//结束时间
long end = System.currentTimeMillis();
System.out.println("批量发布"+message_count+"条用时："+(end - begin)+"mills");
}
3. 异步批量确认
异步批量确认模式更进一步，发布者在发送批量消息后，不必等待 RabbitMQ 的确认信号，而是可以继续发送更多消息。RabbitMQ 在后台处理确认，这样可以进一步提高效率。
优点：
- 最高的性能和吞吐量，几乎没有延迟。
- 发布者可以充分利用 CPU 和网络资源。
缺点：
- 实现复杂度增加，需要处理可能出现的消息丢失。
- 需要额外的逻辑来确保消息的可靠性。
//异步发布确认
public static void publishMessageAsync() throws Exception {
Channel channel = RabbitMqUtils.getChannel();
String queue_name = UUID.randomUUID().toString();
channel.queueDeclare(queue_name,false,false,false,null);
//开启发布确认
channel.confirmSelect();
/**
* 线程安全有序的哈希表 适用于高并发
* 1、将序号和消息关联
* 2、轻松删除条目，主要给到序号
* 3、支持高并发
*/
ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();
//开始时间
long begin = System.currentTimeMillis();
//消息确认成功的回调
ConfirmCallback ackCallBack = (deliveryTag, multiple)->{
//2:删除确认的消息
if(multiple){
ConcurrentNavigableMap<Long,String> confirmed =
outstandingConfirms.headMap(deliveryTag);
confirmed.clear();
}else {
outstandingConfirms.remove(deliveryTag);
}
System.out.println("确认消息："+deliveryTag);
};
/**
* 1、消息类型
* 2、是否批量
*/
ConfirmCallback NackCallBack = (deliveryTag, multiple)->{
//3:打印未确认的
String message = outstandingConfirms.get(deliveryTag);
System.out.println("发布的消息"+message+"未被确认，序列号"+deliveryTag);
};
//准备消息的监听器
//1.监听成功的
//2.监听失败的
channel.addConfirmListener(ackCallBack,NackCallBack);
for (int i = 1; i <= message_count; i++) {
String message = i+"";
channel.basicPublish("",queue_name,null,message.getBytes());
//1:此处记录所有要发送的消息
outstandingConfirms.put(channel.getNextPublishSeqNo(),message);
}
//结束时间
long end = System.currentTimeMillis();
System.out.println("批量发布"+message_count+"条用时："+(end - begin)+"mills");
}
三、速度对比
三种时间对比：
在速度和吞吐量方面，三种模式各有优势。一般来说，单个确认模式的速度最慢，而异步批量确认模式则能提供最佳性能。在实际应用中，选择合适的确认模式应根据业务需求、消息量以及系统的可承受负载来综合考虑。
发布确认模式为 RabbitMQ 提供了灵活的消息发布保障机制。根据具体需求选择合适的确认方式，将有助于实现高效、可靠的消息传递系统。
原文地址：https://blog.csdn.net/dhrmt/article/details/143083003
免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！