SpringBoot集成RabbitMq，RabbitMq消费与生产，消费失败重发机制，发送签收确认机制_rabbitmq 消费者 ack 抛异常 消息重发

1. RabbitMq集成spring boot

• RabbitMq集成依赖

         这里spring-boot依赖版本为2.3.7版本，RabbitMq集成amqp包，版本在spring-boot中有涵盖，不单独指明版本了。

  <parent>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-parent</artifactId>
      <version>2.3.7.RELEASE</version>
  </parent>
  
  <dependencyManagement>
    	<dependencies>
          <dependency>
              <groupId>org.springframework.boot</groupId>
              <artifactId>spring-boot-dependencies</artifactId>
              <version>2.3.7.RELEASE</version>
              <type>pom</type>
          </dependency>
      </dependencies>
  </dependencyManagement>	
  
  <dependencies>
     <!-- rabbitMQ -->
      <dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-starter-amqp</artifactId>
      </dependency>
  </dependencies>
  
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21
  22
  23
  24
  25

• RabbitMq配置

  spring:
    rabbitmq:
      # 基础项
      host:  ip地址
      port: 端口
      username:  用户名
      password: 密码
      # virtualhost需要提前在MQ的Web管理界面里手动创建，或者配置默认host"/"
      virtual-host: /
      # 生产者
      #确认消息已发送到交换机(Exchange)
      publisher-confirm-type: correlated
      #开启消息发送确认机制，默认为false
      #如果没有本条配置信息，当消费者收到生产者发送的消息后，生产者无法收到确认成功的回调信息
      publisher-confirms: true
      #支持消息发送失败返回队列,默认为false
      publisher-returns: true
      # 消费者
      listener:
        type: simple
        simple:
          #自动签收auto  手动 manual
          acknowledge-mode: auto
          #个字段一定要设置成 false 不然无法消费的数据不会进入死信队列的
          default-requeue-rejected: false
          prefetch: 1 #限制每次发送一条数据
          max-concurrency: 1 #启动消费者最大数量
          concurrency: 1 #同一个队列启动几个消费者
          retry:
            enabled: true #是否支持重试
            max-attempts: 3         # 最大重试次数，默认为3
            initial-interval: 30s # 重试间隔时间，默认1000(单位毫秒)
            max-interval: 120s # 重试最大间隔
            # 时间间隔的乘子，下一次间隔的时间=间隔时间 × 乘子，但最大不超过重试最大间隔
            multiplier: 1	
  
  	```
  
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21
  22
  23
  24
  25
  26
  27
  28
  29
  30
  31
  32
  33
  34
  35
  36
  37
  38

• RabbitMq生产者，队列，交换通道配置，消费者示例

  Exchange 交换机配置

  @Component
  public class DnfxExchangeConfig {
  
      @Autowired
      RabbitMqConfig rabbitMqConfig;
      /**
       * topic交换机起名
       * 如果rabbitmq设置的类型是topic 就用topic类型的Exchange
       *
       * @return
       */
      @Bean
      TopicExchange dnfxOrderExchange() {
          return new TopicExchange(rabbitMqConfig.getFxexchange());
      }
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16

  队列queue配置

  @Component
  public class DnfxQueueConfig {
      @Autowired
      RabbitMqConfig rabbitMqConfig;
  
      /**
       * 队列起名
       *
       * @return
       */
      @Bean
      public Queue dnfxOrderQueue() {
          Map<String, Object> argsMap = new HashMap<String, Object>();
          //队列优先级  argsMap.put("x-max-priority", 5);
           //true 是否持久 
          return new Queue(rabbitMqConfig.getFxqueue(), true, false, false, argsMap);
      }
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18

  将队列和交换机绑定, 并设置用于匹配键

   @Component
  public class DnfxRoutingConfig {
      @Autowired
      RabbitMqConfig rabbitMqConfig;
  
      @Autowired
      DnfxQueueConfig queueConfig;
  
      @Autowired
      DnfxExchangeConfig exchangeConfig;
  
      /**
       * 绑定：将队列和交换机绑定, 并设置用于匹配键 myDirectRouting
       *
       * @return
       */
      @Bean
      Binding bindingOrderRouting() {
          return BindingBuilder.bind(queueConfig.dnfxOrderQueue()).to(exchangeConfig.dnfxOrderExchange()).with(rabbitMqConfig.getFxrouting());
      }
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21

  配置加载

  @Configuration
  @ConfigurationProperties(prefix = "xx.mq")
  @Data
  public class RabbitMqConfig {
      private String fxqueue;
  
      private String fxexchange;
  
      private String fxrouting;	
  }	
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10

  RabbitTemplate

  @Configuration
  public class DnfxRabbitMqConfig {
      @Autowired
      RabbitMqConfig rabbitMqConfig;
      @Bean
      public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
          return new RabbitTemplate(connectionFactory);
      }
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9

  生产者

  @Component
  public class DemoTestProduce {
      @Autowired
      private RabbitTemplate rabbitTemplate;
  
      @Autowired
      RabbitMqConfig rabbitMqConfig;
  
      public void sendDemoMsg() {
          String message = "测试消息发送";
          rabbitTemplate.convertAndSend(rabbitMqConfig.getFxexchange(), rabbitMqConfig.getFxrouting(), message);
      }
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13

  消费者

  @Component
  public class DnfxAliBbMessageListener {
  
      private final static Logger logger = LoggerFactory.getLogger(DnfxAliBbMessageListener.class);
  
      @RabbitListener(containerFactory = "rabbitListenerContainerFactory", queues = "${xx.mq.fxqueue}")
      public void listenSimpleQueueMessage(String msg) throws IOException {
          logger.info("接收到的1688回执消息：{}", msg);
      }
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10

2. RabbitMq消息确认机制

• 消息确认机制分自动确认，和手动确认

  消息确认签收配置

    # 消费者
  listener:
    type: simple
    simple:
      #自动签收 auto  手动 manual
      acknowledge-mode: auto
  1
  2
  3
  4
  5
  6

         消息确认签收机制不过多赘述，网上有大把说明，这里简单描述一下，以及记录一下个人使用心得。
         acknowledge-mode: auto 配置为自动签收时候，消息送达至消费者手上后，Mq自动签收，并移除消息出消息队列。
         acknowledge-mode: false 配置为手动签收时候，消息送达至消费者后，消费者需要手动触发签收动作，如果消费者没有发送ACK消息，RabbitMQ服务器就会认为该消息还没有被消费，会将该消息重新发送给其他消费者。例如下图，手动签收模式，没有主动向MQ发送签收讯息，那么当前消费的这条消息会被标记为Unacked
  
  关于签收 确认机制可以参考 https://blog.csdn.net/qq_42331185/article/details/131696949 ，这里贴部分这个博主的结论

  	@RabbitListener(containerFactory = "rabbitListenerContainerFactory", queues = "${xx.mq.fxqueue}")
      public void listenSimpleQueueMessage(Message message, Channel channel) throws IOException {
          String msgBody = new String(message.getBody());
          logger.info("接收到的1688回执消息：{}", msgBody);
          long deliveryTag = message.getMessageProperties().getDeliveryTag();
          channel.basicReject(deliveryTag, false);
  1
  2
  3
  4
  5
  6

  deliveryTag：消息传递标签，格式为序列号，必须使用这个标签，不然信道会关闭，详情下面会说到
  multiple：为true则表示序号deliverTag之前的消息均被确认或拒绝（basicNack），false表示当前消息。为true的时候就可以做到批量确认
  requeue：为true表示，失败的消息将会重新排队，不会丢弃或者死信，为false则表示丢弃

   1、消息成功签收  basicAck(deliveryTag，multiple)
   channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
   
   2、失败确认 basicNack(deliveryTag，multiple，requeue)
   channel.basicNack(message.getEnvelope().getDeliveryTag(),false, true);
   
   3、失败确认：basicReject(deliveryTag，requeue)
   channel.basicReject(message.getEnvelope().getDeliveryTag(), true);
  1
  2
  3
  4
  5
  6
  7
  8

  注：关于以上手动确认multiple属性为true时，批量确认这个元素个人未进行验证，失败确认requeue为true时，当前消息会重新丟至MQ队列中，等待下次消费（已验证）
         关于消息确认机制，自动确认可能导致消息丢失，如果单条消息发送至消费者后，消费者处理报错，最多触发消息重发机制，重发达到重发上限后，便会抛弃此消息，造成消息丢失。
         手动确认签收，千万不要在cath中或者final中进行失败重发签收，即basicNack basicReject 失败签收时requeue 为true，否则当前消息若真为异常消息，此消息会一直消费，失败签收，重新排队，进行循环，导致消息积压或者资源浪费

3. 消息重发机制

• 消息重发配置

         注意： 如果遗漏 max-interval multiplier两个属性，消息重发机制仍会生效，但是重发间隔时间为默认10秒重发， initial-interval 重发间隔时间将不会生效。此处已验证，尚未确认是bug或者本身就是联动配置

   # 消费者
  listener:
    type: simple
    simple:
      retry:
        enabled: true #是否支持重试
        max-attempts: 3         # 最大重试次数，默认为3
        initial-interval: 30s # 重试间隔时间，默认1000(单位毫秒)
        max-interval: 120s # 重试最大间隔
        # 时间间隔的乘子，下一次间隔的时间=间隔时间 × 乘子，但最大不超过重试最大间隔
        multiplier: 1
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11

• 消息重发如何触发

          消息重发机制，与消息确认签收机制是两种不同的机制，这个概念不要弄混了，消息确认签收机制亦可以将消息重新放入队列进行二次消费
         消息重发机制，在消费者进行消费时，如果rabbitmq开启了消息重发机制，当消费者处理消息时候抛出了异常，即触发消息重发机制，注意，处理消息逻辑不要用try-catch捕捉异常，异常被捕捉后，会抛出异常信息，但不会影响代码正常执行，amqp aop会视为正常消费，不会触发重发机制。

  	@RabbitListener(containerFactory = "rabbitListenerContainerFactory", queues = "${zcwl.mq.fxqueue}")
      public void listenSimpleQueueMessage(Message message, Channel channel) throws IOException {
          String msgBody = new String(message.getBody());
          logger.info("接收到的1688回执消息：{}", msgBody);
  
          long deliveryTag = message.getMessageProperties().getDeliveryTag();
          //此处会抛出异常
          int a = 1 / 0;
          //确认签收机制为手动签收，一定要进行签收，否则触发重发机制后，此条消息仍会被标记为unacked
          channel.basicReject(deliveryTag, false);
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10

4. 延时消息队列

       延时消息队列需要配合RabbitMq延时消息队列插件使用，安装延时消息队列插件此处不赘述，网上搜一大把
       延时消息队列创建队列以及绑定key时没什么特殊的，在创建exchange交换机时，需要注意选项，如下图所示即可。
       x-delayed-type = redirect 如果不能创建，报错时，那么=topic也是可以的

注册exchange交换机时候，注意给入 x-delayed-type 参数，队列注册，以及队列交换机绑定与普通队列一样即可

	@Bean
    CustomExchange dnfxOrderDelayExchange() {
        Map<String, Object> args = new HashMap<String, Object>();
        args.put("x-delayed-type", "topic");
        return new CustomExchange(rabbitMqConfig.getFxOrderDelayExchange(), "x-delayed-message", true, false, args);
    }
1
2
3
4
5
6

测试发送延时消息方法，队列监听与普通消息一样即可

 public void sendDelayMsg() {
        System.out.println(LocalDateTime.now() + "：发送延时消息");
        String message = "这里是测试延时发送消息";
        this.rabbitTemplate.convertAndSend(rabbitMqConfig.getFxOrderDelayExchange(), rabbitMqConfig.getFxOrderDelayRouting(), message, message1 -> {
        	//delay的单位是毫秒
            message1.getMessageProperties().setDelay(1000 * 60);
            return message1;
        });
    }
1
2
3
4
5
6
7
8
9

5. 接收返回结果队列

尚未研究后续用到补充

6. 遇到的报错

• 启动报错 Channel shutdown: channel error; protocol method:

  报错详情：

  Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'type' for exchange 'fx-bb-msg-exchange' in vhost '/': received 'direct' but current is 'topic', class-id=40, method-id=10)
  1

         此错误为注册交换机时候抛出的错误，错误信息为注册交换机的属性，与RabbitMq已经创建好的交换机属性不一致，程序试图修改属性报错。
         错误示范：
         当前exchange交换机创建时候，创建的类型Type为topic类型，在注册exchange交换机时，返回的却是DirectExchange，那么系统便会尝试修改属性，从而引发报错
  
  
         修复方式：
         创建时，返回TopicExchange即可，与 创建的交换机类型保持一致