快捷導(dǎo)航

spring-data-redis 動態(tài)切換數(shù)據(jù)源的方法

更新時間：2021年08月31日 08:51:02 作者：干貨滿滿張哈希

最近遇到了一個麻煩的需求，我們需要一個微服務(wù)應(yīng)用同時訪問兩個不同的 Redis 集群，一般情況下我們會怎么處理呢，下面通過場景分析給大家介紹spring-data-redis 動態(tài)切換數(shù)據(jù)源的方法，感興趣的朋友一起看看吧

最近遇到了一個麻煩的需求，我們需要一個微服務(wù)應(yīng)用同時訪問兩個不同的 Redis 集群。一般我們不會這么使用 Redis，但是這兩個 Redis 本來是不同業(yè)務(wù)集群，現(xiàn)在需要一個微服務(wù)同時訪問。

其實我們在實際業(yè)務(wù)開發(fā)的時候，可能還會遇到類似的場景。例如 Redis 讀寫分離，這個也是 spring-data-redis 沒有提供的功能，底層連接池例如 Lettuce 或者 Jedis 都提供了獲取只讀連接的 API，但是缺陷有兩個：

上層 spring-data-redis 并沒有封裝這種接口
基于 redis 的架構(gòu)實現(xiàn)的，哨兵模式需要配置 sentinel 的地址，集群模式需要感知集群拓?fù)洌谠圃h(huán)境中，這些都默認(rèn)被云提供商隱藏了，暴露到外面的只有一個個動態(tài) VIP 域名。

因此，我們需要在 spring-data-redis 的基礎(chǔ)上實現(xiàn)一個動態(tài)切換 Redis 連接的機(jī)制。

spring-data-redis 的配置類為：org.springframework.boot.autoconfigure.data.redis.RedisProperties，可以配置單個 Redis 實例或者 Redis 集群的連接配置。根據(jù)這些配置，會生成統(tǒng)一的 Redis 連接工廠 RedisConnectionFactory

spring-data-redis 核心接口與背后的連接相關(guān)抽象關(guān)系為：

通過這個圖，我們可以知道，我們實現(xiàn)一個可以動態(tài)返回不同 Redis 連接的 RedisConnectionFactory 即可，并且根據(jù) spring-data-redis 的自動裝載源碼可以知道，框架內(nèi)的所有 RedisConnectionFactory 是 @ConditionalOnMissingBean 的，即我們可以使用我們自己實現(xiàn)的 RedisConnectionFactory 進(jìn)行替換。

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

我們可以給 RedisProperties 配置外層封裝一個多 Redis 連接的配置，即MultiRedisProperties：

@Data
@NoArgsConstructor
@ConfigurationProperties(prefix = "spring.redis")
public class MultiRedisProperties {
    /**
     * 默認(rèn)連接必須配置，配置 key 為 default
     */
    public static final String DEFAULT = "default";

    private boolean enableMulti = false;
    private Map<String, RedisProperties> multi;
}

這個配置是在原有配置基礎(chǔ)上的，也就是用戶可以使用原有配置，也可以使用這種多 Redis 配置，就是需要配置 spring.redis.enable-multi=true。multi 這個 Map 中放入的 key 是數(shù)據(jù)源名稱，用戶可以在使用 RedisTemplate 或者 ReactiveRedisTemplate 之前，通過這個數(shù)據(jù)源名稱指定用哪個 Redis。

接下來我們來實現(xiàn) MultiRedisLettuceConnectionFactory，即可以動態(tài)切換 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);
    }
}

邏輯非常簡單，就是提供了設(shè)置 Redis 數(shù)據(jù)源的接口，并且放入了 ThreadLocal 中，并且僅對當(dāng)前一次有效，讀取后就清空。

然后，將 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);
    }

}

我們來測試下，使用 embedded-redis 來啟動本地 redis，從而實現(xiàn)單元測試。我們啟動兩個 Redis，在兩個 Redis 中放入不同的 Key，驗證是否存在，并且測試同步接口，多線程調(diào)用同步接口，和多次異步接口無等待訂閱從而測試有效性。：

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) {
        //使用默認(rèn)連接，設(shè)置 "testDefault" + suffix, "testDefault" 鍵值對
        redisTemplate.opsForValue().set("testDefault" + suffix, "testDefault");
        //使用 test 連接，設(shè)置 "testSecond" + suffix, "testDefault" 鍵值對
        multiRedisLettuceConnectionFactory.setCurrentRedis("test");
        redisTemplate.opsForValue().set("testSecond" + suffix, "testSecond");
        //使用默認(rèn)連接，驗證 "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;
                });
    }

    //多次調(diào)用 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);
    }
}

運(yùn)行測試，通過。

到此這篇關(guān)于spring-data-redis 動態(tài)切換數(shù)據(jù)源的文章就介紹到這了,更多相關(guān)spring-data-redis 動態(tài)切換數(shù)據(jù)源內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: