Spring Boot集成redis集群拓扑动态刷新

末蓝、 2023-10-16 15:31 9阅读 0赞

## 项目场景： ##

`Spring Boot集成Redis集群，使用lettuce连接Cluster集群实例。`

--------------------

## 问题描述 ##

redis`其中一个节点挂了之后，springboot集成redis集群配置信息没有及时刷新，出现读取操作报错。```

java.lang.IllegalArgumentException: Connection to 127.0.0.1:6379 not allowed. This connection point is not known in the cluster view
    exceptionStackTrace
    io.lettuce.core.cluster.PooledClusterConnectionProvider.getConnectionAsync(PooledClusterConnectionProvider.java:359)
    io.lettuce.core.cluster.ClusterDistributionChannelWriter.write(ClusterDistributionChannelWriter.java:93)
    io.lettuce.core.cluster.ClusterCommand.complete(ClusterCommand.java:56)
    io.lettuce.core.protocol.CommandHandler.decode(CommandHandler.java:563)
    io.lettuce.core.protocol.CommandHandler.channelRead(CommandHandler.java:516)

--------------------

## 原因分析： ##

> lettuce默认是没有开始拓扑更新及读写分离导致的

--------------------

## 解决方案： ##

> 这里分为几种情况：
> 
> 1.  springboot 1.x之前版本默认使用jedis，无需要手动开启刷新
> 2.  springboot 2.x默认为Lettuce，需要代码设置开启刷新节点拓扑策略
> 3.  springboot 2.3.0开始，支持集群拓扑刷新功能，属性配置开启即可

### 第一种情况：springboot1.x版本环境 ###

springboot1.x之前版本默认使用jedis，无需手动开启动态刷新。

### 第二种情况：springboot2.0~2.3版本环境 ###

> springboot2.0-2.3版本默认使用lettuce，默认不支持属性配置集群拓扑刷新。使用lettuce，需要增加配置类，需要手动开启刷新。

配置类如下：

package com.test.config;
    
    import com.fasterxml.jackson.annotation.JsonAutoDetect;
    import com.fasterxml.jackson.annotation.PropertyAccessor;
    import com.fasterxml.jackson.databind.ObjectMapper;
    import io.lettuce.core.cluster.ClusterClientOptions;
    import io.lettuce.core.cluster.ClusterTopologyRefreshOptions;
    import lombok.extern.java.Log;
    import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
    import org.springframework.cache.annotation.EnableCaching;
    import org.springframework.cache.interceptor.KeyGenerator;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.data.redis.connection.RedisClusterConfiguration;
    import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
    import org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration;
    import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
    import org.springframework.data.redis.connection.lettuce.LettucePoolingClientConfiguration;
    import org.springframework.data.redis.core.RedisTemplate;
    import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
    import org.springframework.data.redis.serializer.RedisSerializer;
    import org.springframework.data.redis.serializer.StringRedisSerializer;
    
    import java.lang.reflect.Method;
    import java.time.Duration;
    
    
    @Log
    @Configuration
    @EnableCaching // 开启缓存支持
    public class RedisConfig {
    
        @Autowired
        private RedisProperties redisProperties;
    
        @Bean(destroyMethod = "destroy")	//销毁这个bean之前调用这个destroy回调方法释放资源
        public LettuceConnectionFactory redisConnectionFactory() {
            // redis单节点
            if (null == redisProperties.getCluster() || null == redisProperties.getCluster().getNodes()) {
                RedisStandaloneConfiguration configuration = new RedisStandaloneConfiguration(redisProperties.getHost(),
                    redisProperties.getPort());
                configuration.setPassword(redisProperties.getPassword());	//RedisPassword.of(redisProperties.getPassword())
    			configuration.setDatabase(redisProperties.getDatabase());
                return new LettuceConnectionFactory(configuration);
            }
    
            // redis集群
            RedisClusterConfiguration redisClusterConfiguration = new RedisClusterConfiguration(redisProperties.getCluster().getNodes());
            redisClusterConfiguration.setPassword(redisProperties.getPassword());	//RedisPassword.of(redisProperties.getPassword())
            redisClusterConfiguration.setMaxRedirects(redisProperties.getCluster().getMaxRedirects());
    
            GenericObjectPoolConfig genericObjectPoolConfig = new GenericObjectPoolConfig();
            genericObjectPoolConfig.setMaxTotal(redisProperties.getLettuce().getPool().getMaxActive());
            genericObjectPoolConfig.setMaxIdle(redisProperties.getLettuce().getPool().getMaxIdle());
            genericObjectPoolConfig.setMinIdle(redisProperties.getLettuce().getPool().getMinIdle());
            genericObjectPoolConfig.setMaxWaitMillis(redisProperties.getLettuce().getPool().getMaxWait().getSeconds());
    
            // 支持自适应集群拓扑刷新和动态刷新源
            ClusterTopologyRefreshOptions clusterTopologyRefreshOptions = ClusterTopologyRefreshOptions.builder()
                .enableAllAdaptiveRefreshTriggers()
                // 开启自适应刷新
                .enableAdaptiveRefreshTrigger()
                // 开启定时刷新
                .enablePeriodicRefresh(Duration.ofSeconds(5))
                .build();
    
            ClusterClientOptions clusterClientOptions = ClusterClientOptions.builder()
                .topologyRefreshOptions(clusterTopologyRefreshOptions).build();
    
            LettuceClientConfiguration lettuceClientConfiguration = LettucePoolingClientConfiguration.builder()
                .poolConfig(genericObjectPoolConfig)	//如果使用默认配置可以注释genericObjectPoolConfig
    //            .readFrom(ReadFrom.SLAVE_PREFERRED)  //读写分离：主写从读模式配置
                .clientOptions(clusterClientOptions).build();
    
            LettuceConnectionFactory lettuceConnectionFactory = new LettuceConnectionFactory(redisClusterConfiguration, lettuceClientConfiguration);
            lettuceConnectionFactory.setShareNativeConnection(false);// 是否允许多个线程操作共用同一个缓存连接，默认 true，false 时每个操作都将开辟新的连接
            lettuceConnectionFactory.resetConnection();// 重置底层共享连接, 在接下来的访问时初始化
            return lettuceConnectionFactory;
        }
    
        /**
         * RedisTemplate配置
         */
        @Bean
        public RedisTemplate<Object, Object> redisTemplate(LettuceConnectionFactory factory) {
            RedisTemplate<Object, Object> redisTemplate = new RedisTemplate<>();
            redisTemplate.setConnectionFactory(factory);
    
            // 使用注解@Bean返回RedisTemplate的时候，同时配置hashkey和hashValue的序列虎方式
            // key采用String的序列化方式
            redisTemplate.setKeySerializer(keySerializer());
            // value使用jackson序列化方式
            redisTemplate.setValueSerializer(valueSerializer());
    
            // hash的key采用String的序列化方式
            redisTemplate.setHashKeySerializer(keySerializer());
            // hash的value使用jackson序列化方式
            redisTemplate.setHashValueSerializer(valueSerializer());
    
            /**必须执行这个函数,初始化RedisTemplate*/
            // 需要先调用afterPropertiesSet方法,此方法是应该是初始化参数和初始化工作。
            redisTemplate.afterPropertiesSet();
            log.info("序列化完成！");
            return redisTemplate;
        }
    
    
    
    
        @Bean
        public KeyGenerator keyGenerator() {
            return new KeyGenerator() {
                @Override
                public Object generate(Object target, Method method, Object... params) {
                    StringBuffer sb = new StringBuffer();
                    sb.append(target.getClass().getName());
                    sb.append(method.getName());
                    for (Object obj : params) {
                        sb.append(obj.toString());
                    }
                    return sb.toString();
                }
            };
        }
    
    
    
        /**
         * key键序列化方式
         *
         * @return RedisSerializer
         */
        private RedisSerializer<String> keySerializer() {
            return new StringRedisSerializer();
        }
    
    
        /**
         * value值序列化方式
         *
         * @return
         */
        private Jackson2JsonRedisSerializer valueSerializer() {
            Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
            ObjectMapper om = new ObjectMapper();
            om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
            om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
            jackson2JsonRedisSerializer.setObjectMapper(om);
            return jackson2JsonRedisSerializer;
        }
    }

配置文件：

#Redis Configuration
    spring.redis.cluster.max-redirects=10
    spring.redis.cluster.nodes=127.0.0.1:8001,127.0.0.1:8002
    spring.redis.timeout=60000ms
    spring.redis.password=
    spring.redis.lettuce.pool.max-active=10
    spring.redis.lettuce.pool.max-idle=8
    spring.redis.lettuce.pool.min-idle=0
    spring.redis.lettuce.pool.max-wait=-1ms

注意：注入LettuceConnectionFactory后，一定要记得注入RedisTemplate，并 redisTemplate.setConnectionFactory(factory);

apache commons-pool2 包提供了一个通用的对象池技术的实现。可以很方便的基于它来实现自己的对象池，比如 DBCP 和 Jedis 他们的内部对象池的实现就是依赖于 commons-pool2 。

<dependency>
        <groupId>org.apache.commons</groupId>
        <artifactId>commons-pool2</artifactId>
        <version>2.8.0</version>
    </dependency>

### 第三种情况：springboot2.3之后版本环境 ###

> springboot2.3之后版本默认使用lettuce，默认支持属性配置开启集群拓扑刷新，其解决方案：属性配置开启即可。

spring.redis.lettuce.cluster.refresh.adaptive= true
    spring.redis.lettuce.cluster.refresh.period=30000    # 30秒自动刷新一次

关联文章：[Spring Boot集成Redis集群报错UnsupportedOperationException-CSDN博客][Spring Boot_Redis_UnsupportedOperationException-CSDN]

[Spring Boot_Redis_UnsupportedOperationException-CSDN]: https://blog.csdn.net/u011974797/article/details/133268647