Redis基于JedisPubSub使用过期策略

青旅半醒 2021-11-17 10:56 321阅读 0赞

最近在项目中遇到了token过期实时推送的问题，我们的token是存在redis中半个小时不操作自动失效过期，token过期后需要个触发机制，这个触发机制就是Redis的订阅/监听机制。由于集成了Jedis，这里使用JedisPubSub完成整个操作。

首先是对消息的订阅 下面 config(Jedis jedis)方法 **建议在redis客户端的配置文件redis.conf里修改notify-keyspace-events 为Ex**

notify-keyspace-events "Ex"

# K    键空间通知，以__keyspace@<db>__为前缀
    # E    键事件通知，以__keysevent@<db>__为前缀
    # g    del , expipre , rename 等类型无关的通用命令的通知, ...
    # $    String命令
    # l    List命令
    # s    Set命令
    # h    Hash命令
    # z    有序集合命令
    # x    过期事件（每次key过期时生成）
    # e    驱逐事件（当key在内存满了被清除时生成）
    # A    g$lshzxe的别名，因此”AKE”意味着所有的事件

Ex代表了过期事件

@Configuration
    public class RedisConfiguration {
    
        private String redisTopic = “__keyevent@0__:expired”; //订阅事件失效通道  0表示db0 根据自己的dbIndex选择合适的数字
    
        @Bean(name= "jedis.pool")
        @Autowired
        public JedisPool jedisPool() {
            JedisPool pool = new JedisPool(config, host, port);
            initRedisMQ(pool.getResource());
            return pool;
        }
        
        private void initRedisMQ(Jedis jedis){
            Runnable runnable = new Runnable() {
                @Override
                public void run() {
                    jedis.psubscribe(new KeyExpiredListener(), redisTopic);
    //                config(jedis);  //设置redis监听过期策略
                }
            };
            Thread thread = new Thread(runnable);
            thread.start();
        }
    
        private void config(Jedis jedis){
            String parameter = "notify-keyspace-events";
            List<String> notify = jedis.configGet(parameter);
            if(notify.get(1).equals("")){
                jedis.configSet(parameter, "Ex"); //过期事件
            }
        }
    }

还需要一个类KeyExpiredListener需要继承JedisPubSub做订阅监听的主要功能

public class KeyExpiredListener extends JedisPubSub {
    
        private final String prefix = "USER:CACHE:TOKEN:";//会话过期
    
        @Override
        public void onPSubscribe(String pattern, int subscribedChannels) {
            logger.info("onPSubscribe "+ pattern + " " + subscribedChannels);
        }
    
        @Override
        public void onPMessage(String pattern, String channel, String key) {
            if(key.contains(prefix)){
                key=key.replaceAll(prefix,"");
                //生成消息实体进行推送
                WebSocket.sendWebClient(key, MsgPushDto.buildMsg(MsgTypeEnum.TOKEN_EX.getMsg(), MsgTypeEnum.TOKEN_EX.name()),true);
            }
        }
    
    }

这里推荐还是把消息实体封装一下，以便后面新类型消息的构造

@Getter 
    @AllArgsConstructor 
    public enum  MsgTypeEnum {
    
        TOKEN_EX( "长时间未操作，token已失效，请重新登录！"),
        private String msg;
    }

@AllArgsConstructor 
    public class MsgPushDto {
    
        private String msg;
        private String type;
        
        public static String buildMsg(String msg, String type){
            return JSON.toJSONString(new MsgPushDto(msg, type));
        }
    }

以此篇博客作为笔记记录使用