RabbitMQ总结

重试机制

背景

• 线上的系统（Spring Boot 2.2.11，rabbitmq为3.2.0），某一天突然有大量的错误日志写入，进几台服务器的硬盘都写满了。查看日志发现是RabbitMQ的消费者在接收消息消费时，抛出了异常错误，此时会不断重新进入消费重新打印错误日志，循环如此进硬盘写满了。

RabbitMQ的消息重试机制，也就是消息失败后进行重试，重试机制是默认开启的，但是如果没有重试机制相关的配置会导致消息一直无间隔的重试，直到消费成功，所以要使用重试机制一定要有相关配置。

例如，原配置如下：

spring.rabbitmq:
  addresses: 10.xxx.xxx.67:5672,10.xxx.xxx.130:5672,10.xxx.xxx.28:5672
  username: admin
  password: admi232ewd$@!32
  virtual-host: /
  # 发布确认模式：配置消息到达交换器的确认回调
  publisher-confirm-type: correlated
  # 发布返回消息：配置消息到达队列的回调
  publisher-returns: true
  # 配置消费端-手动签收，也可在 @RabbitListener 注解中设置参数 ackMode="MANUAL" 开启
  # listener.simple.acknowledge-mode: manual
  # 消息预读数量 1表示每次从队列中读取一条消息
  # listener.simple.prefetch: 1
  listener:
    simple:
      auto-startup: true

这里配置是没有限制重试次数的，因此需要重试次数限制

抛异常消费者:

@Component
@Slf4j(topic = "logger-mq-common#ProcessCompletedEventConsumer")
@RabbitListener(queues = "${spring.rabbitmq.queues-configs.process.event.completed}")
public class ProcessCompletedEventConsumer {
	  @RabbitHandler
    public void process(Map message) {
		    log.info("接收到一个【测试消息重试次数】消息：{}", JSONUtil.toJsonStr(message));
		    JSONObject obj = null;
        // 这里抛异常就一直rabbitmq重新尝试消费
				Assert.isTrue(obj != null, () -> new ApiException("实例不存在"));
    }
}

自动ACK + RabbitMQ重试机制

 .....
  listener:
    simple:
      auto-startup: true
      # ACK模式(默认为auto)
      acknowledge-mode: auto
      # 开启重试
      retry:
        enabled: true
        max-attempts: 3
        initial-interval: 5000

当ACK模式是自动时，达到最大重试上限后，消息会发送到死信队列。

死信就是消息在特定场景下的一种表现形式，这些场景包括：

• 消息被拒绝(basic.reject / basic.nack)，并且requeue = false
• 消息的 TTL 过期时
• 消息队列达到最大长度
• 达到最大重试限制

消息在这些场景中时，被称为死信

死信队列就是用于储存死信的消息队列，在死信队列中，有且只有死信构成，不会存在其余类型的消息。死信队列也是一个普通队列，也可以被消费者消费，区别在于业务队列需要绑定在死信队列上，才能正常地把死信发送到死信队列上。

设置之后，消费者抛异常重试次数也就受到限制。

手动ACK + 手动重试

另外，如果acknowledge-mode是manual，在抛出异常的时候仍会触发重试，但是达到重试上限之后，会永远处于Unacked状态，不会进入到死信队列，必须要手动拒绝才可以进入死信队列，所以说这里不用配置重试机制而是采用手动重试的方式。

/**
 * 消息最大重试次数
 */
private static final int MAX_RETRIES = 3;
 
/**
 * 重试间隔(秒)
 */
private static final long RETRY_INTERVAL = 5;
 
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
	
@RabbitListener(queues = RabbitMqConfig.USER_ADD_QUEUE, concurrency = "10")
public void userAddReceiver(String data, Message message, Channel channel) throws IOException, InterruptedException {
	UserVo vo = OBJECT_MAPPER.readValue(data, UserVo.class);
	// 重试次数
	int retryCount = 0;
	boolean success = false;
    // 消费失败并且重试次数<=重试上限次数
	while (!success && retryCount < MAX_RETRIES) {
		retryCount++;
		// 具体业务逻辑
		success = messageHandle(vo);
		// 如果失败则重试
		if (!success) {
			String errorTip = "第" + retryCount + "次消费失败" +
					((retryCount < 3) ? "," + RETRY_INTERVAL + "s后重试" : ",进入死信队列");
			log.error(errorTip);
			Thread.sleep(RETRY_INTERVAL * 1000);
		}
	}
	if (success) {
		// 消费成功，确认
		channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
		log.info("创建订单数据消费成功");
	} else {
		// 重试多次之后仍失败，进入死信队列
		channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
		log.info("创建订单数据消费失败");
	}
}

总结：两种方案都可以达到我们的预期效果，相比起来方案一会更加的方便简洁，方案二的可控性更高

参考： https://blog.csdn.net/cwr452829537/article/details/124967634