RabbitMQ 可靠投递

生产端：

1. 修改 spring.rabbitmq.publisher-confirms: true

   spring:
   rabbitmq:
    addresses: 172.28.0.132:5672
    username: admin
    password: aompMq123!
    virtual-host: local-zhengpei
    publisher-confirms: true # 开启发送端确认
    publisher-returns: true  # 开启发送端消息抵达队列的确认
    template:
      mandatory: true    # 只要抵达队列，以异步模式优先回调这个return confirm
    listener:
      simple:
        acknowledge-mode: manual # 手动确认消息

2. 编写initRabbitTemplate（）方法

ack为true,表示消息已经到达broker
mq配置类

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class MyRabbitConfig {
  private static final Logger logger = LoggerFactory.getLogger(MyRabbitConfig.class);
  private RabbitTemplate rabbitTemplate;
  @Bean
  public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
    RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
    this.rabbitTemplate = rabbitTemplate;
    // 确认开启ConfirmCallback回调
    rabbitTemplate.setMandatory(true);
    rabbitTemplate.setMessageConverter(messageConverter());
    initRabbitTemplate();
    return rabbitTemplate;
  }
  @Bean
  public MessageConverter messageConverter() {
    return new Jackson2JsonMessageConverter();
  }
  /**
   * 定制RabbitTemplate
   * 1、服务收到消息就会回调
   * 1.1、spring.rabbitmq.publisher-confirms: true
   * 1.2、设置确认回调
   * 2、消息正确抵达队列就会进行回调
   * 2.1、spring.rabbitmq.publisher-returns: true // 开启发送端确认
   * 2.2 spring.rabbitmq.template.mandatory: true
   * 2.3、设置确认回调ReturnCallback
   * 3、消费端确认(保证每个消息都被正确消费，此时才可以broker删除这个消息)
   */
  public void initRabbitTemplate() {
    //设置确认回调
    rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
      /**
       * 1、只要消息抵达Broker就ack=true,
       * @param correlationData 当前消息的唯一关联数据（这个是消息的唯一ID）
       * @param ack 消息是否成功收到
       * @param cause 失败的原因
       */
      @Override
      public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        logger.info("confirm correlationData:[{}],ack:[{}],cause:[{}]", correlationData, ack, cause);
      }
    });
    // 设置下消息抵达队列的确认回调
    rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
      /**
       * 只要消息没有投递给指定的队列，就触发这个失败回调
       * @param message 投递失败的消息详细信息
       * @param replyCode 回复的状态码
       * @param replyText 回复的文本内容
       * @param exchange 当时这个消息发给哪个交换机
       * @param routingKey 当时这个消息用哪个路邮键
       */
      @Override
      public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        logger.error("Fail Message[" + message + "]==>replyCode[" + replyCode + "]" +
            "==>replyText[" + replyText + "]==>exchange[" + exchange + "]==>routingKey[" + routingKey + "]");
      }
    });  
  }
}

消费端：

/**
 * @author zhengpi
 * @since 2021/6/6
 */
@Component
public class DirectReceiver {
  private static final Logger logger = LoggerFactory.getLogger(DirectReceiver.class);
  // @RabbitHandler 可加可不加
  @RabbitListener(queues = "testDirectQueue")//监听的队列名称 testDirectQueue
  public void process(Message message, Channel channel) {
    String msg = new String(message.getBody());
    // channel 内按照顺序内递增的
    long deliveryTag = message.getMessageProperties().getDeliveryTag();
    logger.info("deliveryTag:[{}]", deliveryTag);
    // 消费消息
    try {
      //签收消息
      channel.basicAck(deliveryTag, false);
      //不签收消息
//      channel.basicNack(deliveryTag,false,true);
    } catch (Exception e) {
      // 网络中断
    }
  }
}