RabbitMQ并发

• 起因
• 通过注解并发
• 配置工厂的时候设置并发

起因

项目中有一个业务需要先把数据从接口入到原始表，再通过MQ机制，让业务处理程序监听并处理。跑了几天以后发现有大量消息堆积，通过rabbitmq控制台看到Ready了几万条数据，Pulish是3/s左右，但Consumer ack则1/s左右，代码分析后发现有很多mysql的查询操作，但是优化起来比较有难度，于是考虑到消息并发处理

通过注解并发

搜了一些方法，最多的就是在RabbitListeners注解上加上concurrency属性

 @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(value = "myQueue1"),
                    exchange = @Exchange(value = "myExchange1"),
                    key = "routingKey1"
            ),
            concurrency =  "10"
    )
    public void process1(Message message) throws Exception {
        // 执行程序
    }

bingdings部分可以忽略，关键是我发现我的RabbitListener注解没有concurrency属性，查看版本，我的是1.7.x，而concurrency是2.x才有的属性

配置工厂的时候设置并发

如果使用1.x版本的也可以直接在配置工厂的时候设置并发数

 @Bean
    public RabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory mqConnectionFactory){
        SimpleRabbitListenerContainerFactory listenerContainerFactory=new SimpleRabbitListenerContainerFactory();
        listenerContainerFactory.setConnectionFactory(mqConnectionFactory);
        listenerContainerFactory.setMessageConverter(new Jackson2JsonMessageConverter());
        // 最大并发数
        listenerContainerFactory.setMaxConcurrentConsumers(20);
        // 初始并发数
        listenerContainerFactory.setConcurrentConsumers(10);
        return listenerContainerFactory;
    }

再使用RabbitListener注解监听消息的时候没有concurrency属性也会并发