redis订阅内存过大如何处理 redis内存占用过大

文章摘要：redis订阅内存过大如何处理 redis内存占用过大

redis订阅内存过大的处理方法： 为监听类自定义一个线程池即可，代码如下： package com.sec. […]

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());

}

}

}