当前位置：首页 > news >正文

RabbitMQ 发布确认模式

news 2025/7/14 4:04:34

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");}