spring boot-使用redis的Keyspace Notifications实现定时任务队列

    notify-keyspace-events Ex  // 打开此配置，其中Ex表示键事件通知里面的key过期事件，每当有过期键被删除时，会发送通知

 package com.chhliu.springboot.redis.config;

    
  
    
  
    
 import org.springframework.data.redis.connection.Message;
    
 import org.springframework.data.redis.connection.MessageListener;
    
 import org.springframework.stereotype.Component;
    
  
    
 @Component
    
 public class TopicMessageListener implements MessageListener {
    
  
    
 	@Override
    
 	public void onMessage(Message message, byte[] pattern) {// 客户端监听订阅的topic，当有消息的时候，会触发该方法
    
     	byte[] body = message.getBody();// 请使用valueSerializer
    
     	byte[] channel = message.getChannel();
    
     	String topic = new String(channel);
    
     	String itemValue = new String(body);
    
     	// 请参考配置文件，本例中key，value的序列化方式均为string。
    
     	System.out.println("topic:"+topic);
    
     	System.out.println("itemValue:"+itemValue);
    
 	}
    
 }

 package com.chhliu.springboot.redis.config;

    
  
    
 import java.util.concurrent.Executor;
    
 import java.util.concurrent.ThreadPoolExecutor;
    
  
    
 import org.springframework.beans.factory.annotation.Autowired;
    
 import org.springframework.beans.factory.annotation.Value;
    
 import org.springframework.context.annotation.Bean;
    
 import org.springframework.context.annotation.Configuration;
    
 import org.springframework.data.redis.core.RedisTemplate;
    
 import org.springframework.data.redis.listener.ChannelTopic;
    
 import org.springframework.data.redis.listener.RedisMessageListenerContainer;
    
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    
  
    
 @Configuration
    
 public class RedisMessageListenerContainerConfig {
    
 	
    
 	@Autowired
    
 	private RedisTemplate<Object,Object> redisTemplate;
    
 	
    
 	@Autowired
    
 	private TopicMessageListener messageListener;
    
 	
    
 	@Autowired
    
 	private TaskThreadPoolConfig config;
    
 	
    
 	@Value("spring.redis.topic")
    
 	private String topic;
    
 	
    
 	@Bean
    
 	public RedisMessageListenerContainer configRedisMessageListenerContainer(Executor executor){
    
 		RedisMessageListenerContainer container = new RedisMessageListenerContainer();
    
 		// 设置Redis的连接工厂
    
 		container.setConnectionFactory(redisTemplate.getConnectionFactory());
    
 		// 设置监听使用的线程池
    
 		container.setTaskExecutor(executor);
    
 		// 设置监听的Topic
    
 		ChannelTopic channelTopic = new ChannelTopic("__keyevent@0__:expired");
    
 		// 设置监听器
    
 		container.addMessageListener(messageListener, channelTopic);
    
 		return container;
    
 	}
    
 	
    
 	@Bean // 配置线程池
    
 	public Executor myTaskAsyncPool() {
    
 		ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    
 		executor.setCorePoolSize(config.getCorePoolSize());
    
 		executor.setMaxPoolSize(config.getMaxPoolSize());
    
 		executor.setQueueCapacity(config.getQueueCapacity());
    
 		executor.setKeepAliveSeconds(config.getKeepAliveSeconds());
    
 		executor.setThreadNamePrefix(config.getThreadNamePrefix());
    
  
    
 		// rejection-policy：当pool已经达到max size的时候，如何处理新任务
    
 		// CALLER_RUNS：不在新线程中执行任务，而是由调用者所在的线程来执行
    
 		executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
    
 		executor.initialize();
    
 		return executor;
    
 	}
    
 }

 server.port=9999

    
 ########################################################
    
 ###REDIS (RedisProperties) redis基本配置；
    
 ########################################################
    
 # database name
    
 spring.redis.database=0
    
 # server host1 单机使用，对应服务器ip
    
 #spring.redis.host=127.0.0.1  
    
 # server password 密码，如果没有设置可不配
    
 #spring.redis.password=
    
 #connection port  单机使用，对应端口号
    
 #spring.redis.port=6379
    
 # pool settings ...池配置
    
 spring.redis.pool.max-idle=8
    
 spring.redis.pool.min-idle=0
    
 spring.redis.pool.max-active=8
    
 spring.redis.pool.max-wait=-1
    
 # name of Redis server  哨兵监听的Redis server的名称
    
 spring.redis.sentinel.master=mymaster
    
 # comma-separated list of host:port pairs  哨兵的配置列表
    
 spring.redis.sentinel.nodes=192.168.1.108:26379,192.168.1.108:26479,192.168.1.108:26579
    
  
    
 spring.task.pool.corePoolSize=10
    
 spring.task.pool.maxPoolSize=20
    
 spring.task.pool.keepAliveSeconds=60
    
 spring.task.pool.queueCapacity=100
    
 spring.task.pool.threadNamePrefix=myThreadPool
    
  
    
 spring.redis.topic=__keyevent@0__:expired

 package com.chhliu.springboot.redis;

    
  
    
 import org.springframework.boot.SpringApplication;
    
 import org.springframework.boot.autoconfigure.SpringBootApplication;
    
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
    
  
    
 import com.chhliu.springboot.redis.config.TaskThreadPoolConfig;
    
  
    
 @SpringBootApplication
    
 @EnableConfigurationProperties({TaskThreadPoolConfig.class} ) // 开启配置属性支持
    
 public class SpringbootRedisApplication {
    
 	
    
 	public static void main(String[] args) {
    
 		SpringApplication.run(SpringbootRedisApplication.class, args);
    
 	}
    
 }

 set myname chhliu

    
 expire myname 2
    
 或者如下：
    
 set myname chhliu px 2000

 2017-04-12 20:23:16.367  INFO 12464 --- [  myThreadPool2] c.c.s.redis.config.TopicMessageListener  : 是否获取到锁:true

    
 topic:__keyevent@0__:expired
    
 itemValue:myname
    
 2017-04-12 20:23:16.369  INFO 12464 --- [  myThreadPool2] c.c.s.redis.config.TopicMessageListener  : 任务结束，释放锁!