Redis集群下过期key监听
Posted OkidoGreen
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Redis集群下过期key监听相关的知识,希望对你有一定的参考价值。
背景
基于Redis的主动事件的处理,比如:当用户购买了会员卡
十分钟
内没有付款,需要通过小程序或者APP向用户主动推送购买会员卡的优势,引导用户继续完成支付并购买等,类似的场景需要用户在指定的时间点后
主动通知或者继续引导,使用Redis过期键Event
优雅、快捷的实现
实战
在Redis中有两种通知:
类型:
- 键空间通知:Keyspace@<db>_:mykey(针对mykey的所有的操作的通知)
- 键事件通知:Keyevent@<db>_:expired (键的过期事件)
解释:
- KeySpace键空间通知。即针对
指定key
发生的一切改动,推送给订阅的客户端,侧重于针对指定Key
的操作;比如Keyspace@0:mykey,当更新、删除mykey时,会推送该事件 - keyEvent键时间通知。侧重于指定的事件,如:expired(过期事件)、DEL(删除键事件)等;所有的键不针对指定的键,如果要筛选键要通过代码完成
实现方式共有两种方式:(基于Spring Boot完成)RedisMessageListenerContainer
在Spring Boot 启动之后,最后通过(SmartLifecycle
)启动,通过定时任务每隔两秒执行一次订阅的任务,一旦有消息,则遍历所有的MessageListener,通过OnMessage方法执行;所以:
- 继承
MessageListenerAdapter
监听器,重写OnMessage方法 - 继承
KeyExpirationEventMessageListener
监听器, 重写doHandleMessage方法
以下分别详细讲述实现方式:(关于Redis相关数据源的配置,暂不包括)
MessageListenerAdapter监听器
通过适配器自定义监听器后,需要单独配置RedisMessageListerContainer的MessageListener以及监听Pattern(Keyspace或者Keyevent、db、事件),如下:
@Bean
@ConditionalOnMissingBean(RedisMessageListenerContainer.class)
public RedisMessageListenerContainer container()
RedisMessageListenerContainer listenerContainer = new RedisMessageListenerContainer();
//连接池的设置,暂不描述
listenerContainer.setConnectionFactory(jedisConnectionFactory());
//添加自定义的监听器,并将指定的事件添加进去
listenerContainer.addMessageListener(new MyKeyExpireListener(),new PatternTopic("__keyevent@12__:expired"));
return listenerContainer;
KeyExpirationEventMessageListener监听器
KeyExpirationEventMessageListener继承自KeyspaceEventMessageListener,而KeyspaceEventMessageListener监听所有db的Keyevent,而KeyExpirationEventMessageListener专门监听Key的过期事件的类;然后基于Spring Boot 的事件通知机制,当监听到所有db库中的key有过期事件时,通过ApplicationContext发送RedisKeyExpiredEvent事件,消息体即Key;通过继承KeyExpirationEventMessageListener无需再设置RedisMEssageListenerContainer,通过@Compont注解,让IOC容器发现即可
缺点:
Key的过期事件通知,无法将value也通知到,消息体只有key,在某些场景下,需要将key-Value在储存另外一份作为映射,或者拿到key以后再基于数据库进行操作;
1. 前言
在使用redis集群时,发现过期key始终监听不到。网上也没有现成的解决方案。于是想,既然不能监听集群,那我可以建立多个redis连接,分别对每个redis的key过期进行监听。以上做法可能不尽人意,目前也没找到好的解决方案,如果有好的想法,请留言告知哦!不多说,直接贴我自己的代码!
2. 代码实现
关于Redis集群配置代码此处不贴,直接贴配置监听类代码!
redis.host1: 10.113.56.68
redis.port1: 7030
redis.host2: 10.113.56.68
redis.port2: 7031
redis.host3: 10.113.56.68
redis.port3: 7032
redis.host4: 10.113.56.68
redis.port4: 7033
redis.host5: 10.113.56.68
redis.port5: 7034
redis.host6: 10.113.56.68
redis.port6: 7035
application配置类
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cache.CacheManager;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.cache.RedisCacheManager;
import org.springframework.data.redis.connection.RedisClusterConfiguration;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPoolConfig;
import java.util.Arrays;
/**
* @Author xiabing5
* @Create 2019/8/6 14:46
* @Desc 监听redis中Key过期事件
**/
@Configuration
public class RedisListenerConfig
@Value("$redis.host1")
private String host1;
@Value("$redis.host2")
private String host2;
@Value("$redis.host3")
private String host3;
@Value("$redis.host4")
private String host4;
@Value("$redis.host5")
private String host5;
@Value("$redis.host6")
private String host6;
@Value("$redis.port1")
private int port1;
@Value("$redis.port2")
private int port2;
@Value("$redis.port3")
private int port3;
@Value("$redis.port4")
private int port4;
@Value("$redis.port5")
private int port5;
@Value("$redis.port6")
private int port6;
@Bean
JedisPoolConfig jedisPoolConfig()
JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
jedisPoolConfig.setMaxIdle(100);
jedisPoolConfig.setMaxWaitMillis(1000);
return jedisPoolConfig;
// redis-cluster不支持key过期监听,建立多个连接,对每个redis节点进行监听
@Bean
RedisMessageListenerContainer redisContainer1()
final RedisMessageListenerContainer container = new RedisMessageListenerContainer();
JedisConnectionFactory jedisConnectionFactory = new JedisConnectionFactory();
jedisConnectionFactory.setHostName(host1);
jedisConnectionFactory.setPort(port1);
jedisConnectionFactory.setPoolConfig(jedisPoolConfig());
jedisConnectionFactory.afterPropertiesSet();
container.setConnectionFactory(jedisConnectionFactory);
return container;
@Bean
RedisMessageListenerContainer redisContainer2()
final RedisMessageListenerContainer container = new RedisMessageListenerContainer();
JedisConnectionFactory jedisConnectionFactory = new JedisConnectionFactory();
jedisConnectionFactory.setHostName(host2);
jedisConnectionFactory.setPort(port2);
jedisConnectionFactory.setPoolConfig(jedisPoolConfig());
jedisConnectionFactory.afterPropertiesSet();
container.setConnectionFactory(jedisConnectionFactory);
return container;
@Bean
RedisMessageListenerContainer redisContainer3()
final RedisMessageListenerContainer container = new RedisMessageListenerContainer();
JedisConnectionFactory jedisConnectionFactory = new JedisConnectionFactory();
jedisConnectionFactory.setHostName(host3);
jedisConnectionFactory.setPort(port3);
jedisConnectionFactory.setPoolConfig(jedisPoolConfig());
jedisConnectionFactory.afterPropertiesSet();
container.setConnectionFactory(jedisConnectionFactory);
return container;
@Bean
RedisMessageListenerContainer redisContainer4()
final RedisMessageListenerContainer container = new RedisMessageListenerContainer();
JedisConnectionFactory jedisConnectionFactory = new JedisConnectionFactory();
jedisConnectionFactory.setHostName(host4);
jedisConnectionFactory.setPort(port4);
jedisConnectionFactory.setPoolConfig(jedisPoolConfig());
jedisConnectionFactory.afterPropertiesSet();
container.setConnectionFactory(jedisConnectionFactory);
return container;
@Bean
RedisMessageListenerContainer redisContainer5()
final RedisMessageListenerContainer container = new RedisMessageListenerContainer();
JedisConnectionFactory jedisConnectionFactory = new JedisConnectionFactory();
jedisConnectionFactory.setHostName(host5);
jedisConnectionFactory.setPort(port5);
jedisConnectionFactory.setPoolConfig(jedisPoolConfig());
jedisConnectionFactory.afterPropertiesSet();
container.setConnectionFactory(jedisConnectionFactory);
return container;
@Bean
RedisMessageListenerContainer redisContainer6()
final RedisMessageListenerContainer container = new RedisMessageListenerContainer();
JedisConnectionFactory jedisConnectionFactory = new JedisConnectionFactory();
jedisConnectionFactory.setHostName(host6);
jedisConnectionFactory.setPort(port6);
jedisConnectionFactory.setPoolConfig(jedisPoolConfig());
jedisConnectionFactory.afterPropertiesSet();
container.setConnectionFactory(jedisConnectionFactory);
return container;
@Bean
RedisKeyExpirationListener redisKeyExpirationListener1()
return new RedisKeyExpirationListener(redisContainer1());
@Bean
RedisKeyExpirationListener redisKeyExpirationListener2()
return new RedisKeyExpirationListener(redisContainer2());
@Bean
RedisKeyExpirationListener redisKeyExpirationListener3()
return new RedisKeyExpirationListener(redisContainer3());
@Bean
RedisKeyExpirationListener redisKeyExpirationListener4()
return new RedisKeyExpirationListener(redisContainer4());
@Bean
RedisKeyExpirationListener redisKeyExpirationListener5()
return new RedisKeyExpirationListener(redisContainer5());
@Bean
RedisKeyExpirationListener redisKeyExpirationListener6()
return new RedisKeyExpirationListener(redisContainer6());
Bean配置类
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.listener.KeyExpirationEventMessageListener;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import java.util.Date;
/**
* @Author xiabing5
* @Create 2019/9/4 9:47
* @Desc redis过期监听
**/
public class RedisKeyExpirationListener extends KeyExpirationEventMessageListener
@Autowired
RedisUtil redisUtil;
@Autowired
LoginUserStatisticsMapper loginUserStatisticsMapper;
public RedisKeyExpirationListener(RedisMessageListenerContainer listenerContainer)
super(listenerContainer);
@Override
public void onMessage(Message message, byte[] pattern)
// 用户做自己的业务处理即可,message.toString()可以获取失效的key
String mesg = message.toString();
监听操作类
3. Redis防止过期key重复监听
对于项目集群情况下,部署多个服务后,容易出现redis过期被多个服务同时监听到,从而执行相同的业务逻辑,这不是我们期望的。单机部署下方法的同步可以采用synchronize关键字。但集群下,就得采用分布式锁。在需要加锁的地方,只要加锁和解锁即可。此处正好写到Redis,那就贴一个自己用的redis分布式锁。
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;
import java.util.Collections;
import java.util.UUID;
/**
* @Author xiabing5
* @Create 2019/9/6 15:54
* @Desc redis分布式锁
**/
@Component
public class RedisLock
@Autowired
Jedis jedis;
private static final String SET_IF_NOT_EXIST = "NX"; // NX表示如果不存在key就设置value
private static final String SET_WITH_EXPIRE_TIME = "PX"; // PX表示毫秒
// 加锁
public String tryLock(String key,Long acquireTimeout)
// 生成随机value
String identifierValue = UUID.randomUUID().toString();
// 设置超时时间
Long endTime = System.currentTimeMillis() + acquireTimeout;
// 循环获取锁
while (System.currentTimeMillis() < endTime)
String result = jedis.set(key,identifierValue, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, acquireTimeout);
if("OK".equals(result))
return identifierValue;
return null;
// 解锁
// public void delLock(String key,String identifierValue)
// // 判断是否是同一把锁
// try
// if(jedis.get(key).equals(identifierValue))
// // 此处操作非原子性,容易造成释放非自己的锁
// jedis.del(key);
//
// catch(Exception e)
// e.printStackTrace();
//
//
// 使用Lua代码解锁
public void delLock(String key,String identifierValue)
try
String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
Long result = (Long) jedis.eval(script, Collections.singletonList(key), Collections.singletonList(identifierValue));
if (1 == result)
System.out.println(result+"释放锁成功");
if (0 == result)
System.out.println(result+"释放锁失败");
catch (Exception e)
e.printStackTrace();
Redis锁解决分布式同步问题
4. 总结
自己实现的一个小demo,废话比较少。小白自己写的配置类,理解有问题请留言!自己实现的方案感觉不妥,只是基本完成需求,还得继续研究。
以上是关于Redis集群下过期key监听的主要内容,如果未能解决你的问题,请参考以下文章