RabbitMQ【消息的可靠性投递（概念、确认模式、退回模式、Ack）】(六)-全面详解（学习总结---从入门到深化）

spring:
 rabbitmq:
   host: 192.168.0.162
   port: 5672
   username: itxiaotong
   password: itxiaotong
   virtual-host: /
  
#日志格式
logging:
 pattern:
   console: '%d{HH:mm:ss.SSS} %clr(%-5level) --- [%-15thread] %cyan(%-50logger{50}):%msg%n'

@Configuration
public class RabbitConfig {
    private final String EXCHANGE_NAME="my_topic_exchange";
    private final String QUEUE_NAME="my_queue";
    // 1.创建交换机
    @Bean("bootExchange")
    public Exchange getExchange(){
        return ExchangeBuilder
               .topicExchange(EXCHANGE_NAME) // 交换机类型
               .durable(true) // 是否持久化
               .build();
   }
    // 2.创建队列
    @Bean("bootQueue")
    public Queue getMessageQueue(){
        return QueueBuilder
               .durable(QUEUE_NAME) // 队列持久化
               .build();
   }
    // 3.将队列绑定到交换机
    @Bean
    public Binding bindMessageQueue(@Qualifier("bootExchange") Exchange exchange, @Qualifier("bootQueue") Queue queue){
        return BindingBuilder.bind(queue).to(exchange).with("my_routing").noargs();
   }
}

spring:
 rabbitmq:
   host: 192.168.0.162
   port: 5672
   username: itxiaotong
   password: itxiaotong
   virtual-host: /
    # 开启确认模式
   publisher-confirm-type: correlated

@SpringBootTest
public class ProducerTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    public void testConfirm(){
        // 定义确认模式的回调方法，消息向交换机发送后会调用confirm方法
      rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            /**
             * 被调用的回调方法
             * @param correlationData 相关配置信息
             * @param ack 交换机是否成功收到了消息
             * @param cause 失败原因
             */
            @Override
            public void confirm(CorrelationData correlationData,boolean ack, String cause) {
                if (ack){
                  System.out.println("confirm接受成功!");
               }else{
                  System.out.println("confirm接受失败，原因为："+cause);
                    // 做一些处理。
               }
           }
       });
      rabbitTemplate.convertAndSend("my_topic_exchange","my_routing","send message...");
   }
}

spring:
 rabbitmq:
   host: 192.168.0.162
   port: 5672
   username: itxiaotong
   password: itxiaotong
   virtual-host: /
    # 开启确认模式
   publisher-confirm-type: correlated
    # 开启回退模式
   publisher-returns: true

@SpringBootTest
public class ProducerTest {
    @Autowired
     private RabbitTemplate rabbitTemplate;
    @Test
    public void testReturn(){
        // 定义退回模式的回调方法。交换机发送到队列失败后才会执行returnedMessage方法
       rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            /**
             * @param returned 失败后将失败信息封装到参数中
             */
            @Override
            public void  returnedMessage(ReturnedMessage returned)
              {
                System.out.println("消息对象："+returned.getMessage());
                System.out.println("错误码："+returned.getReplyCode());
                System.out.println("错误信息："+returned.getReplyText());
                System.out.println("交换机："+returned.getExchange());
                System.out.println("路由键："+returned.getRoutingKey());
                // 处理消息...
           }
       });
       rabbitTemplate.convertAndSend("my_topic_exchange","my_routing1","send message...");
   }
}

spring:
 rabbitmq:
   host: 192.168.0.162
   port: 5672
   username: itxiaotong
   password: itxiaotong
   virtual-host: /
    # 开启手动签收
   listener:
     simple:
       acknowledge-mode: manual

@Component
public class AckConsumer {
    @RabbitListener(queues = "my_queue")
    public void listenMessage(Message message, Channel channel) throws IOException, InterruptedException {
        // 消息投递序号，消息每次投递该值都会+1
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
             int i = 1/0; //模拟处理消息出现bug
             System.out.println("成功接受到消息:"+message);
             // 签收消息
             /**
              * 参数1：消息投递序号
              * 参数2：是否一次可以签收多条消息
              */
             channel.basicAck(deliveryTag,true);
          }catch (Exception e){
            System.out.println("消息消费失败！");
            Thread.sleep(2000);
            // 拒签消息
            /**
             * 参数1：消息投递序号
             * 参数2：是否一次可以拒签多条消息
             * 参数3：拒签后消息是否重回队列
             */
           channel.basicNack(deliveryTag,true,true);
       }
   }
}