spring-data-redis 动态切换数据源

我会带着你远行 2022-09-09 09:23 217阅读 0赞

![image][]

最近遇到了一个麻烦的需求，我们需要一个微服务应用同时访问两个不同的 Redis 集群。一般我们不会这么使用 Redis，但是这两个 Redis 本来是不同业务集群，现在需要一个微服务同时访问。

其实我们在实际业务开发的时候，可能还会遇到类似的场景。例如 Redis 读写分离，这个也是 spring-data-redis 没有提供的功能，底层连接池例如 Lettuce 或者 Jedis 都提供了获取只读连接的 API，但是缺陷有两个：

*  上层 spring-data-redis 并没有封装这种接口
 *  基于 redis 的架构实现的，哨兵模式需要配置 sentinel 的地址，集群模式需要感知集群拓扑，在云原生环境中，这些都默认被云提供商隐藏了，暴露到外面的只有一个个动态 VIP 域名。

因此，我们需要在 spring-data-redis 的基础上**实现一个动态切换 Redis 连接的机制**。

![image][image 1]

spring-data-redis 的配置类为：`org.springframework.boot.autoconfigure.data.redis.RedisProperties`，可以配置单个 Redis 实例或者 Redis 集群的连接配置。根据这些配置，会生成统一的 Redis 连接工厂 `RedisConnectionFactory`

spring-data-redis 核心接口与背后的连接相关抽象关系为：

![image][image 2]

通过这个图，我们可以知道，我们实现一个可以动态返回不同 Redis 连接的 `RedisConnectionFactory` 即可，并且根据 spring-data-redis 的自动装载源码可以知道，框架内的所有 `RedisConnectionFactory` 是 `@ConditionalOnMissingBean` 的，即我们可以使用我们自己实现的 `RedisConnectionFactory` 进行替换。

![image][image 3]

> 项目地址：https://github.com/JoJoTec/spring-boot-starter-redis-related

我们可以给 `RedisProperties` 配置外层封装一个多 Redis 连接的配置，即`MultiRedisProperties`：

@Data
    @NoArgsConstructor
    @ConfigurationProperties(prefix = "spring.redis")
    public class MultiRedisProperties {
        /**
         * 默认连接必须配置，配置 key 为 default
         */
        public static final String DEFAULT = "default";
    
        private boolean enableMulti = false;
        private Map<String, RedisProperties> multi;
    }

这个配置是在原有配置基础上的，也就是用户可以使用原有配置，也可以使用这种多 Redis 配置，就是需要配置 `spring.redis.enable-multi=true`。**multi 这个 Map 中放入的 key 是数据源名称，用户可以在使用 RedisTemplate 或者 ReactiveRedisTemplate 之前，通过这个数据源名称指定用哪个 Redis**。

接下来我们来实现 `MultiRedisLettuceConnectionFactory`，即可以动态切换 Redis 连接的 `RedisConnectionFactory`，我们的项目采用的 Redis 客户端是 Lettuce：

public class MultiRedisLettuceConnectionFactory
            implements InitializingBean, DisposableBean, RedisConnectionFactory, ReactiveRedisConnectionFactory {
        private final Map<String, LettuceConnectionFactory> connectionFactoryMap;
    
        private static final ThreadLocal<String> currentRedis = new ThreadLocal<>();
    
        public MultiRedisLettuceConnectionFactory(Map<String, LettuceConnectionFactory> connectionFactoryMap) {
            this.connectionFactoryMap = connectionFactoryMap;
        }
    
        public void setCurrentRedis(String currentRedis) {
            if (!connectionFactoryMap.containsKey(currentRedis)) {
                throw new RedisRelatedException("invalid currentRedis: " + currentRedis + ", it does not exists in configuration");
            }
            MultiRedisLettuceConnectionFactory.currentRedis.set(currentRedis);
        }
    
        @Override
        public void destroy() throws Exception {
            connectionFactoryMap.values().forEach(LettuceConnectionFactory::destroy);
        }
    
        @Override
        public void afterPropertiesSet() throws Exception {
            connectionFactoryMap.values().forEach(LettuceConnectionFactory::afterPropertiesSet);
        }
    
        private LettuceConnectionFactory currentLettuceConnectionFactory() {
            String currentRedis = MultiRedisLettuceConnectionFactory.currentRedis.get();
            if (StringUtils.isNotBlank(currentRedis)) {
                MultiRedisLettuceConnectionFactory.currentRedis.remove();
                return connectionFactoryMap.get(currentRedis);
            }
            return connectionFactoryMap.get(MultiRedisProperties.DEFAULT);
        }
    
        @Override
        public ReactiveRedisConnection getReactiveConnection() {
            return currentLettuceConnectionFactory().getReactiveConnection();
        }
    
        @Override
        public ReactiveRedisClusterConnection getReactiveClusterConnection() {
            return currentLettuceConnectionFactory().getReactiveClusterConnection();
        }
    
        @Override
        public RedisConnection getConnection() {
            return currentLettuceConnectionFactory().getConnection();
        }
    
        @Override
        public RedisClusterConnection getClusterConnection() {
            return currentLettuceConnectionFactory().getClusterConnection();
        }
    
        @Override
        public boolean getConvertPipelineAndTxResults() {
            return currentLettuceConnectionFactory().getConvertPipelineAndTxResults();
        }
    
        @Override
        public RedisSentinelConnection getSentinelConnection() {
            return currentLettuceConnectionFactory().getSentinelConnection();
        }
    
        @Override
        public DataAccessException translateExceptionIfPossible(RuntimeException ex) {
            return currentLettuceConnectionFactory().translateExceptionIfPossible(ex);
        }
    }

逻辑非常简单，就是提供了设置 Redis 数据源的接口，并且放入了 ThreadLocal 中，并且仅对当前一次有效，读取后就清空。

然后，将 MultiRedisLettuceConnectionFactory 作为 Bean 注册到我们的 ApplicationContext 中：

@ConditionalOnProperty(prefix = "spring.redis", value = "enable-multi", matchIfMissing = false)
    @Configuration(proxyBeanMethods = false)
    public class RedisCustomizedConfiguration {
    
        /**
         * @param builderCustomizers
         * @param clientResources
         * @param multiRedisProperties
         * @return
         * @see org.springframework.boot.autoconfigure.data.redis.LettuceConnectionConfiguration
         */
        @Bean
        public MultiRedisLettuceConnectionFactory multiRedisLettuceConnectionFactory(
                ObjectProvider<LettuceClientConfigurationBuilderCustomizer> builderCustomizers,
                ClientResources clientResources,
                MultiRedisProperties multiRedisProperties,
                ObjectProvider<RedisSentinelConfiguration> sentinelConfigurationProvider,
                ObjectProvider<RedisClusterConfiguration> clusterConfigurationProvider
        ) {
            //读取配置
            Map<String, LettuceConnectionFactory> connectionFactoryMap = Maps.newHashMap();
            Map<String, RedisProperties> multi = multiRedisProperties.getMulti();
            multi.forEach((k, v) -> {
                //这个其实就是框架中原有的源码使用 RedisProperties 的方式，我们其实就是在 RedisProperties 外面包装了一层而已
                LettuceConnectionConfiguration lettuceConnectionConfiguration = new LettuceConnectionConfiguration(
                        v,
                        sentinelConfigurationProvider,
                        clusterConfigurationProvider
                );
                LettuceConnectionFactory lettuceConnectionFactory = lettuceConnectionConfiguration.redisConnectionFactory(builderCustomizers, clientResources);
                connectionFactoryMap.put(k, lettuceConnectionFactory);
            });
            return new MultiRedisLettuceConnectionFactory(connectionFactoryMap);
        }
    
    }

![image][image 4]

我们来测试下，使用 embedded-redis 来启动本地 redis，从而实现单元测试。我们启动两个 Redis，在两个 Redis 中放入不同的 Key，验证是否存在，并且测试同步接口，多线程调用同步接口，和多次异步接口无等待订阅从而测试有效性。：

import com.github.jojotech.spring.boot.starter.redis.related.lettuce.MultiRedisLettuceConnectionFactory;
    import org.junit.jupiter.api.AfterAll;
    import org.junit.jupiter.api.Assertions;
    import org.junit.jupiter.api.BeforeAll;
    import org.junit.jupiter.api.Test;
    import org.junit.jupiter.api.extension.ExtendWith;
    import org.redisson.api.RedissonClient;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
    import org.springframework.data.redis.core.StringRedisTemplate;
    import org.springframework.test.context.junit.jupiter.SpringExtension;
    import reactor.core.publisher.Mono;
    import redis.embedded.RedisServer;
    
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.atomic.AtomicBoolean;
    
    @ExtendWith(SpringExtension.class)
    @SpringBootTest(properties = {
            "spring.redis.enable-multi=true",
            "spring.redis.multi.default.host=127.0.0.1",
            "spring.redis.multi.default.port=6379",
            "spring.redis.multi.test.host=127.0.0.1",
            "spring.redis.multi.test.port=6380",
    })
    public class MultiRedisTest {
        //启动两个 redis
        private static RedisServer redisServer;
        private static RedisServer redisServer2;
    
        @BeforeAll
        public static void setUp() throws Exception {
            System.out.println("start redis");
            redisServer = RedisServer.builder().port(6379).setting("maxheap 200m").build();
            redisServer2 = RedisServer.builder().port(6380).setting("maxheap 200m").build();
            redisServer.start();
            redisServer2.start();
            System.out.println("redis started");
        }
    
        @AfterAll
        public static void tearDown() throws Exception {
            System.out.println("stop redis");
            redisServer.stop();
            redisServer2.stop();
            System.out.println("redis stopped");
        }
    
        @EnableAutoConfiguration
        @Configuration
        public static class App {
        }
    
        @Autowired
        private StringRedisTemplate redisTemplate;
        @Autowired
        private ReactiveStringRedisTemplate reactiveRedisTemplate;
        @Autowired
        private MultiRedisLettuceConnectionFactory multiRedisLettuceConnectionFactory;
    
        private void testMulti(String suffix) {
            //使用默认连接，设置 "testDefault" + suffix, "testDefault" 键值对
            redisTemplate.opsForValue().set("testDefault" + suffix, "testDefault");
            //使用 test 连接，设置 "testSecond" + suffix, "testDefault" 键值对
            multiRedisLettuceConnectionFactory.setCurrentRedis("test");
            redisTemplate.opsForValue().set("testSecond" + suffix, "testSecond");
            //使用默认连接，验证 "testDefault" + suffix 存在，"testSecond" + suffix 不存在
            Assertions.assertTrue(redisTemplate.hasKey("testDefault" + suffix));
            Assertions.assertFalse(redisTemplate.hasKey("testSecond" + suffix));
            //使用 test 连接，验证 "testDefault" + suffix 不存在，"testSecond" + suffix 存在
            multiRedisLettuceConnectionFactory.setCurrentRedis("test");
            Assertions.assertFalse(redisTemplate.hasKey("testDefault" + suffix));
            multiRedisLettuceConnectionFactory.setCurrentRedis("test");
            Assertions.assertTrue(redisTemplate.hasKey("testSecond" + suffix));
        }
    
        //单次验证
        @Test
        public void testMultiBlock() {
            testMulti("");
        }
    
        //多线程验证
        @Test
        public void testMultiBlockMultiThread() throws InterruptedException {
            Thread thread[] = new Thread[50];
            AtomicBoolean result = new AtomicBoolean(true);
            for (int i = 0; i < thread.length; i++) {
                int finalI = i;
                thread[i] = new Thread(() -> {
                    try {
                        testMulti("" + finalI);
                    } catch (Exception e) {
                        e.printStackTrace();
                        result.set(false);
                    }
                });
            }
            for (int i = 0; i < thread.length; i++) {
                thread[i].start();
            }
            for (int i = 0; i < thread.length; i++) {
                thread[i].join();
            }
            Assertions.assertTrue(result.get());
        }
    
        //reactive 接口验证
        private Mono<Boolean> reactiveMulti(String suffix) {
            return reactiveRedisTemplate.opsForValue().set("testReactiveDefault" + suffix, "testReactiveDefault")
                    .flatMap(b -> {
                        multiRedisLettuceConnectionFactory.setCurrentRedis("test");
                        return reactiveRedisTemplate.opsForValue().set("testReactiveSecond" + suffix, "testReactiveSecond");
                    }).flatMap(b -> {
                        return reactiveRedisTemplate.hasKey("testReactiveDefault" + suffix);
                    }).map(b -> {
                        Assertions.assertTrue(b);
                        System.out.println(Thread.currentThread().getName());
                        return b;
                    }).flatMap(b -> {
                        return reactiveRedisTemplate.hasKey("testReactiveSecond" + suffix);
                    }).map(b -> {
                        Assertions.assertFalse(b);
                        System.out.println(Thread.currentThread().getName());
                        return b;
                    }).flatMap(b -> {
                        multiRedisLettuceConnectionFactory.setCurrentRedis("test");
                        return reactiveRedisTemplate.hasKey("testReactiveDefault" + suffix);
                    }).map(b -> {
                        Assertions.assertFalse(b);
                        System.out.println(Thread.currentThread().getName());
                        return b;
                    }).flatMap(b -> {
                        multiRedisLettuceConnectionFactory.setCurrentRedis("test");
                        return reactiveRedisTemplate.hasKey("testReactiveSecond" + suffix);
                    }).map(b -> {
                        Assertions.assertTrue(b);
                        return b;
                    });
        }
    
        //多次调用 reactive 验证，并且 subscribe，这本身就是多线程的
        @Test
        public void testMultiReactive() throws InterruptedException {
            for (int i = 0; i < 10000; i++) {
                reactiveMulti("" + i).subscribe(System.out::println);
            }
            TimeUnit.SECONDS.sleep(10);
        }
    }

运行测试，通过。

> **微信搜索“我的编程喵”关注公众号，每日一刷，轻松提升技术，斩获各种offer**：

![3f58a37c57d531d20ebc9a795bc54dcf.gif][]

[image]: /images/20220829/77fcfa7ba95d440b8a37109651d9e9c9.png
[image 1]: /images/20220829/83bb018a624f4c0484ca9bc492059cc9.png
[image 2]: /images/20220829/b2a979092478451a9ecc470bfcb0db74.png
[image 3]: /images/20220829/fd6639af2e53493da650fe0a8adf8448.png
[image 4]: /images/20220829/8fc8f8c8852b4a128af551c380a175d7.png
[3f58a37c57d531d20ebc9a795bc54dcf.gif]: /images/20220829/41fb1ef98fa94d168994e2e0d41b4914.png