Redis集群下过期key监听

迈不过友情╰ 2021-06-22 15:37 822阅读 0赞

背景

> 基于Redis的主动事件的处理，比如：当用户购买了会员卡`十分钟`内没有付款，需要通过小程序或者APP向用户主动推送购买会员卡的优势，引导用户继续完成支付并购买等，类似的场景需要用户在`指定的时间点后`主动通知或者继续引导，使用 `Redis过期键Event`优雅、快捷的实现

实战

在Redis中有两种通知：

类型：

1.  键空间通知：*Keyspace@<db>*\_:mykey(针对mykey的所有的操作的通知)
2.  键事件通知：*Keyevent@<db>*\_:expired (键的过期事件)

解释：

1.  KeySpace键空间通知。即针对`指定key`发生的一切改动，推送给订阅的客户端，侧重于针对`指定Key`的操作；比如*Keyspace@0*:mykey,当更新、删除mykey时，会推送该事件
2.  keyEvent键时间通知。侧重于指定的事件，如：expired(过期事件)、DEL(删除键事件)等；所有的键不针对指定的键，如果要筛选键要通过代码完成

实现方式共有两种方式:(基于Spring Boot完成)  
`RedisMessageListenerContainer` 在Spring Boot 启动之后，最后通过(`SmartLifecycle`)启动，通过定时任务每隔两秒执行一次订阅的任务，一旦有消息，则遍历所有的MessageListener，通过OnMessage方法执行；所以：

1.  继承`MessageListenerAdapter` 监听器，重写OnMessage方法
2.  继承`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,废话比较少。小白自己写的配置类，理解有问题请留言！自己实现的方案感觉不妥，只是基本完成需求，还得继续研究。