思维导图

• 一，分析数据丢失的原因

  分析RabbitMQ消息丢失的情况，不妨先看看一条消息从生产者发送到消费者消费的过程：

  

  可以修剪，一条消息整个过程要经历两次的网络传输：从生产者发送到RabbitMQ服务器，从RabbitMQ服务器发送到消费者。

  在消费者未消费前存储在人数（队列）中。

  所以可以知道，有三个场景下是会发生消息丢失的：

  • 存储在特定中，如果物理没有对消息持久化，RabbitMQ服务器停机机重启会丢失数据。
  • 生产者发送消息到RabbitMQ服务器过程中，RabbitMQ服务器如果停机机停止服务，消息会丢失。
  • 消费者从兔子MQ服务器获取了中存储的数据消费，但消费者程序错误或停机，而没有正确消费，导致数据丢失。

  针对以上场景，RabbitMQ提供了三种解决的方式，分别是消息持久化，确认机制，ACK事务机制。

  

  二，消息持久化

  RabbitMQ是支持消息持久化的，消息持久化需要设置：Exchange为持久化和队列持久化，这样当消息发送到RabbitMQ服务器时，消息就会持久化。

  首先看Exchange交换机的类图：

  

  看这个类图实际上是要要说明上一篇文章介绍的布局开关都是AbstractExchange抽象类的子类，所以根据java的特性，创建子类的实例会先调用父类的构造器，父类也就是AbstractExchange的构造器是怎么样的呢？

  

  从上面的注释可以看到持久参数表示是否永久化。或者是持久化（true）。创建持久化的Exchange可以这样写：

      @Beanpublic DirectExchange rabbitmqDemoDirectExchange() {//Direct交换机return new DirectExchange(RabbitMQConfig.RABBITMQ_DEMO_DIRECT_EXCHANGE, true, false);}
  复制代码
  

  接着是Queue，我们先看看Queue的构造器是怎么样的：

  

  也是通过durable参数设置是否永久化，或者是true。所以创建时可以不指定：

      @Beanpublic Queue fanoutExchangeQueueA() {//只需要指定名称，默认是持久化的return new Queue(RabbitMQConfig.FANOUT_EXCHANGE_QUEUE_TOPIC_A);}
  复制代码
  

  这就完成了消息持久化的设置，随后启动项目，发送几条消息，我们可以看到：

  

  怎么证明是已经持久化了呢，实际上可以找到对应的文件：

  

  找到对应磁盘中的目录：

  

  消息持久化可以防止消息在RabbitMQ Server中不会因为停机机重启而丢失。

  三，消息确认机制

  3.1确认机制

  在生产者发送到RabbitMQ Server时有可能因为网络问题导致投递失败，从而丢失数据。我们可以使用确认模式防止数据丢失。工作流程是怎么样的呢，看以下图解：

  

  从上图中可以看到是通过两个以上函数**confirm（），returnedMessage（）**进行通知。

  一条消息从生产者发送到RabbitMQ，首先会发送到Exchange，对应指向函数confirm（）。第二步从Exchange路由分配到队列中，对应替换函数则是returnedMessage（）。

  代码怎么实现呢，请看演示：

  首先在application.yml配置文件中加上如下配置：

  spring:rabbitmq:publisher-confirms: true
  #    publisher-returns: truetemplate:mandatory: true
  # publisher-confirms：设置为true时。当消息投递到Exchange后，会回调confirm()方法进行通知生产者
  # publisher-returns：设置为true时。当消息匹配到Queue并且失败时，会通过回调returnedMessage()方法返回消息
  # spring.rabbitmq.template.mandatory: 设置为true时。指定消息在没有被队列接收时会通过回调returnedMessage()方法退回。
  复制代码
  

  有个小细节，Publisher-returns和mandatory如果都设置的话，优先级以mandatory优先。可以看源码：

  

  接下来我们需要定义某些方法：

  @Component
  public class RabbitmqConfirmCallback implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {private Logger logger = LoggerFactory.getLogger(RabbitmqConfirmCallback.class);/*** 监听消息是否到达Exchange** @param correlationData 包含消息的唯一标识的对象* @param ack             true 标识 ack，false 标识 nack* @param cause           nack 投递失败的原因*/@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {if (ack) {logger.info("消息投递成功~消息Id：{}", correlationData.getId());} else {logger.error("消息投递失败，Id：{}，错误提示：{}", correlationData.getId(), cause);}}@Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {logger.info("消息没有路由到队列，获得返回的消息");Map map = byteToObject(message.getBody(), Map.class);logger.info("message body: {}", map == null ? "" : map.toString());logger.info("replyCode: {}", replyCode);logger.info("replyText: {}", replyText);logger.info("exchange: {}", exchange);logger.info("routingKey: {}", exchange);logger.info("------------> end <------------");}@SuppressWarnings("unchecked")private <T> T byteToObject(byte[] bytes, Class<T> clazz) {T t;try (ByteArrayInputStream bis = new ByteArrayInputStream(bytes);ObjectInputStream ois = new ObjectInputStream(bis)) {t = (T) ois.readObject();} catch (Exception e) {e.printStackTrace();return null;}return t;}
  }
  复制代码
  

  我这里就简单地地打印某种方法返回的消息，在实际项目中，可以把返回的消息存储到日志表中，使用定时任务进行进一步的处理。

  我这里是使用RabbitTemplate进行发送，所以在服务层的RabbitTemplate需要设置一下：

  @Service
  public class RabbitMQServiceImpl implements RabbitMQService {@Resourceprivate RabbitmqConfirmCallback rabbitmqConfirmCallback;@Resourceprivate RabbitTemplate rabbitTemplate;@PostConstructpublic void init() {//指定 ConfirmCallbackrabbitTemplate.setConfirmCallback(rabbitmqConfirmCallback);//指定 ReturnCallbackrabbitTemplate.setReturnCallback(rabbitmqConfirmCallback);}@Overridepublic String sendMsg(String msg) throws Exception {Map<String, Object> message = getMessage(msg);try {CorrelationData correlationData = (CorrelationData) message.remove("correlationData");rabbitTemplate.convertAndSend(RabbitMQConfig.RABBITMQ_DEMO_DIRECT_EXCHANGE, RabbitMQConfig.RABBITMQ_DEMO_DIRECT_ROUTING, message, correlationData);return "ok";} catch (Exception e) {e.printStackTrace();return "error";}}private Map<String, Object> getMessage(String msg) {String msgId = UUID.randomUUID().toString().replace("-", "").substring(0, 32);CorrelationData correlationData = new CorrelationData(msgId);String sendTime = sdf.format(new Date());Map<String, Object> map = new HashMap<>();map.put("msgId", msgId);map.put("sendTime", sendTime);map.put("msg", msg);map.put("correlationData", correlationData);return map;}
  }
  复制代码
  

  大功告成！接下来我们进行测试，发送一条消息，我们可以控制台：

  

  假设发送一条信息没有路由匹配到层次，可以看到如下信息：

  

  这就是确认模式。它的作用是为了保障生产者投递消息到兔子MQ不会出现消息丢失。

  3.2事务机制（ACK）

  最开始的那张图已经讲过，消费者从层次中获取到消息后，会直接确认签收，假设消费者停机机或程序出现异常，数据没有正常消费，这种情况就会出现数据丢失。

  所以关键在于把自动签收改成手动签收，正常消费则返回确认签收，如果出现异常，则返回拒绝签收重回副本。

  

  代码怎么实现呢，请看演示：

  首先在消费者的application.yml文件中设置事务提交为手册手动模式：

  spring:rabbitmq:listener:simple:acknowledge-mode: manual # 手动ack模式concurrency: 1 # 最少消费者数量max-concurrency: 10 # 最大消费者数量
  复制代码
  

  然后编写消费者的监听器：

  @Component
  public class RabbitDemoConsumer {enum Action {//处理成功SUCCESS,//可以重试的错误，消息重回队列RETRY,//无需重试的错误，拒绝消息，并从队列中删除REJECT}@RabbitHandler@RabbitListener(queuesToDeclare = @Queue(RabbitMQConfig.RABBITMQ_DEMO_TOPIC))public void process(String msg, Message message, Channel channel) {long tag = message.getMessageProperties().getDeliveryTag();Action action = Action.SUCCESS;try {System.out.println("消费者RabbitDemoConsumer从RabbitMQ服务端消费消息：" + msg);if ("bad".equals(msg)) {throw new IllegalArgumentException("测试：抛出可重回队列的异常");}if ("error".equals(msg)) {throw new Exception("测试：抛出无需重回队列的异常");}} catch (IllegalArgumentException e1) {e1.printStackTrace();//根据异常的类型判断，设置action是可重试的，还是无需重试的action = Action.RETRY;} catch (Exception e2) {//打印异常e2.printStackTrace();//根据异常的类型判断，设置action是可重试的，还是无需重试的action = Action.REJECT;} finally {try {if (action == Action.SUCCESS) {//multiple 表示是否批量处理。true表示批量ack处理小于tag的所有消息。false则处理当前消息channel.basicAck(tag, false);} else if (action == Action.RETRY) {//Nack，拒绝策略，消息重回队列channel.basicNack(tag, false, true);} else {//Nack，拒绝策略，并且从队列中删除channel.basicNack(tag, false, false);}channel.close();} catch (Exception e) {e.printStackTrace();}}}
  }
  复制代码
  

  解释一下上面的代码，如果没有异常，则手动确认回复RabbitMQ服务端basicAck（消费成功）。

  如果抛出某些可以重回命名的异常，我们就回复basicNack和设置重回副本。

  如果是抛出不可重回骨骼的异常，就回复basicNack和设置从RabbitMQ的量化中删除。

  接下来进行测试，发送一条普通的消息“你好”：

  

  解释ack返回的三个方法的意思。

  ①成功确认

  void basicAck(long deliveryTag, boolean multiple) throws IOException;
  复制代码
  

  消费者成功处理后调用此方法对消息进行确认。

  • deliveryTag：该消息的索引
  • true：将一次性ack所有小于deliveryTag的消息。

  ②失败确认

  void basicNack(long deliveryTag, boolean multiple, boolean requeue) throws IOException;
  复制代码
  

  • deliveryTag：该消息的索引。
  • true：将一次性拒绝所有小于deliveryTag的消息。
  • requeue：被拒绝的是否重新入级。

  ③失败确认

  void basicReject(long deliveryTag, boolean requeue) throws IOException;
  复制代码
  

  • deliveryTag：该消息的索引。
  • requeue：被拒绝的是否重新入级。

  basicNack（）和basicReject（）的区别在于：basicNack（）可以批量拒绝，basicReject（）一次只能拒接一条消息。

  四，遇到的坑

  4.1启用nack机制后，导致的死循环

  测试发送一条“坏”，然后会引发重回堆栈的异常。然后有个问题：重回所有权后消费者又消费，消费抛出又又重回形态，就造成了死循环。

  

  那怎么避免这种情况呢？

  既然nack会造成死循环的话，我提供的一个思路是不使用basicNack（），把抛出异常的消息落库到一张表中，记录抛出的异常，消息体，消息Id。通过定时任务去处理。

  如果你有什么好的解决方案，也可以留言讨论〜

  4.2双重确认

  有的时候比较粗心，不小心开启了自动Ack模式，又手动回复了Ack。那就会报这个错误：

  消费者RabbitDemoConsumer从RabbitMQ服务端消费消息：java技术爱好者
  2020-08-02 22:52:42.148 ERROR 4880 --- [ 127.0.0.1:5672] o.s.a.r.c.CachingConnectionFactory       : Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - unknown delivery tag 1, class-id=60, method-id=80)
  2020-08-02 22:52:43.102  INFO 4880 --- [cTaskExecutor-1] o.s.a.r.l.SimpleMessageListenerContainer : Restarting Consumer@f4a3a8d: tags=[{amq.ctag-8MJeQ7el_PNbVJxGOOw7Rw=rabbitmq.demo.topic}], channel=Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.0.1:5672/,5), conn: Proxy@782a1679 Shared Rabbit Connection: SimpleConnection@67c5b175 [delegate=amqp://guest@127.0.0.1:5672/, localPort= 56938], acknowledgeMode=AUTO local queue size=0
  复制代码
  

  出现这个错误，可以检查一下yml文件是否添加了以下配置：

  spring:rabbitmq:listener:simple:acknowledge-mode: manualconcurrency: 1max-concurrency: 10
  复制代码
  

  如果上面这个配置已经添加了，还是报错，有可能你使用@Configuration配置了SimpleRabbitListenerContainerFactory，根据SpringBoot的特性，代码配置，代码的配置覆盖了yml的配置，并且忘记了设置手动模式：

  @Beanpublic SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();factory.setConnectionFactory(connectionFactory);//设置手动ack模式factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);return factory;}
  复制代码
  

  如果你还是有报错，那可能是写错地方了，写在生产者的项目了。以上的配置应该配置在消费者的项目。因为ack模式是针对消费者而言的。我就是写错了，写在生产者，折腾了几个小时，泪目〜

  4.3性能问题

  其实手动ACK相对于自动ACK肯定是会慢很多，我在网上查了一些资料，性能相差大概有10倍。所以一般在实际应用中不太建议开手动ACK模式。不过也不是绝对不可以开，具体情况具体分析，看并发量，还有数据的已有等等。

  所以在实际项目中还需要权衡一下并发量和数据的本质，再决定具体的方案。

  4.4启用手动ack模式，如果没有及时回复，会造成异常

  如果开启了手动ACK模式，但是由于代码有错误的原因，没有回复RabbitMQ服务端，那么这条消息就会放到Unacked状态的消息堆里，只有等到消费者的连接才会变成Ready消息。如果消费者一直没有以上连接，那Unacked的消息就会越来越多，占用内存就越来越大，最后就会出现异常。

  这个问题，我没法用我的电脑演示，我的电脑太卡了。

  五，总结

  通过上面的学习后，总结了RabbitMQ防止数据丢失有三种方式：

  • 消息持久化
  • 生产者消息确认机制（confirmmode）
    问题

  其实手动ACK相对于自动ACK肯定是会慢很多，我在网上查了一些资料，性能相差大概有10倍。所以一般在实际应用中不太建议开手动ACK模式。不过也不是绝对不可以开，具体情况具体分析，看并发量，还有数据的已有等等。

  所以在实际项目中还需要权衡一下并发量和数据的本质，再决定具体的方案。

  4.4启用手动ack模式，如果没有及时回复，会造成异常

  如果开启了手动ACK模式，但是由于代码有错误的原因，没有回复RabbitMQ服务端，那么这条消息就会放到Unacked状态的消息堆里，只有等到消费者的连接才会变成Ready消息。如果消费者一直没有以上连接，那Unacked的消息就会越来越多，占用内存就越来越大，最后就会出现异常。

  这个问题，我没法用我的电脑演示，我的电脑太卡了。

  五，总结

  通过上面的学习后，总结了RabbitMQ防止数据丢失有三种方式：

  • 消息持久化
  • 生产者消息确认机制（confirmmode）
  • 消费者消息确认模式（ack模式）