掌控消息全链路(4)——RabbitMQ/Spring-AMQP高级特性详解之事务与消息分发

1.事务

AMQP(高级消息队列协议)实现了事务机制，主要用于确保消息的原子性发布和确认。换言之，它允许你将多个操作(如发送消息、确认消息)绑定在一起，要么全部成功，要么全部失败

• 配置队列

  @Configuration
  public class RabbitMQConfig {
      @Bean("transQueue")
      public Queue transQueue(){
          return QueueBuilder.durable(Constants.TRANS_QUEUE).build();
      }
  }
  

• 配置RabbitTemplate和事务管理器

  @Configuration
  public class RabbitTemplateConfig {
      @Bean("transRabbitTemplate")
      public RabbitTemplate transRabbitTemplate(ConnectionFactory connectionFactory) {
          RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
          rabbitTemplate.setChannelTransacted(true);
          return rabbitTemplate;
      }
      @Bean
      public RabbitTransactionManager rabbitTransactionManager(ConnectionFactory connectionFactory){
          return new RabbitTransactionManager(connectionFactory);
      }
  }
  

• 发送消息

  @RestController
  @RequestMapping("/producer")
  public class ProducerController {
      @Resource(name = "transRabbitTemplate")
      private RabbitTemplate transRabbitTemplate;
      @Transactional
      @RequestMapping("/trans")
      public String trans(){
          transRabbitTemplate.convertAndSend("",Constants.TRANS_QUEUE, "trans test ---> 1");
          int num = 5 / 0;
          transRabbitTemplate.convertAndSend("",Constants.TRANS_QUEUE, "trans test ---> 2");
          return "发送成功";
      }
  }
  

  

  
  Spring Boot的RabbitMQ自动配置默认会确认模式，但RabbitMQ不允许同一个通道同时使用事务模式和确认模式，所以需要确保publisher confirms被禁用

  spring:
  rabbitmq:
  publisher-confirm-type: none
  publisher-returns: false
  

2.消息限流

消息限流(Flow Control) 是RabbitMQ防止生产者发送消息速度超过消费者处理能力，导致消息积压和系统崩溃的保护机制

• 声明和配置交换器、队列和绑定关系

  @Configuration
  public class RabbitMQConfig {
      @Bean("qosQueue")
      public Queue qosQueue(){
          return QueueBuilder.durable(Constants.QOS_QUEUE).build();
      }
      @Bean("qosExchange")
      public DirectExchange qosExchange(){
          return ExchangeBuilder.directExchange(Constants.QOS_EXCHANGE).build();
      }
      @Bean("qosBinding")
      public Binding qosBinding(@Qualifier("qosExchange") DirectExchange directExchange,@Qualifier("qosQueue") Queue queue){
          return  BindingBuilder.bind(queue).to(directExchange).with("qos");
      }
  }
  

• 限制每个消费者未确认的最大消息数

  spring:
    rabbitmq:
      listener:
        simple:
          acknowledge-mode: manual # 消费者确认机制
          prefetch: 5
  

• 配置消费者

  @Component
  @Slf4j
  public class QosListener {
      @RabbitListener(queues = Constants.QOS_QUEUE)
      public void handMessage(Message message, Channel channel) throws IOException {
          long deliveryTag = message.getMessageProperties().getDeliveryTag();
          try{
              log.info("接收到消息:{},deliveryTag:{}",
                      new String(message.getBody(), StandardCharsets.UTF_8),
                      deliveryTag);
              log.info("处理成功");
              // channel.basicAck(deliveryTag, true); // 消费者不确认消息
          }catch(Exception e){
              channel.basicNack(deliveryTag,true,true);
          }
      }
  }
  

• 发送消息

  @RestController
  @RequestMapping("/producer")
  public class ProducerController {
      @Resource(name = "rabbitTemplate")
      private RabbitTemplate rabbitTemplate;
      @RequestMapping("/qos")
      public String qos(){
          for (int i = 0; i < 20; i++) {
              rabbitTemplate.convertAndSend(Constants.QOS_EXCHANGE, "qos", "qos test:" + i);
          }
          return "发送成功";
      }
  }