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

RabbitMQ深入 —— 持久化和发布确认

news 2025/7/16 10:55:07

前言

前面的文章荔枝梳理了如何去配置RabbitMQ环境并且也介绍了两种比较简单的运行模式，在这篇文章中荔枝将会继续梳理有关RabbitMQ的持久化机制以及发布确认模式的相关知识，希望能够帮助到大家~~~

文章目录

前言

一、持久化

1.1 队列持久化

1.2 消息持久化

1.3 不公平分发

1.4 预取值

二、发布确认机制

2.1 单个发布确认

2.2 批量发布确认

2.3 异步发布确认

2.4 处理异步未确认消息的机制

总结

一、持久化

跟MySQL和Redis一样，持久化的操作就是为了保存数据，避免因为RabbitMQ宕机停止服务之后可以确保消息不丢失。默认情况下RabbitMQ不会开启持久化，他会忽视队列和消息。

1.1 队列持久化

队列的持久化比较简单，只需要在生产者创建信道的时候将queueDeclare()函数的durable参数设置为ture即可。如果之前就已经创建过该队列，修改完durable之后可能会报错，这时候仅需要管理后台中删除队列即可。

boolean durable = ture;
//信道申明
channel.queueDeclare(TASK_QUEUE_ACK, durable ,false,false,null);

1.2 消息持久化

消息持久化的实现也需要在消息生产者中修改demo，只需要在basicPublish()中设置第三个参数为MessageProperties.PERSISTENT_TEXT_PLAIN。

channel.basicPublish("",TASK_QUEUE_ACK, MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes("UTF-8"));

需要注意的是：只有消息持久化并不能完全保证不会丢失消息。尽管它告诉RabbitMQ将消息保存到磁盘，但是这里依然存在当消息刚准备存储在磁盘的时候但是还没有存储完，消息还在缓存的一个间隔点。此时并没有真正写入磁盘。持久性保证并不强，但是对于我们的简单任务队列而言已经足够，如果需要更强地保证持久化，可能要用到发布确认模式。

1.3 不公平分发

默认情况下RabbitMQ对队列中的消息采用的是轮询分发的机制，但这种机制会造成资源浪费，因为处理消息的快的消费者更多的时间是处于空闲状态。不公平分发的机制开启比较简单，只需要在消费者处的代码设置信道时，设置其basicQos的值为1即可。

channel.basicQos(1);

1.4 预取值

由于消息的确认和发送都是异步的，因此消费者处会有一个未确认消息的缓冲区，为了使得该缓冲区不会无限制地增大，可以通过basic.qos来设置消费者未确认消息的缓冲区中允许的未确认的消息的最大数量。一旦达到最大数量就会RabbitMQ就会停止在通道上传递更多消息直至至少一个未被确认的消息被确认处理。这时候为了尽量避免自动分发和手动分发模式下无限制确认消息缓冲区所造成了消费者的RAM消耗，我们需要根据需求场景设置一个比较好的预取值。

//预取值是10
int prefetchCount = 10;
channel.basicQos(prefetchCount);

需要注意的是预取值的设置是在消费者一方！

二、发布确认机制

在前面我们设置了队列持久化和消息持久化，但是单纯开启持久化并不能保证完全持久化，消息有可能还没来得及保存在磁盘中就发生了RabbitMQ宕机，因此我们还需要引入发布确认模式。

在消息发布者处开启发布确认模式

//开启发布确认模式
channel.confirmSelect();

下面我们来看看几种确认发布的模式：

2.1 单个发布确认

单个发布确认是一种简单的确认方式，它是一种同步确认发布的方式。也就是发布一个消息之后只有它被确认发布，后续的消息才能继续发布，waitForConfirms()这个方法只有在消息被确认的时候才返回，如果在指定时间范围内这个消息没有被确认那么它将抛出异常。

缺点：发布速度特别的慢，因为如果没有确认发布的消息就会阻塞所有后续消息的发布，这种方式最多提供每秒不超过数百条发布消息的吞吐量。

    //单个确认发布public static void publishMessageIndividually() throws Exception{Channel channel = RabbitMqUtil.getChannel();//随机生成队列声明String queueName = UUID.randomUUID().toString();channel.queueDeclare(queueName,false,false,false,null);//开启发布确认channel.confirmSelect();//开始时间long begin = System.currentTimeMillis();//循环发布消息for(int i=0;i<MESSAGE_COUNT;i++){String message = i+"";channel.basicPublish("",queueName,null,message.getBytes());boolean flag = channel.waitForConfirms();if(flag){System.out.println("消息发送成功"+i);}}long end = System.currentTimeMillis();long ttl = end - begin;System.out.println("单次发布模式下发送"+MESSAGE_COUNT+"次消息的耗时："+ttl+"ms");}

2.2 批量发布确认

批量消息确认发布是指不再是像单个发布确认的模式那样发一条消息等待确认后再发一条，而是一次性发送一批消息之后再统一确认发布。批量发布确认确实可以极大的提高消息队列的吞吐量，但缺点却是不清楚哪个消息出现问题。

    //批量确认发布public static void publishMessageBatch() throws Exception {Channel channel = RabbitMqUtil.getChannel();//随机生成队列声明String queueName = UUID.randomUUID().toString();channel.queueDeclare(queueName,false,false,false,null);//开启发布确认channel.confirmSelect();//开始时间long begin = System.currentTimeMillis();//批量操作的信息大小int batchSize = 10;for (int i=0;i<MESSAGE_COUNT;i++){String message = i+"";channel.basicPublish("",queueName,null,message.getBytes());//判断是否达到sizeif (i%batchSize==0){channel.waitForConfirms();}}long end = System.currentTimeMillis();long ttl = end - begin;System.out.println("批量确认发布模式下发送消息"+MESSAGE_COUNT+"次消息的耗时："+ttl+"ms");}

2.3 异步发布确认

相比于前面两种发布确认模式，异步确认发布在实现起来更加复杂，但是性能和效率更高。它是通过回调函数来达到消息可靠性传递和保证消息投递成功的。在异步发布确认模式下一般至少有三个线程，一个线程是用来发送消息的；另外两个线程分别是处理消息确认成功和消息确认失败的回调函数。

    //异步确认发布public static void publishMessageAsync() throws Exception{Channel channel = RabbitMqUtil.getChannel();//随机生成队列声明String queueName = UUID.randomUUID().toString();channel.queueDeclare(queueName,false,false,false,null);//开启发布确认channel.confirmSelect();//开始时间long begin = System.currentTimeMillis();//消息确认成功的回调函数ConfirmCallback ackCallback = (deliveryTag,multiple)->{System.out.println("确认的消息："+deliveryTag);};//消息确认失败的回调函数(消息的标记，是否为批量处理)ConfirmCallback nackCallback = (deliveryTag,multiple)->{System.out.println("未确认的消息："+deliveryTag);};//设置消息监听器channel.addConfirmListener(ackCallback,nackCallback); //这个监听是异步的//消息发送for(int i = 0;i<MESSAGE_COUNT;i++){String message = i+"";channel.basicPublish("",queueName,null,message.getBytes());}//结束时间long end = System.currentTimeMillis();long ttl = end - begin;System.out.println("异步确认发布模式下发送消息"+MESSAGE_COUNT+"次消息的耗时："+ttl+"ms");}

2.4 处理异步未确认消息的机制

处理异步未确认消息的最好解决方案就是把未确认的消息放到一个基于内存的能被发布线程访问的队列，比如说用ConcurrentSkipListMap这个队列在confirm callbacks与发布线程之间进行消息的传递。这里为了存储发送的消息体我们使用的是ConcurrentSkipListMap这个类型对象，该类对象是线程安全的有序的哈希表，适用于高并发的场景。

public static void publishMessageAsync() throws Exception{Channel channel = RabbitMqUtil.getChannel();//随机生成队列声明String queueName = UUID.randomUUID().toString();channel.queueDeclare(queueName,false,false,false,null);//开启发布确认channel.confirmSelect();//开始时间long begin = System.currentTimeMillis();/*** ConcurrentSkipListMap线程安全有序的一个哈希表，适用于高并发的场景下* 1.轻松的关联序号和消息* 2.轻松批量删除条目* 3.支持高并发场景*/ConcurrentSkipListMap<Long,String> outstandingConfirm = new ConcurrentSkipListMap<>();//消息确认成功的回调函数ConfirmCallback ackCallback = (deliveryTag,multiple)->{//删除掉已经确认的消息 剩下未确认的消息//判断是不是批量删除if(multiple){ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirm.headMap(deliveryTag);confirmed.clear();}else {outstandingConfirm.remove(deliveryTag);}System.out.println("确认的消息："+deliveryTag);};//消息确认失败的回调函数(消息的标记，是否为批量处理)ConfirmCallback nackCallback = (deliveryTag,multiple)->{//打印未确认的消息String message = outstandingConfirm.get(deliveryTag);System.out.println("未确认消息是"+message+"未确认的消息的标签："+deliveryTag);};//设置消息监听器channel.addConfirmListener(ackCallback,nackCallback); //这个监听是异步的//消息发送for(int i = 0;i<MESSAGE_COUNT;i++){String message = i+"";channel.basicPublish("",queueName,null,message.getBytes());//记录下所有的发送消息outstandingConfirm.put(channel.getNextPublishSeqNo(), message);}//结束时间long end = System.currentTimeMillis();long ttl = end - begin;System.out.println("异步确认发布模式下发送消息"+MESSAGE_COUNT+"次消息的耗时："+ttl+"ms");}

outstandingConfirm.headMap(deliveryTag)是为了在批量操作中做一些过滤，返回小于deliveryTag的映射并调用clear方法清空outstandingConfirm 映射中所有小于 deliveryTag 的键值对，而不影响其他键值对。这里其实就是清除所有已经确认的消息，留下未确认的消息。