Redis单机及集群实现key值过期监听机制

爱被打了一巴掌 2023-02-24 07:53 45阅读 0赞

先上代码,redis 集群实现key值过期监听策略

如下配置文档,参考;

修改redis.conf 的记录,该值默认"",然后重启redis即可(也可以config动态修改,但是重启后,就没用了..),所以一步到位,改配置文件.

![20200714190318409.png][]

执行验证:

![2020071419045981.png][]

该值已经证明,key空间时间,可以监听了....xE表示的含义,偷一张图可见...哈哈

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1lvdW5nTGVlMTY_size_16_color_FFFFFF_t_70][]

PSUBSCRIBE \_\_keyevent@\*\_\_:expired

![20200714190835388.png][]

启动监听...然后在同一节点上,起不同的连接进行监听

![20200714191031640.png][]

测试监听没问题,通过....

单机版:

配置文件:

spring:
      redis:
        host: 192.168.241.239
        password: 123456
        port: 6379

配置环境类:

package com.young.springdemo.config.redis;
    
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.data.redis.connection.RedisConnectionFactory;
    import org.springframework.data.redis.listener.RedisMessageListenerContainer;
    
    /**
     * @program: springdemo
     * @description
     * @author: Young
     * @create: 2020-07-10 16:47
     **/
    @Configuration
    public class RedisListenerConfig {
        @Bean
        RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory) {
            RedisMessageListenerContainer container = new RedisMessageListenerContainer();
            container.setConnectionFactory(connectionFactory);
            return container;
        }
    }

具体监听实现服务类:

package com.young.springdemo.config.redis;
    
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.data.redis.connection.Message;
    import org.springframework.data.redis.listener.KeyExpirationEventMessageListener;
    import org.springframework.data.redis.listener.RedisMessageListenerContainer;
    import org.springframework.stereotype.Component;
    
    /**
     * @program: springdemo
     * @description
     * @author: Young
     * @create: 2020-07-10 16:49
     **/
    @Component
    @Slf4j
    public class RedisKeyExpirationListener extends KeyExpirationEventMessageListener {
        public RedisKeyExpirationListener(RedisMessageListenerContainer listenerContainer) {
            super(listenerContainer);
        }
    
        @Override
        public void onMessage(Message message, byte[] pattern) {
            // 用户做自己的业务处理即可,注意message.toString()可以获取失效的key
            String expiredKey = message.toString();
            log.info("key[{}] 已过期", expiredKey);
            //todo 业务
        }
    }

集群版:

下面是代码实现为:

配置文件

(一些其他性能的配置项就没有贴,使用默认值..nodes也可以用- - - - 来进行区分..)

spring:
      redis:
        password: 123456
        cluster:
          nodes: 192.168.241.239:7000,192.168.241.239:7001,192.168.241.239:7002,192.168.241.239:7003,192.168.241.239:7004,192.168.241.239:7005

configuration配置类:

package com.young.springdemo.config.redis;
    
    /**
     * @program: springdemo
     * @description
     * @author: Young
     * @create: 2020-07-11 15:32
     **/
    import org.springframework.beans.factory.BeanFactory;
    import org.springframework.boot.autoconfigure.AutoConfigureAfter;
    import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
    import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
    import org.springframework.boot.autoconfigure.jackson.JacksonAutoConfiguration;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.data.redis.connection.MessageListener;
    import org.springframework.data.redis.connection.RedisConnectionFactory;
    import org.springframework.data.redis.connection.jedis.JedisConnection;
    import org.springframework.data.redis.core.RedisOperations;
    import org.springframework.data.redis.listener.PatternTopic;
    import org.springframework.data.redis.listener.RedisMessageListenerContainer;
    
    import redis.clients.jedis.Jedis;
    
    @Configuration
    @ConditionalOnClass({ JedisConnection.class, RedisOperations.class, Jedis.class, MessageListener.class })
    @AutoConfigureAfter({ JacksonAutoConfiguration.class,RedisAutoConfiguration.class })
    public class RedisAutoConfiguration {
    
        @Configuration
        @ConditionalOnExpression("!'${spring.redis.host:}'.isEmpty()")
        public static class RedisStandAloneAutoConfiguration {
            @Bean
            public RedisMessageListenerContainer customizeRedisListenerContainer(
                    RedisConnectionFactory redisConnectionFactory,MessageListener messageListener) {
                RedisMessageListenerContainer redisMessageListenerContainer = new RedisMessageListenerContainer();
                redisMessageListenerContainer.setConnectionFactory(redisConnectionFactory);
                redisMessageListenerContainer.addMessageListener(messageListener,new PatternTopic("__keyevent@0__:expired"));
                return redisMessageListenerContainer;
            }
        }
    
    
        @Configuration
        @ConditionalOnExpression("'${spring.redis.host:}'.isEmpty()")
        public static class RedisClusterAutoConfiguration {
            @Bean
            public RedisMessageListenerFactory redisMessageListenerFactory(BeanFactory beanFactory,
                                                                           RedisConnectionFactory redisConnectionFactory) {
                RedisMessageListenerFactory beans = new RedisMessageListenerFactory();
                beans.setBeanFactory(beanFactory);
                beans.setRedisConnectionFactory(redisConnectionFactory);
                return beans;
            }
        }
    }

监听配置类:

package com.young.springdemo.config.redis;
    
    /**
     * @program: springdemo
     * @description
     * @author: Young
     * @create: 2020-07-11 15:41
     **/
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.BeansException;
    import org.springframework.beans.factory.BeanFactory;
    import org.springframework.beans.factory.BeanFactoryAware;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.beans.factory.config.BeanDefinition;
    import org.springframework.beans.factory.support.BeanDefinitionBuilder;
    import org.springframework.beans.factory.support.DefaultListableBeanFactory;
    import org.springframework.context.ApplicationListener;
    import org.springframework.context.event.ContextRefreshedEvent;
    import org.springframework.data.redis.connection.MessageListener;
    import org.springframework.data.redis.connection.RedisClusterConnection;
    import org.springframework.data.redis.connection.RedisClusterNode;
    import org.springframework.data.redis.connection.RedisConnectionFactory;
    import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
    import org.springframework.data.redis.listener.PatternTopic;
    import org.springframework.data.redis.listener.RedisMessageListenerContainer;
    
    import redis.clients.jedis.JedisShardInfo;
    
    @Slf4j
    public class RedisMessageListenerFactory implements BeanFactoryAware, ApplicationListener<ContextRefreshedEvent> {
    
        @Value("${spring.redis.password}")
        private String password;
    
        private DefaultListableBeanFactory beanFactory;
    
        private RedisConnectionFactory redisConnectionFactory;
    
        @Autowired
        private MessageListener messageListener;
    
        public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
            this.beanFactory = (DefaultListableBeanFactory) beanFactory;
        }
    
        public void setRedisConnectionFactory(RedisConnectionFactory redisConnectionFactory) {
            this.redisConnectionFactory = redisConnectionFactory;
        }
    
        @Override
        public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {
            RedisClusterConnection redisClusterConnection = redisConnectionFactory.getClusterConnection();
            if (redisClusterConnection != null) {
                Iterable<RedisClusterNode> nodes = redisClusterConnection.clusterGetNodes();
                for (RedisClusterNode node : nodes) {
                    if (node.isMaster()) {
                        log.info("获取到redis的master节点为[{}]",node.toString());
                        String containerBeanName = "messageContainer" + node.hashCode();
                        if (beanFactory.containsBean(containerBeanName)) {
                            return;
                        }
                        JedisShardInfo jedisShardInfo = new JedisShardInfo(node.getHost(), node.getPort());
                        jedisShardInfo.setPassword(password);
                        JedisConnectionFactory factory = new JedisConnectionFactory(jedisShardInfo);
                        BeanDefinitionBuilder containerBeanDefinitionBuilder = BeanDefinitionBuilder
                                .genericBeanDefinition(RedisMessageListenerContainer.class);
                        containerBeanDefinitionBuilder.addPropertyValue("connectionFactory", factory);
                        containerBeanDefinitionBuilder.setScope(BeanDefinition.SCOPE_SINGLETON);
                        containerBeanDefinitionBuilder.setLazyInit(false);
                        beanFactory.registerBeanDefinition(containerBeanName,
                                containerBeanDefinitionBuilder.getRawBeanDefinition());
    
                        RedisMessageListenerContainer container = beanFactory.getBean(containerBeanName,
                                RedisMessageListenerContainer.class);
                        String listenerBeanName = "messageListener" + node.hashCode();
                        if (beanFactory.containsBean(listenerBeanName)) {
                            return;
                        }
                        container.addMessageListener(messageListener, new PatternTopic("__keyevent@0__:expired"));
                        container.start();
                    }
                }
            }
        }
    
    }

具体监听实现业务代码:

package com.young.springdemo.config.redis;
    
    /**
     * @program: springdemo
     * @description
     * @author: Young
     * @create: 2020-07-11 15:42
     **/
    import org.springframework.data.redis.connection.Message;
    import org.springframework.data.redis.connection.MessageListener;
    import org.springframework.stereotype.Component;
    @Component
    public class KeyExpiredEventMessageListener implements MessageListener {
    
        @Override
        public void onMessage(Message message, byte[] pattern) {
            String expired = message.toString();
            System.out.println("======接收监听===="+expired);
        }
    
    }

具体执行的可以在拿到expired主键后进行操作...

注意一点的是,此时只能拿到key...因为key已经过期,所以value值是拿不到的....业务酌情考虑.或者设置键值时进行处理.

[20200714190318409.png]: /images/20230209/12596306f949418b838c44591c8f5e5d.png
[2020071419045981.png]: /images/20230209/5b599f4b387a4437b135d5bfd793ed8a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1lvdW5nTGVlMTY_size_16_color_FFFFFF_t_70]: /images/20230209/53d140eeb64445eb94f2e1f180c34e28.png
[20200714190835388.png]: /images/20230209/8249cc9a3f164bbfab0ae1fd07900b1e.png
[20200714191031640.png]: https://img-blog.csdnimg.cn/20200714191031640.png