redis集群下，监听过期通知

怼烎@ 2022-03-22 15:56 557阅读 0赞

踩坑记录！做一个redis过期通知需求

我采用修改redis.conf中的notify-keyspace-events "Ex"，再建监听类来监听这一事件。但是由于我的redis是集群环境，有3\*2=6个节点，这就导致我的过期通知只有1/6的机会监听的到！！即有时成功有时失败，我找了很久原因！

正确处理，直接上代码：

pom.xml:

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-data-redis</artifactId>
                <version>2.0.2.RELEASE</version>
            </dependency>

配置：

#redis过期监听
    redis.sub.cluster = redis://192.168.1.80:7081
    
    
    spring.redis.database = 0
    spring.redis.timeout = 10000ms
    spring.redis.cluster.max-redirects = 3
    spring.redis.lettuce.pool.min-idle = 50
    spring.redis.lettuce.pool.max-idle = 100
    spring.redis.lettuce.pool.max-wait = 60000ms
    spring.redis.lettuce.pool.max-active = 100
    spring.redis.cluster.nodes[1] = 192.168.1.80:7081
    spring.redis.cluster.nodes[2] = 192.168.1.80:7082
    spring.redis.cluster.nodes[3] = 192.168.1.81:7081
    spring.redis.cluster.nodes[4] = 192.168.1.81:7082
    spring.redis.cluster.nodes[5] = 192.168.1.82:7081
    spring.redis.cluster.nodes[0] = 192.168.1.82:7082

配置类：

package com.ggtm.ggtmbdp.redis.config;
    
    import io.lettuce.core.RedisURI;
    import io.lettuce.core.cluster.RedisClusterClient;
    import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
    import io.lettuce.core.resource.ClientResources;
    import io.lettuce.core.resource.DefaultClientResources;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    /**
     * @Description:
     * @Author: xuyi
     * @Date: 2019-01-16
     * @Version
     */
    @Configuration
    public class LettuceConfig {
    
        @Value("${redis.sub.cluster}")
        String redisSubURI;
    
        @Bean(destroyMethod = "shutdown")
        ClientResources clientResources() {
            return DefaultClientResources.create();
        }
    
        @Bean(destroyMethod = "shutdown")
        RedisClusterClient redisClusterClient(ClientResources clientResources) {
    
            RedisURI redisURI = RedisURI.create(redisSubURI);
    
            return RedisClusterClient.create(clientResources, redisURI);
        }
    
        @Bean(destroyMethod = "close")
        StatefulRedisClusterConnection statefulRedisClusterConnection(RedisClusterClient redisClusterClient) {
            return redisClusterClient.connect();
        }
    }

过期消息处理：

package com.ggtm.ggtmbdp.redis.config;
    
    import io.lettuce.core.cluster.models.partitions.RedisClusterNode;
    import io.lettuce.core.cluster.pubsub.RedisClusterPubSubAdapter;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.stereotype.Component;
    
    
    /**
     * @Description:
     * @Author: xuyi
     * @Date: 2019-01-16
     * @Version
     */
    @Component
    public class ClusterGrooveAdapter extends RedisClusterPubSubAdapter {
    
        private static Logger logger = LoggerFactory.getLogger(ClusterGrooveAdapter.class);
    
        @Override
        public void message(RedisClusterNode node, Object channel, Object message) {
            String expiredKey = null;
            try {
                expiredKey = (String) message;
            } catch (Exception e) {
                logger.error("错误的过期消息类型");
                return;
            }
            logger.info("过期的key:"+expiredKey);
            logger.info("处理过期后的业务:");
        }
    }

启动监听：

package com.ggtm.ggtmbdp.redis.config;
    
    import io.lettuce.core.cluster.RedisClusterClient;
    import io.lettuce.core.cluster.pubsub.StatefulRedisClusterPubSubConnection;
    import io.lettuce.core.cluster.pubsub.api.async.NodeSelectionPubSubAsyncCommands;
    import io.lettuce.core.cluster.pubsub.api.async.PubSubAsyncNodeSelection;
    import io.lettuce.core.pubsub.RedisPubSubAdapter;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.Resource;
    
    /**
     * @Description:
     * @Author: xuyi
     * @Date: 2019-01-16
     * @Version
     */
    @Component
    public class LettuceSubscriber extends RedisPubSubAdapter {
    
        private static Logger logger = LoggerFactory.getLogger(LettuceSubscriber.class);
    
        private static final String EXPIRED_CHANNEL = "__keyevent@0__:expired";
    
    
        @Resource
        RedisClusterClient clusterClient;
    
        @Autowired
        private ClusterGrooveAdapter clusterGrooveAdapter;
    
    
        /**
         * 启动监听
         */
        public void startListener() {
            // 异步订阅
            StatefulRedisClusterPubSubConnection<String, String> pubSubConnection = clusterClient.connectPubSub();
            pubSubConnection.setNodeMessagePropagation(true);
            pubSubConnection.addListener(clusterGrooveAdapter);
    
            PubSubAsyncNodeSelection<String, String> masters = pubSubConnection.async().masters();
            NodeSelectionPubSubAsyncCommands<String, String> commands = masters.commands();
            commands.subscribe(EXPIRED_CHANNEL);
        }
    }

此文用作记载此处坑点，如有转发需求请标明出处