java - 从 spring 模板连接到加密的 Redis 集群

Question

我正在尝试从 Spring Boot 连接到加密的传输中 ElastiCache 集群以用于 session 存储库。我有适用于未加密集群的代码，但当我打开 ssl 时，我无法让它工作。这是我的代码

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.session.ExpiringSession;
import org.springframework.session.SessionRepository;
import org.springframework.session.data.redis.config.ConfigureRedisAction;
import org.springframework.session.data.redis.config.annotation.web.http.EnableRedisHttpSession;
import org.springframework.session.data.redis.config.annotation.web.http.RedisHttpSessionConfiguration;
import org.springframework.session.web.http.SessionRepositoryFilter;

   @Configuration
   @EnableRedisHttpSession
   @ConditionalOnProperty(value = "spring.session.enabled", havingValue = "true")
   public class RedisSessionConfig extends RedisHttpSessionConfiguration {

    private final String NAMESPACE = "myname";

    public RedisSessionConfig() {

        // when extending RedisHttpSessionConfiguration to override the repository filter
        // we need to manually set the namespace
        this.setRedisNamespace(NAMESPACE);

    }

   @Autowired
   private RedisTemplate<Object, Object> redisTemplate;

   @Bean
    public static ConfigureRedisAction configureRedisAction() {
        return ConfigureRedisAction.NO_OP;
    }

   @Bean
   @Override
   public <S extends ExpiringSession> SessionRepositoryFilter<? extends ExpiringSession> springSessionRepositoryFilter(SessionRepository<S> sessionRepository) {
        return super.springSessionRepositoryFilter(new SafeDeserializationRepository<>(sessionRepository, redisTemplate, NAMESPACE));
    }

}

和配置

spring:
   redis:
        url: mycluster.id.region.cache.amazonaws.com:port

有人知道怎么做吗？

Answer 1

所以我必须添加自己的 LettuceConnectionFactory 实现，然后由于连接速度特别慢，我还需要实现自己的连接池。

这是代码:

@Configuration
class MyLettuceConnectionFactory extends LettuceConnectionFactory {

    @Autowired
    public MyLettuceConnectionFactory(
            @Value("${redis-configuration.clusterEndpoint}") String clusterNodes,
            @Value("${redis-configuration.port}") int port,
            @Value("${redis-configuration.ssl}") boolean ssl,
            @Value("${redis-configuration.pool.minimumIdle}") int minIdle,
            @Value("${redis-configuration.pool.maximumIdle}") int maxIdle
            ) {
        super(new MyLettucePool(clusterNodes, port, ssl, minIdle, maxIdle));
        this.setUseSsl(ssl);
    }

    @Override
    public void afterPropertiesSet() {
        super.afterPropertiesSet();
        DirectFieldAccessor accessor = new DirectFieldAccessor(this);
        AbstractRedisClient client = (AbstractRedisClient) accessor.getPropertyValue("client");
        if(client instanceof RedisClusterClient){
            RedisClusterClient clusterClient = (RedisClusterClient) client;
            clusterClient.setOptions(ClusterClientOptions.builder().validateClusterNodeMembership(false).build());
        }
    }
}

对于我的自定义连接池:

public class MyLettucePool implements LettucePool {

    private RedisClusterClient client;
    private int dbIndex=0;
    private GenericObjectPool<StatefulConnection<byte[], byte[]>> internalPool;


    public MyLettucePool(String clusterEndpoint, int port, boolean useSsl, int minIdle, int maxIdle) {
        RedisURI uri = new RedisURI();
        uri.setSsl(useSsl);
        uri.setPort(port);
        uri.setHost(clusterEndpoint);


        GenericObjectPoolConfig<StatefulConnection<byte[], byte[]>> config = new GenericObjectPoolConfig<>();
        config.setMinIdle(minIdle);
        config.setMaxIdle(maxIdle);

        this.client = RedisClusterClient.create(uri);
        this.client.setOptions(ClusterClientOptions.builder().autoReconnect(true).validateClusterNodeMembership(false).build());
        this.internalPool = ConnectionPoolSupport.createGenericObjectPool(() -> this.client.connect(new ByteArrayCodec()), new GenericObjectPoolConfig());
}


    @Override
    public AbstractRedisClient getClient() {
        return this.client;
    }

    @Override
    @SuppressWarnings("unchecked")
    public StatefulConnection<byte[], byte[]> getResource() {
       try {
            return internalPool.borrowObject();
        } catch (Exception e) {
            throw new PoolException("Could not get a resource from the pool", e);
        }
    }

    @Override
    public void returnBrokenResource(final StatefulConnection<byte[], byte[]> resource) {

        try {
            internalPool.invalidateObject(resource);
        } catch (Exception e) {
            throw new PoolException("Could not invalidate the broken resource", e);
        }
    }


    @Override
    public void returnResource(final StatefulConnection<byte[], byte[]> resource) {
        try {
            internalPool.returnObject(resource);
        } catch (Exception e) {
            throw new PoolException("Could not return the resource to the pool", e);
        }
    }

    @Override
    public void destroy() {
        try {
            client.shutdown();
            internalPool.close();
        } catch (Exception e) {
            throw new PoolException("Could not destroy the pool", e);
        }
    }