Spring Boot 2.3.0 Redis拓扑动态感应，使用Lettuce拓扑刷新

2023-05-16

背景

关于 Redis 在生产中我们一般情况下都会选择 redis cluster 高可用架构部署，既能保证数据分片并且实现节点的故障自动转移。基本部署拓扑如下：

创建测试集群

这里通过我封装的 pig4cloud/redis-cluster:4.0 镜像，即可构建一个 6 个节点的 redis cluster 测试环境。

docker run --name redis-cluster -d -e CLUSTER_ANNOUNCE_IP=宿主机IP \
-p 7000-7005:7000-7005 -p 17000-17005:17000-17005  pig4cloud/redis-cluster:4.0
复制代码

查看集群节点信息

⋊> ./redis-cli -h 172.17.0.111 -p 7000 -c                     16:09:48
172.17.0.111:7000> cluster nodes
3d882206d40935beef84ff564b538d57369e4fd9 172.17.0.111:7003@17003 slave b8d24150df4a221c1045cd9a0696bd1972912d52 0 1591344590000 4 connected
b8d24150df4a221c1045cd9a0696bd1972912d52 172.17.0.111:7001@17001 master - 0 1591344590513 2 connected 5461-10922
c21167a6da7f8af31d2dd612d449cdf92ad2e7e9 172.17.0.111:7005@17005 slave 810baa140db6e008a137708f09d4335f5207ede3 0 1591344591000 6 connected
810baa140db6e008a137708f09d4335f5207ede3 172.17.0.111:7000@17000 myself,master - 0 1591344590000 1 connected 0-5460
05d2f9884d350a50ac9e38f575b57f19e864e74c 172.17.0.111:7004@17004 slave b3cf24a918d96a1949f49a1d7b3a965ff9dc858c 0 1591344590011 5 connected
b3cf24a918d96a1949f49a1d7b3a965ff9dc858c 172.17.0.111:7002@17002 master - 0 1591344591617 3 connected 10923-16383
复制代码

应用层接入集群

这里使用 spring boot 2.2 演示，默认的连接池使用 lettuce

spring:
  redis:
    cluster:
      nodes:
        - 172.17.0.111:7000
        - 172.17.0.111:7001
        - 172.17.0.111:7002
        - 172.17.0.111:7003
        - 172.17.0.111:7004
        - 172.17.0.111:7005
复制代码

简单使用 redisTemplate 操作集群

@RestController
public class DemoController {
    @Autowired
    private RedisTemplate redisTemplate;

    @GetMapping("/add")
    public String redis() {
        redisTemplate.opsForValue().set("k1", "v1");
        return "ok";
    }
}
复制代码

调用查看日志

⋊> curl http://localhost:8080/add
ok⏎
复制代码

我们会发现操作 k1 是在 7000 节点进行操作写入

[channel=0x5ff7aa8f, /172.17.0.156:50783 -> /172.17.0.111:7000, epid=0x8] write() writeAndFlush command ClusterCommand [command=AsyncCommand [type=SET, output=StatusOutput [output=null, error='null'], commandType=io.lettuce.core.protocol.Command], redirections=0, maxRedirections=5]
[channel=0x5ff7aa8f, /172.17.0.156:50783 -> /172.17.0.111:7000, epid=0x8] write() done
复制代码

模拟单点故障

关闭 7000 节点

./redis-cli -h 172.17.0.111 -p 7000 -c
172.17.0.111:7000> SHUTDOWN
复制代码

查看 redis cluster 集群日志 docker logs -f redis-cluster

我们可以看到此时集群选举完毕，完成故障转移

23:S 05 Jun 08:24:49.387 # Starting a failover election for epoch 7.
29:M 05 Jun 08:24:49.388 # Failover auth granted to c21167a6da7f8af31d2dd612d449cdf92ad2e7e9 for epoch 7
26:M 05 Jun 08:24:49.388 # Failover auth granted to c21167a6da7f8af31d2dd612d449cdf92ad2e7e9 for epoch 7
23:S 05 Jun 08:24:49.389 # Failover election won: I'm the new master.
23:S 05 Jun 08:24:49.389 # configEpoch set to 7 after successful failover
23:M 05 Jun 08:24:49.389 # Setting secondary replication ID to 5253748ecf5bd7ab3536058fba8cad62d2d5e825, valid up to offset: 1622. New replication ID is 21d6a0b199a1ba655c0279d9c78f9682477ac9a3
23:M 05 Jun 08:24:49.389 * Discarding previously cached master state.
23:M 05 Jun 08:24:49.390 # Cluster state changed: ok
复制代码

此时集群状态。7005 从 slave 变更为 master

应用层日志

大量输出连接 7000 节点异常

io.netty.channel.AbstractChannel$AnnotatedConnectException: Connection refused: /172.17.0.111:7000
Caused by: java.net.ConnectException: Connection refused

io.netty.channel.AbstractChannel$AnnotatedConnectException: Connection refused: /172.17.0.111:7000
Caused by: java.net.ConnectException: Connection refused


io.netty.channel.AbstractChannel$AnnotatedConnectException: Connection refused: /172.17.0.111:7000
Caused by: java.net.ConnectException: Connection refused
复制代码

再次操作 redisTemplate 会发现卡死，等待结果返回

⋊> curl http://localhost:8080/add
复制代码

原因分析

此时还是操作 k1, 根据 slot 对应连接到 7000 节点，已经连接不到无限尝试重连的问题。 lettuce 客户端并未和 redis cluster 集群状态同步刷新，把宕机节点移除，完成故障转移。

集群拓扑动态感应

拓扑动态感应即客户端能够根据 redis cluster 集群的变化，动态改变客户端的节点情况，完成故障转移。

我们只需要在 spring boot 2.3.0 版本中开启此特性即可。

spring.redis.lettuce.cluster.refresh.adaptive=true
spring.redis.lettuce.cluster.refresh.period=10000  #单位毫秒

spring:
  redis:
    lettuce:
      cluster:
        refresh:
          adaptive: true

其实 lettuce 官方一直有这个功能，但 spring data redis 并未跟进，具体内容可以参考 user-content-refreshing-the-cluster-topology-view 章节

其他解决方案

解决方案1：

升级到SpringBoot2.3.0或以上版本。并添加如下配置项

spring.redis.timeout=60s
spring.redis.lettuce.cluster.refresh.period=60s
spring.redis.lettuce.cluster.refresh.adaptive=true

解决方案2：
配置LettuceConnectionFactory ，设置拓扑刷新策略。

@Bean
public DefaultClientResources lettuceClientResources() {
    return DefaultClientResources.create();
}

@Bean
public LettuceConnectionFactory lettuceConnectionFactory(RedisProperties redisProperties, ClientResources clientResources) {

    ClusterTopologyRefreshOptions topologyRefreshOptions = ClusterTopologyRefreshOptions.builder()
            .enablePeriodicRefresh(Duration.ofSeconds(30)) //按照周期刷新拓扑
            .enableAllAdaptiveRefreshTriggers() //根据事件刷新拓扑
            .build();

    ClusterClientOptions clusterClientOptions = ClusterClientOptions.builder()
            //redis命令超时时间,超时后才会使用新的拓扑信息重新建立连接
            .timeoutOptions(TimeoutOptions.enabled(Duration.ofSeconds(10)))
            .topologyRefreshOptions(topologyRefreshOptions)
            .build();

    LettuceClientConfiguration clientConfiguration = LettuceClientConfiguration.builder()
            .clientResources(clientResources)
            .clientOptions(clusterClientOptions)
            .build();

    RedisClusterConfiguration clusterConfig = new RedisClusterConfiguration(redisProperties.getCluster().getNodes());
    clusterConfig.setMaxRedirects(redisProperties.getCluster().getMaxRedirects());
    clusterConfig.setPassword(RedisPassword.of(redisProperties.getPassword()));

    LettuceConnectionFactory lettuceConnectionFactory = new LettuceConnectionFactory(clusterConfig, clientConfiguration);

    return lettuceConnectionFactory;
}

解决方案3:

将spring-boot-starter-data-redis依赖的Lettuce，修改成Jedis。

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
    <exclusions>
        <exclusion>
            <groupId>io.lettuce</groupId>
            <artifactId>lettuce-core</artifactId>
        </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
</dependency>

旧版本兼容

我们只需要参考 adaptive 开关打开后做了哪些事情，给自己的项目配置上 topology-view 即可

例子1

	@Bean
	public LettuceConnectionFactory redisConnectionFactory(RedisProperties redisProperties) {
		RedisClusterConfiguration redisClusterConfiguration = new RedisClusterConfiguration(redisProperties.getCluster().getNodes());

		// https://github.com/lettuce-io/lettuce-core/wiki/Redis-Cluster#user-content-refreshing-the-cluster-topology-view
		ClusterTopologyRefreshOptions clusterTopologyRefreshOptions = ClusterTopologyRefreshOptions.builder()
				.enablePeriodicRefresh()
				.enableAllAdaptiveRefreshTriggers()
				.refreshPeriod(Duration.ofSeconds(5))
				.build();

		ClusterClientOptions clusterClientOptions = ClusterClientOptions.builder()
				.topologyRefreshOptions(clusterTopologyRefreshOptions).build();

		// https://github.com/lettuce-io/lettuce-core/wiki/ReadFrom-Settings
		LettuceClientConfiguration lettuceClientConfiguration = LettuceClientConfiguration.builder()
				.readFrom(ReadFrom.REPLICA_PREFERRED)
				.clientOptions(clusterClientOptions).build();

		return new LettuceConnectionFactory(redisClusterConfiguration, lettuceClientConfiguration);
	}

例子2

https://github.com/lettuce-io/lettuce-core/wiki/Client-Options

这里描述了很多特殊场景下设置的客户端选项，可以视自身情况去设置调整

 @Autowired
    private RedisProperties redisProperties;
 
    @Bean
    public GenericObjectPoolConfig<?> genericObjectPoolConfig(Pool properties) {
        GenericObjectPoolConfig<?> config = new GenericObjectPoolConfig<>();
        config.setMaxTotal(properties.getMaxActive());
        config.setMaxIdle(properties.getMaxIdle());
        config.setMinIdle(properties.getMinIdle());
        if (properties.getTimeBetweenEvictionRuns() != null) {
            config.setTimeBetweenEvictionRunsMillis(properties.getTimeBetweenEvictionRuns().toMillis());
        }
        if (properties.getMaxWait() != null) {
            config.setMaxWaitMillis(properties.getMaxWait().toMillis());
        }
        return config;
    }
    
    @Bean(destroyMethod = "destroy")
    public LettuceConnectionFactory lettuceConnectionFactory() {
        
        //开启 自适应集群拓扑刷新和周期拓扑刷新
        ClusterTopologyRefreshOptions clusterTopologyRefreshOptions =  ClusterTopologyRefreshOptions.builder()
                // 开启全部自适应刷新
                .enableAllAdaptiveRefreshTriggers() // 开启自适应刷新,自适应刷新不开启,Redis集群变更时将会导致连接异常
                // 自适应刷新超时时间(默认30秒)
                .adaptiveRefreshTriggersTimeout(Duration.ofSeconds(30)) //默认关闭开启后时间为30秒
                // 开周期刷新 
                .enablePeriodicRefresh(Duration.ofSeconds(20))  // 默认关闭开启后时间为60秒 ClusterTopologyRefreshOptions.DEFAULT_REFRESH_PERIOD 60  .enablePeriodicRefresh(Duration.ofSeconds(2)) = .enablePeriodicRefresh().refreshPeriod(Duration.ofSeconds(2))
                .build();
        
        // https://github.com/lettuce-io/lettuce-core/wiki/Client-Options
        ClientOptions clientOptions = ClusterClientOptions.builder()
                .topologyRefreshOptions(clusterTopologyRefreshOptions)
                .build();
 
        LettuceClientConfiguration clientConfig = LettucePoolingClientConfiguration.builder()
                .poolConfig(genericObjectPoolConfig(redisProperties.getLettuce().getPool()))
                //.readFrom(ReadFrom.MASTER_PREFERRED)
                .clientOptions(clientOptions)
                .commandTimeout(redisProperties.getTimeout()) //默认RedisURI.DEFAULT_TIMEOUT 60  
                .build();
        
        List<String> clusterNodes = redisProperties.getCluster().getNodes();
        Set<RedisNode> nodes = new HashSet<RedisNode>();
        clusterNodes.forEach(address -> nodes.add(new RedisNode(address.split(":")[0].trim(), Integer.valueOf(address.split(":")[1]))));
        
        RedisClusterConfiguration clusterConfiguration = new RedisClusterConfiguration();
        clusterConfiguration.setClusterNodes(nodes);
        clusterConfiguration.setPassword(RedisPassword.of(redisProperties.getPassword()));
        clusterConfiguration.setMaxRedirects(redisProperties.getCluster().getMaxRedirects());
        
        LettuceConnectionFactory lettuceConnectionFactory = new LettuceConnectionFactory(clusterConfiguration, clientConfig);
        // lettuceConnectionFactory.setShareNativeConnection(false); //是否允许多个线程操作共用同一个缓存连接，默认true，false时每个操作都将开辟新的连接
        // lettuceConnectionFactory.resetConnection(); // 重置底层共享连接, 在接下来的访问时初始化
        return lettuceConnectionFactory;
    }

开启自适应刷新并设定刷新频率

例子3

继续查找，发现了Lettuce官方给了开启拓扑刷新的代码例子

# Example 37. Enabling periodic cluster topology view updates
RedisClusterClient clusterClient = RedisClusterClient.create(RedisURI.create("localhost", 6379));

ClusterTopologyRefreshOptions topologyRefreshOptions = ClusterTopologyRefreshOptions.builder()
                                .enablePeriodicRefresh(10, TimeUnit.MINUTES)
                                .build();

clusterClient.setOptions(ClusterClientOptions.builder()
                                .topologyRefreshOptions(topologyRefreshOptions)
                                .build());
...

clusterClient.shutdown();


# Example 38. Enabling adaptive cluster topology view updates
RedisURI node1 = RedisURI.create("node1", 6379);
RedisURI node2 = RedisURI.create("node2", 6379);

RedisClusterClient clusterClient = RedisClusterClient.create(Arrays.asList(node1, node2));

ClusterTopologyRefreshOptions topologyRefreshOptions = ClusterTopologyRefreshOptions.builder()
                                .enableAdaptiveRefreshTrigger(RefreshTrigger.MOVED_REDIRECT, RefreshTrigger.PERSISTENT_RECONNECTS)
                                .adaptiveRefreshTriggersTimeout(30, TimeUnit.SECONDS)
                                .build();

clusterClient.setOptions(ClusterClientOptions.builder()
                                .topologyRefreshOptions(topologyRefreshOptions)
                                .build());
...

clusterClient.shutdown();
复制代码

根据示例我们修改一下我们的项目代码

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 org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
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.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisNode;
import org.springframework.data.redis.connection.RedisPassword;
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.StringRedisSerializer;


import java.time.Duration;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

@Configuration
public class LettuceRedisConfig {


    @Autowired
    private RedisProperties redisProperties;

    /**
     * 配置RedisTemplate
     * 【Redis配置最终一步】
     *
     * @param lettuceConnectionFactoryUvPv redis连接工厂实现
     * @return 返回一个可以使用的RedisTemplate实例
     */
    @Bean
    public RedisTemplate redisTemplate(@Qualifier("lettuceConnectionFactoryUvPv") RedisConnectionFactory lettuceConnectionFactoryUvPv) {
        RedisTemplate<String, Object> template = new RedisTemplate<String, Object>();
        template.setConnectionFactory(lettuceConnectionFactoryUvPv);
        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);
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
        // key采用String的序列化方式
        template.setKeySerializer(stringRedisSerializer);
        // hash的key也采用String的序列化方式
        template.setHashKeySerializer(stringRedisSerializer);
        // value序列化方式采用jackson
        template.setValueSerializer(jackson2JsonRedisSerializer);
        // hash的value序列化方式采用jackson
        template.setHashValueSerializer(jackson2JsonRedisSerializer);
        template.afterPropertiesSet();
        return template;
    }


    /**
     * 为RedisTemplate配置Redis连接工厂实现
     * LettuceConnectionFactory实现了RedisConnectionFactory接口
     * UVPV用Redis
     *
     * @return 返回LettuceConnectionFactory
     */
    @Bean(destroyMethod = "destroy")
    //这里要注意的是，在构建LettuceConnectionFactory 时，如果不使用内置的destroyMethod，可能会导致Redis连接早于其它Bean被销毁
    public LettuceConnectionFactory lettuceConnectionFactoryUvPv() throws Exception {

        List<String> clusterNodes = redisProperties.getCluster().getNodes();
        Set<RedisNode> nodes = new HashSet<RedisNode>();
        clusterNodes.forEach(address -> nodes.add(new RedisNode(address.split(":")[0].trim(), Integer.valueOf(address.split(":")[1]))));
        RedisClusterConfiguration clusterConfiguration = new RedisClusterConfiguration();
        clusterConfiguration.setClusterNodes(nodes);
        clusterConfiguration.setPassword(RedisPassword.of(redisProperties.getPassword()));
        clusterConfiguration.setMaxRedirects(redisProperties.getCluster().getMaxRedirects());

        GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();
        poolConfig.setMaxIdle(redisProperties.getLettuce().getPool().getMaxIdle());
        poolConfig.setMinIdle(redisProperties.getLettuce().getPool().getMinIdle());
        poolConfig.setMaxTotal(redisProperties.getLettuce().getPool().getMaxActive());

        return new LettuceConnectionFactory(clusterConfiguration, getLettuceClientConfiguration(poolConfig));
    }

    /**
     * 配置LettuceClientConfiguration 包括线程池配置和安全项配置
     *
     * @param genericObjectPoolConfig common-pool2线程池
     * @return lettuceClientConfiguration
     */
    private LettuceClientConfiguration getLettuceClientConfiguration(GenericObjectPoolConfig genericObjectPoolConfig) {
        /*
        ClusterTopologyRefreshOptions配置用于开启自适应刷新和定时刷新。如自适应刷新不开启，Redis集群变更时将会导致连接异常！
         */
        ClusterTopologyRefreshOptions topologyRefreshOptions = ClusterTopologyRefreshOptions.builder()
                //开启自适应刷新
                //.enableAdaptiveRefreshTrigger(ClusterTopologyRefreshOptions.RefreshTrigger.MOVED_REDIRECT, ClusterTopologyRefreshOptions.RefreshTrigger.PERSISTENT_RECONNECTS)
                //开启所有自适应刷新，MOVED，ASK，PERSISTENT都会触发
                .enableAllAdaptiveRefreshTriggers()
                // 自适应刷新超时时间(默认30秒)
                .adaptiveRefreshTriggersTimeout(Duration.ofSeconds(25)) //默认关闭开启后时间为30秒
                // 开周期刷新
                .enablePeriodicRefresh(Duration.ofSeconds(20))  // 默认关闭开启后时间为60秒 ClusterTopologyRefreshOptions.DEFAULT_REFRESH_PERIOD 60  .enablePeriodicRefresh(Duration.ofSeconds(2)) = .enablePeriodicRefresh().refreshPeriod(Duration.ofSeconds(2))
                .build();
        return LettucePoolingClientConfiguration.builder()
                .poolConfig(genericObjectPoolConfig)
                .clientOptions(ClusterClientOptions.builder().topologyRefreshOptions(topologyRefreshOptions).build())
                //将appID传入连接，方便Redis监控中查看
                //.clientName(appName + "_lettuce")
                .build();
    }
}
复制代码

之后在工具类中注入redisTemplate类即可

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

参考：

https://juejin.im/post/5ee183b26fb9a047aa66065a

https://blog.csdn.net/ankeway/article/details/100136675

https://juejin.im/post/5e12e39cf265da5d381d0f00

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)

Spring

Boot

Redis

Lettuce

拓扑动态感应