RabbitMQ消费者多线程
作者:互联网
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属性也会并发
标签:listenerContainerFactory,消费者,RabbitListener,RabbitMQ,并发,value,concurrency,注解,多线程 来源: https://blog.csdn.net/weixin_39080782/article/details/112788005