SpringBoot集成RabbitMq,RabbitMq消费与生产,消费失败重发机制,发送签收确认机制

发布时间:2024年01月12日

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>
    
    
  • 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	
    
    	```
    
    
  • RabbitMq生产者,队列,交换通道配置,消费者示例

    Exchange 交换机配置

    @Component
    public class DnfxExchangeConfig {
    
        @Autowired
        RabbitMqConfig rabbitMqConfig;
        /**
         * topic交换机起名
         * 如果rabbitmq设置的类型是topic 就用topic类型的Exchange
         *
         * @return
         */
        @Bean
        TopicExchange dnfxOrderExchange() {
            return new TopicExchange(rabbitMqConfig.getFxexchange());
        }
    }
    

    队列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);
        }
    }
    

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

     @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());
        }
    }
    

    配置加载

    @Configuration
    @ConfigurationProperties(prefix = "xx.mq")
    @Data
    public class RabbitMqConfig {
        private String fxqueue;
    
        private String fxexchange;
    
        private String fxrouting;	
    }	
    

    RabbitTemplate

    @Configuration
    public class DnfxRabbitMqConfig {
        @Autowired
        RabbitMqConfig rabbitMqConfig;
        @Bean
        public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
            return new RabbitTemplate(connectionFactory);
        }
    }
    

    生产者

    @Component
    public class DemoTestProduce {
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        @Autowired
        RabbitMqConfig rabbitMqConfig;
    
        public void sendDemoMsg() {
            String message = "测试消息发送";
            rabbitTemplate.convertAndSend(rabbitMqConfig.getFxexchange(), rabbitMqConfig.getFxrouting(), message);
        }
    }
    

    消费者

    @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);
        }
    }
    

2. RabbitMq消息确认机制

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

    消息确认签收配置

      # 消费者
    listener:
      type: simple
      simple:
        #自动签收 auto  手动 manual
        acknowledge-mode: auto
    

    ???????消息确认签收机制不过多赘述,网上有大把说明,这里简单描述一下,以及记录一下个人使用心得。
    ???????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);
    

    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);
    

    注:关于以上手动确认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
    
  • 消息重发如何触发

    ??????? 消息重发机制,与消息确认签收机制是两种不同的机制,这个概念不要弄混了,消息确认签收机制亦可以将消息重新放入队列进行二次消费
    ???????消息重发机制,在消费者进行消费时,如果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);
    

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);
    }

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

 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;
        });
    }

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)
    

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

文章来源:https://blog.csdn.net/loney_wolf/article/details/135551805
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。