redis订阅内存过大的处理方法：
为监听类自定义一个线程池即可，代码如下：
package com.sec.ems.util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;
/**
* redis监听发布类
*
*/
@Component("redisPubSub")
public class RedisPubSubImpl implements PublishSubcriber,InitializingBean {
private static final Logger LOGGER = LoggerFactory.getLogger(RedisPubSubImpl.class);
//redis模板
@Autowired
private StringRedisTemplate redisTemplate;
//初始化redis消息监听容器
private RedisMessageListenerContainer listenerContainer = new RedisMessageListenerContainer();
@Autowired
private ThreadPoolTaskExecutor taskExecutor;
@Override
public void publish(String channel, String msg) {
Assert.notNull(channel, "Invalid channel:The channel is required.");
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("Pubilish a message[{}] to channel[{}].",msg,channel);
}
//转换发布消息
redisTemplate.convertAndSend(channel, msg);
}
@Override
public void subscribe(String channel, MessageListener listener) {
LOGGER.info("subscribe a channel:{}",channel);
listenerContainer.addMessageListener(listener, new ChannelTopic(channel));
}
/**
* 线程池配置
* @return
* 2019年11月13日
*/
@Bean
public ThreadPoolTaskExecutor taskExecutor(){
ThreadPoolTaskExecutor springSessionRedisTaskExecutor = new ThreadPoolTaskExecutor();
//核心线程
springSessionRedisTaskExecutor.setCorePoolSize(50);
//最大线程
springSessionRedisTaskExecutor.setMaxPoolSize(300);
//线程最大空闲时间
springSessionRedisTaskExecutor.setKeepAliveSeconds(10);
//队列大小
springSessionRedisTaskExecutor.setQueueCapacity(1000);
//线程名称前缀
springSessionRedisTaskExecutor.setThreadNamePrefix("Spring session redis executor thread: ");
return springSessionRedisTaskExecutor;
}
/**
* 初始化配置
*/
@Override
public void afterPropertiesSet() throws Exception {
//设置监听容器的redis连接工厂类
listenerContainer.setConnectionFactory(redisTemplate.getConnectionFactory());
//监听容器的初始化配置 线程池
listenerContainer.setTaskExecutor(taskExecutor);
listenerContainer.afterPropertiesSet();
//启动监听容器
listenerContainer.start();
if (LOGGER.isInfoEnabled()) {
LOGGER.info("isActive:{},autoRun:{},running:{}", listenerContainer.isActive(), listenerContainer.isAutoStartup(), listenerContainer.isRunning());
}
}
}
