springboot + rabbitmq 如何实现消息确认机制(踩坑经验)

<dependency>
 <groupId>org.springframework.boot</groupId>
 <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

# 发送者开启 confirm 确认机制
spring.rabbitmq.publisher-confirms=true
# 发送者开启 return 确认机制
spring.rabbitmq.publisher-returns=true
####################################################
# 设置消费端手动 ack
spring.rabbitmq.listener.simple.acknowledge-mode=manual
# 是否支持重试
spring.rabbitmq.listener.simple.retry.enabled=true

@Configuration
public class QueueConfig {

 @Bean(name = "confirmTestQueue")
 public Queue confirmTestQueue() {
 return new Queue("confirm_test_queue", true, false, false);
 }

 @Bean(name = "confirmTestExchange")
 public FanoutExchange confirmTestExchange() {
 return new FanoutExchange("confirmTestExchange");
 }

 @Bean
 public Binding confirmTestFanoutExchangeAndQueue(
 @Qualifier("confirmTestExchange") FanoutExchange confirmTestExchange,
 @Qualifier("confirmTestQueue") Queue confirmTestQueue) {
 return BindingBuilder.bind(confirmTestQueue).to(confirmTestExchange);
 }
}

@Slf4j
@Component
public class ConfirmCallbackService implements RabbitTemplate.ConfirmCallback {
 
 @Override
 public void confirm(CorrelationData correlationData, boolean ack, String cause) {

 if (!ack) {
 log.error("消息发送异常!");
 } else {
 log.info("发送者爸爸已经收到确认，correlationData={} ,ack={}, cause={}", correlationData.getId(), ack, cause);
 }
 }
}

@Slf4j
@Component
public class ReturnCallbackService implements RabbitTemplate.ReturnCallback {

 @Override
 public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
 log.info("returnedMessage ===> replyCode={} ,replyText={} ,exchange={} ,routingKey={}", replyCode, replyText, exchange, routingKey);
 }
}

@Autowired
 private RabbitTemplate rabbitTemplate;

 @Autowired
 private ConfirmCallbackService confirmCallbackService;

 @Autowired
 private ReturnCallbackService returnCallbackService;

 public void sendMessage(String exchange, String routingKey, Object msg) {

 /**
 * 确保消息发送失败后可以重新返回到队列中
 * 注意：yml需要配置 publisher-returns: true
 */
 rabbitTemplate.setMandatory(true);

 /**
 * 消费者确认收到消息后，手动ack回执回调处理
 */
 rabbitTemplate.setConfirmCallback(confirmCallbackService);

 /**
 * 消息投递到队列失败回调处理
 */
 rabbitTemplate.setReturnCallback(returnCallbackService);

 /**
 * 发送消息
 */
 rabbitTemplate.convertAndSend(exchange, routingKey, msg,
 message -> {
  message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
  return message;
 },
 new CorrelationData(UUID.randomUUID().toString()));
 }

@Slf4j
@Component
@RabbitListener(queues = "confirm_test_queue")
public class ReceiverMessage1 {
 
 @RabbitHandler
 public void processHandler(String msg, Channel channel, Message message) throws IOException {

 try {
 log.info("小富收到消息：{}", msg);

 //TODO 具体业务
 
 channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

 } catch (Exception e) {
 
 if (message.getMessageProperties().getRedelivered()) {
 
 log.error("消息已重复处理失败,拒绝再次接收...");
 
 channel.basicReject(message.getMessageProperties().getDeliveryTag(), false); // 拒绝消息
 } else {
 
 log.error("消息即将再次返回队列处理...");
 
 channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true); 
 }
 }
 }
}

void basicAck(long deliveryTag, boolean multiple) 

void basicNack(long deliveryTag, boolean multiple, boolean requeue)

void basicReject(long deliveryTag, boolean requeue)

@RabbitHandler
 public void processHandler(String msg, Channel channel, Message message) throws IOException {

 try {
 log.info("消费者 2 号收到：{}", msg);

 int a = 1 / 0;

 channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

 } catch (Exception e) {

 channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
 }
 }

channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
// 重新发送消息到队尾
channel.basicPublish(message.getMessageProperties().getReceivedExchange(),
  message.getMessageProperties().getReceivedRoutingKey(), MessageProperties.PERSISTENT_TEXT_PLAIN,
  JSON.toJSONBytes(msg));