Springboot2.x集成Rabbitmq实现消费者限流，手动ack确认

古城微笑少年丶 2023-06-17 07:57 50阅读 0赞

### 前言 ###

我们在实际项目中，可能在mq中积累了成千上万的消息，如果我们不进行限流，当我们打开消费者的时候一下子成千上万的消息一下子冲击过来，可能会造成服务器宕机，或者业务出现严重漏洞，所以我们需要进行消费者限流。

首先我的springboot版本，springBootVersion = '2.2.1.RELEASE'。其他版本配置差别都不大。

首先看一下配置，这里只用到了没有注释的配置：

server:
      port: 3002
    
    spring:
      application:
        name: zoo-plus-mq
        #https://docs.spring.io/spring-boot/docs/2.1.3.RELEASE/reference/html/common-application-properties.html
      rabbitmq:
        virtual-host: /
        host: 127.0.0.1
        username: guest
        password: guest
        port: 5672
    #    #必须配置这个才会确认回调
    #    publisher-confirm-type: correlated
    #    #支持发布返回
    #    publisher-returns: true
        listener:
          type: simple
          simple:
            #采用手动应答
            acknowledge-mode: manual
            prefetch: 1 #限制每次发送一条数据。
    #        #当前监听容器数 同一个队列启动几个消费者
    #        concurrency: 1
    #        #启动消费者最大数量
    #        max-concurrency: 1
    #        #是否支持重试
    #        retry:
    #          enabled: true
    #          max-attempts: 5
    #          stateless: false
    #          #时间策略乘数因子
    #          multiplier: 1.0
    #          initial-interval: 1000ms
    #          max-interval: 10000ms
    #        default-requeue-rejected: true

新建一个测试队列：

/**
         * 测试队列
         */
        @Bean
        public Queue testQueue() {
            return QueueBuilder.nonDurable("test-queue").build();
        }

生产者：

/**
         * 发送五条数据，测试消费端必须ack才发送第二条，消费者限流
         */
        @GetMapping("ack")
        public Resp testAck() {
            for (int i = 0; i < 5; i++) {
                rabbitTemplate.convertAndSend("test-queue", "测试ack确认模式");
            }
            return Resp.success("ok", null);
        }

消费者：

@RabbitListener(queues = {"test-queue"})
        public void testQueue(Message message, Channel channel) throws IOException {
            log.info("test-queue消费：" + new String(message.getBody()));
    
            /*
                  listener:
                    type: simple
                    simple:
                      #采用手动应答
                      acknowledge-mode: manual
                      prefetch: 1 #限制每次发送一条数据。
             */
    //        采用手动ack，一条条的消费
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }

由上面测试，如果我们没有配置，一下子五条信息就全部拥挤上来了，如果我们加了配置，不在消费端进行ack的话，消息就会unacked，下次重启服务也会再一次发送这条消息，直到消费端ack。