网站seo优化服务,网站架构图怎么画,网站建设的运营计划,黄岛网站建设服务前言#xff1a;在使用RabbitListener注解来指定消费方法的时候#xff0c;默认情况是单线程去监听队列#xff0c;但是这个如果在高并发的场景中会出现很多个任务#xff0c;但是每次只消费一个消息#xff0c;就会很缓慢。单线程处理消息容易引起消息处理缓慢#xff0… 前言在使用RabbitListener注解来指定消费方法的时候默认情况是单线程去监听队列但是这个如果在高并发的场景中会出现很多个任务但是每次只消费一个消息就会很缓慢。单线程处理消息容易引起消息处理缓慢消息堆积不能最大利用硬件资源这个就很伤。 处理办法可以添加配置类设置RabbitMQ的容器工厂参数增加并发处理数量即可实现多线程处理监听队列实现多线程处理消息。 一、编写配置类
package com.quick.config;import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;/*** RabbitMQ配置类*/
Configuration
/*ConditionalOnClass(RabbitTemplate.class) //有RabbitTemplate依赖才会生效否则不生效*/
public class MqConfig {// 定义线程数、最大线程数常量private static final int INITIAL_CONCURRENT_CONSUMERS 10;private static final int MAX_CONCURRENT_CONSUMERS 10;/*** 将多线程配置配置注入容器工厂*/Bean(customContainerFactory)public SimpleRabbitListenerContainerFactory containerFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer,ConnectionFactory connectionFactory) {SimpleRabbitListenerContainerFactory factory new SimpleRabbitListenerContainerFactory();factory.setConcurrentConsumers(INITIAL_CONCURRENT_CONSUMERS); //设置线程数factory.setMaxConcurrentConsumers(MAX_CONCURRENT_CONSUMERS); //最大线程数configurer.configure(factory, connectionFactory);return factory;}} setConcurrentConsumers(int concurrentConsumers): 这个方法设置了容器应该同时启动的监听器消费者线程的数量。这些线程会并发地从RabbitMQ队列中拉取并处理消息。这个值决定了系统初始时能够并行处理消息的能力。 setMaxConcurrentConsumers(int maxConcurrentConsumers): 这个方法设置了容器在需要时可以增加到的最大并发消费者数量。这通常用于处理负载高峰当队列中的消息积压时可以动态地增加并发消费者数量以提高处理速度。然而请注意这并不意味着系统会立即创建所有最大数量的线程而是会根据需要逐渐增加到这个上限。 这个容器负责监听 RabbitMQ 的队列并将接收到的消息分发给相应的处理器即 RabbitListener 注解标记的方法 二、修改监听者
在接收消息方里面的RabbitListener注解中添加配置 RabbitListener(queues {监听队列名},containerFactory customContainerFactory) /*** 接收消息*/
Component
public class StoreListener {Resourceprivate IStoreService storeService;Resourceprivate StoreMapper storeMapper;/*** 更新店铺收藏人数实现收藏人数1* param storeId 店铺id*/RabbitListener(bindings QueueBinding(value Queue(name store.addFavorite.success.queue, durable true), // 队列 起名规则服务名业务名成功队列durable持久化exchange Exchange(name addFavorite.direct), // 交换机名称交换机默认类型就行direct所以不用配置directkey addFavorite.success // 绑定的key),// 在RabbitListener注解中指定容器工厂containerFactory customContainerFactory)public void listenAddFavoriteCountsSuccess(Long storeId){storeService.updateStoreFavoriteUsersCountAdd1(storeId);}/*** 根据传过来的店铺实体类修改店铺信息* param store 店铺实体类*/RabbitListener(bindings QueueBinding(value Queue(name store.updateStore.success.queue, durable true),exchange Exchange(name updateStore.direct),key updateStore.success),// 在RabbitListener注解中指定容器工厂containerFactory customContainerFactory)public void updateStoreByEntity(Store store){storeMapper.updateById(store);}} 多线程的好处
1、提高吞吐量通过并行处理消息系统可以更快地处理大量消息从而提高整体吞吐量。
2、更好的资源利用率在多核处理器上多线程可以更好地利用硬件资源减少处理延迟。
