import com.lambdaworks.redis.RedisURI; //导入依赖的package包/类
@Override
public void init(AbstractConfiguration config) {
    if (!config.getString("redis.type").equals("master_slave")) {
        throw new IllegalStateException("RedisSyncSingleStorageImpl class can only be used with master slave redis setup, but redis.type value is " + config.getString("redis.type"));
    }

    List<String> address = parseRedisAddress(config.getString("redis.address"), 6379);
    int databaseNumber = config.getInt("redis.database", 0);
    String password = StringUtils.isNotEmpty(config.getString("redis.password")) ? config.getString("redis.password") + "@" : "";

    // lettuce
    RedisURI lettuceURI = RedisURI.create("redis://" + password + address.get(0) + "/" + databaseNumber);
    this.lettuceMasterSlave = RedisClient.create(lettuceURI);
    this.lettuceMasterSlaveConn = MasterSlave.connect(this.lettuceMasterSlave, new Utf8StringCodec(), lettuceURI);
    this.lettuceMasterSlaveConn.setReadFrom(ReadFrom.valueOf(config.getString("redis.read")));

    // params
    initParams(config);
}
 

import com.lambdaworks.redis.RedisURI; //导入依赖的package包/类
/**
 * Connects to the Redis server.
 * @param configuration The configuration object.
 * @throws Exception If an error occurs.
 */
public void connect(Configuration configuration) throws Exception {
    if(sync != null) {
        return;
    }
    RedisURI.Builder uri = RedisURI.Builder.redis(configuration.getRedisHost(), configuration.getRedisPort())
            .withDatabase(configuration.getRedisIndex());
    if(!configuration.getRedisAuth().isEmpty()) {
        uri.withPassword(configuration.getRedisAuth());
    }
    RedisClient client = RedisClient.create(uri.build());
    StatefulRedisConnection<String, String> connection = client.connect();
    sync = connection.sync();
    for(Category category : Category.values()) {
        logger.info("Registered the category {} with the type {}.", category, category.getEntry().getType());
        category.getEntry().setCategory(category);
    }
}
 

import com.lambdaworks.redis.RedisURI; //导入依赖的package包/类
private void startUp() {
    logger.info("Starting RedisLettuceService");
    {
        logger.debug("Redis settings:");
        logger.debug("* host={}", cfgHost);
        logger.debug("* port={}", cfgPort);
        logger.debug("* db={}", cfgDb);
    }

    RedisURI redisURI = new RedisURI();
    redisURI.setHost(cfgHost);
    redisURI.setPort(cfgPort);
    redisURI.setDatabase(cfgDb);
    client = RedisClient.create(redisURI);
    connection = client.connect();
    asyncConnection = client.connect();
    syncCommands = connection.sync();
    asyncCommands = asyncConnection.async();
    asyncConnection.setAutoFlushCommands(false);

    logger.info("Connected to a redis client");
}
 

import com.lambdaworks.redis.RedisURI; //导入依赖的package包/类
private static RedisClient findRedisConnection(String host, int port) {
	ClientResources clientResources = DefaultClientResources.builder()//
			.reconnectDelay(Delay.constant(10, TimeUnit.SECONDS))//
			.build();

	RedisURI redisUri = new RedisURI(host, port, 2, TimeUnit.SECONDS);

	ClientOptions clientOptions = ClientOptions.builder() //
			.disconnectedBehavior(DisconnectedBehavior.REJECT_COMMANDS)//
			.build();

	RedisClient redis = RedisClient.create(clientResources, redisUri);
	redis.setOptions(clientOptions);

	return redis;
}
 

import com.lambdaworks.redis.RedisURI; //导入依赖的package包/类
private RedisClient createRedisClient(String host, int port) {
    RedisURI redisUri = new RedisURI(host, port, 2, TimeUnit.SECONDS);
    SocketOptions socketOptions = SocketOptions.builder()
            .connectTimeout(XPipeConsoleConstant.SOCKET_TIMEOUT, TimeUnit.SECONDS)
            .build();
    ClientOptions clientOptions = ClientOptions.builder() //
            .socketOptions(socketOptions)
            .disconnectedBehavior(ClientOptions.DisconnectedBehavior.REJECT_COMMANDS)//
            .build();

    DefaultClientResources clientResources = DefaultClientResources.builder()//
            .reconnectDelay(Delay.constant(1, TimeUnit.SECONDS))//
            .build();
    RedisClient redis = RedisClient.create(clientResources, redisUri);
    redis.setOptions(clientOptions);
    return redis;
}
 

import com.lambdaworks.redis.RedisURI; //导入依赖的package包/类
@Override
public void init(AbstractConfiguration config) {
    if (!config.getString("redis.type").equals("single")) {
        throw new IllegalStateException("RedisSyncSingleStorageImpl class can only be used with single redis setup, but redis.type value is " + config.getString("redis.type"));
    }

    List<String> address = parseRedisAddress(config.getString("redis.address"), 6379);
    int databaseNumber = config.getInt("redis.database", 0);
    String password = StringUtils.isNotEmpty(config.getString("redis.password")) ? config.getString("redis.password") + "@" : "";

    // lettuce
    RedisURI lettuceURI = RedisURI.create("redis://" + password + address.get(0) + "/" + databaseNumber);
    this.lettuce = RedisClient.create(lettuceURI);
    this.lettuceConn = this.lettuce.connect();

    // params
    initParams(config);
}
 

import com.lambdaworks.redis.RedisURI; //导入依赖的package包/类
@Override
public void init(AbstractConfiguration config) {
    if (!config.getString("redis.type").equals("sentinel")) {
        throw new IllegalStateException("RedisSyncSingleStorageImpl class can only be used with sentinel redis setup, but redis.type value is " + config.getString("redis.type"));
    }

    List<String> address = parseRedisAddress(config.getString("redis.address"), 26379);
    int databaseNumber = config.getInt("redis.database", 0);
    String password = StringUtils.isNotEmpty(config.getString("redis.password")) ? config.getString("redis.password") + "@" : "";
    String masterId = config.getString("redis.master");

    // lettuce
    RedisURI lettuceURI = RedisURI.create("redis-sentinel://" + password + String.join(",", address) + "/" + databaseNumber + "#" + masterId);
    this.lettuceSentinel = RedisClient.create(lettuceURI);
    this.lettuceSentinelConn = MasterSlave.connect(this.lettuceSentinel, new Utf8StringCodec(), lettuceURI);
    this.lettuceSentinelConn.setReadFrom(ReadFrom.valueOf(config.getString("redis.read")));

    // params
    initParams(config);
}
 

import com.lambdaworks.redis.RedisURI; //导入依赖的package包/类
@Override
public StatefulRedisConnection<K, V> apply(ConnectionKey key) {

    RedisURI.Builder builder = RedisURI.Builder
            .redis(key.host, key.port)
            .withSsl(initialRedisUri.isSsl())
            .withVerifyPeer(initialRedisUri.isVerifyPeer())
            .withStartTls(initialRedisUri.isStartTls());

    if (initialRedisUri.getPassword() != null && initialRedisUri.getPassword().length != 0) {
        builder.withPassword(initialRedisUri.getPassword());
    }

    if (initialRedisUri.getClientName() != null) {
        builder.withClientName(initialRedisUri.getClientName());
    }
    builder.withDatabase(initialRedisUri.getDatabase());

    StatefulRedisConnection<K, V> connection = redisClient.connect(redisCodec, builder.build());

    synchronized (stateLock) {
        connection.setAutoFlushCommands(autoFlushCommands);
    }

    return connection;
}
 

import com.lambdaworks.redis.RedisURI; //导入依赖的package包/类
/**
 * Open a new connection to a Redis Master-Slave server/servers using the supplied {@link RedisURI} and the supplied
 * {@link RedisCodec codec} to encode/decode keys.
 * <p>
 * This {@link MasterSlave} performs auto-discovery of nodes if the URI is a Redis Sentinel URI. Master/Slave URIs will be
 * treated as static topology and no additional hosts are discovered in such case. Redis Standalone Master/Slave will
 * discover the roles of the supplied {@link RedisURI URIs} and issue commands to the appropriate node.
 * </p>
 *
 * @param redisClient the Redis client
 * @param codec Use this codec to encode/decode keys and values, must not be {@literal null}
 * @param redisURIs the Redis server to connect to, must not be {@literal null}
 * @param <K> Key type
 * @param <V> Value type
 * @return A new connection
 */
public static <K, V> StatefulRedisMasterSlaveConnection<K, V> connect(RedisClient redisClient, RedisCodec<K, V> codec,
        Iterable<RedisURI> redisURIs) {

    LettuceAssert.notNull(redisClient, "RedisClient must not be null");
    LettuceAssert.notNull(codec, "RedisCodec must not be null");
    LettuceAssert.notNull(redisURIs, "RedisURIs must not be null");

    List<RedisURI> uriList = LettuceLists.newList(redisURIs);
    LettuceAssert.isTrue(!uriList.isEmpty(), "RedisURIs must not be empty");

    if (isSentinel(uriList.get(0))) {
        return connectSentinel(redisClient, codec, uriList.get(0));
    } else {
        return connectStaticMasterSlave(redisClient, codec, uriList);
    }
}
 

import com.lambdaworks.redis.RedisURI; //导入依赖的package包/类
RedisMasterSlaveNode(String host, int port, RedisURI seed, Role role) {

        RedisURI.Builder builder = RedisURI.Builder
                .redis(host, port)
                .withSsl(seed.isSsl())
                .withVerifyPeer(seed.isVerifyPeer())
                .withStartTls(seed.isStartTls());
        if (seed.getPassword() != null && seed.getPassword().length != 0) {
            builder.withPassword(new String(seed.getPassword()));
        }

        if (seed.getClientName() != null) {
            builder.withClientName(seed.getClientName());
        }

        builder.withDatabase(seed.getDatabase());

        this.redisURI = builder.build();
        this.role = role;
    }
 

import com.lambdaworks.redis.RedisURI; //导入依赖的package包/类
public Requests requestPing() {

        Requests requests = new Requests();

        for (Map.Entry<RedisURI, StatefulRedisConnection<String, String>> entry : connections.entrySet()) {

            CommandArgs<String, String> args = new CommandArgs<>(StringCodec.ASCII).add(CommandKeyword.NODES);
            Command<String, String, String> command = new Command<>(CommandType.PING, new StatusOutput<>(StringCodec.ASCII),
                    args);
            TimedAsyncCommand<String, String, String> timedCommand = new TimedAsyncCommand<>(command);

            entry.getValue().dispatch(timedCommand);
            requests.addRequest(entry.getKey(), timedCommand);
        }

        return requests;
    }
 

import com.lambdaworks.redis.RedisURI; //导入依赖的package包/类
@Test
public void shouldNotFailOnDuplicateSeedNodes() throws Exception {

    List<RedisURI> seed = Arrays.asList(RedisURI.create("127.0.0.1", 7380), RedisURI.create("127.0.0.1", 7381),
            RedisURI.create("127.0.0.1", 7381));

    when(nodeConnectionFactory.connectToNodeAsync(any(RedisCodec.class), eq(new InetSocketAddress("127.0.0.1", 7380))))
            .thenReturn(completedFuture((StatefulRedisConnection) connection1));

    when(nodeConnectionFactory.connectToNodeAsync(any(RedisCodec.class), eq(new InetSocketAddress("127.0.0.1", 7381))))
            .thenReturn(completedFuture((StatefulRedisConnection) connection2));

    sut.loadViews(seed, true);

    verify(nodeConnectionFactory).connectToNodeAsync(any(RedisCodec.class), eq(new InetSocketAddress("127.0.0.1", 7380)));
    verify(nodeConnectionFactory).connectToNodeAsync(any(RedisCodec.class), eq(new InetSocketAddress("127.0.0.1", 7381)));
}
 

import com.lambdaworks.redis.RedisURI; //导入依赖的package包/类
protected Requests createClusterNodesRequests(int duration, String nodes) {

        RedisURI redisURI = RedisURI.create("redis://localhost:" + duration);
        Connections connections = new Connections();
        connections.addConnection(redisURI, connection);

        Requests requests = connections.requestTopology();
        TimedAsyncCommand<String, String, String> command = requests.getRequest(redisURI);

        command.getOutput().set(ByteBuffer.wrap(nodes.getBytes()));
        command.complete();
        command.encodedAtNs = 0;
        command.completedAtNs = duration;

        return requests;
    }
 

import com.lambdaworks.redis.RedisURI; //导入依赖的package包/类
public Requests requestTopology() {

        Requests requests = new Requests();

        for (Map.Entry<RedisURI, StatefulRedisConnection<String, String>> entry : connections.entrySet()) {

            CommandArgs<String, String> args = new CommandArgs<>(StringCodec.UTF8).add(CommandKeyword.NODES);
            Command<String, String, String> command = new Command<>(CommandType.CLUSTER, new StatusOutput<>(StringCodec.UTF8),
                    args);
            TimedAsyncCommand<String, String, String> timedCommand = new TimedAsyncCommand<>(command);

            entry.getValue().dispatch(timedCommand);
            requests.addRequest(entry.getKey(), timedCommand);
        }

        return requests;
    }
 

import com.lambdaworks.redis.RedisURI; //导入依赖的package包/类
public Requests requestClients() {

        Requests requests = new Requests();

        for (Map.Entry<RedisURI, StatefulRedisConnection<String, String>> entry : connections.entrySet()) {

            CommandArgs<String, String> args = new CommandArgs<>(StringCodec.UTF8).add(CommandKeyword.LIST);
            Command<String, String, String> command = new Command<>(CommandType.CLIENT, new StatusOutput<>(StringCodec.UTF8),
                    args);
            TimedAsyncCommand<String, String, String> timedCommand = new TimedAsyncCommand<>(command);

            entry.getValue().dispatch(timedCommand);
            requests.addRequest(entry.getKey(), timedCommand);
        }

        return requests;
    }
 

import com.lambdaworks.redis.RedisURI; //导入依赖的package包/类
@Test
public void testSslWithPasswordMultipleHosts() {

    List<RedisURI> redisURIs = RedisClusterURIUtil.toRedisURIs(URI.create("redis+tls://[email protected]:6379,host2:6380"));

    assertThat(redisURIs).hasSize(2);

    RedisURI host1 = redisURIs.get(0);
    assertThat(host1.isSsl()).isTrue();
    assertThat(host1.isStartTls()).isTrue();
    assertThat(new String(host1.getPassword())).isEqualTo("password");
    assertThat(host1.getHost()).isEqualTo("host1");
    assertThat(host1.getPort()).isEqualTo(6379);

    RedisURI host2 = redisURIs.get(1);
    assertThat(host2.isSsl()).isTrue();
    assertThat(host2.isStartTls()).isTrue();
    assertThat(new String(host2.getPassword())).isEqualTo("password");
    assertThat(host2.getHost()).isEqualTo("host2");
    assertThat(host2.getPort()).isEqualTo(6380);
}
 

import com.lambdaworks.redis.RedisURI; //导入依赖的package包/类
@Test
public void shouldFailIfNoNodeConnects() throws Exception {

    List<RedisURI> seed = Arrays.asList(RedisURI.create("127.0.0.1", 7380), RedisURI.create("127.0.0.1", 7381));

    when(nodeConnectionFactory.connectToNodeAsync(any(RedisCodec.class), eq(new InetSocketAddress("127.0.0.1", 7380))))
            .thenReturn(completedWithException(new RedisException("connection failed")));
    when(nodeConnectionFactory.connectToNodeAsync(any(RedisCodec.class), eq(new InetSocketAddress("127.0.0.1", 7381))))
            .thenReturn(completedWithException(new RedisException("connection failed")));

    try {
        sut.loadViews(seed, true);
        fail("Missing RedisConnectionException");
    } catch (Exception e) {
        assertThat(e).hasNoCause().hasMessage("Unable to establish a connection to Redis Cluster");
        assertThat(e.getSuppressed()).hasSize(2);
    }

    verify(nodeConnectionFactory).connectToNodeAsync(any(RedisCodec.class), eq(new InetSocketAddress("127.0.0.1", 7380)));
    verify(nodeConnectionFactory).connectToNodeAsync(any(RedisCodec.class), eq(new InetSocketAddress("127.0.0.1", 7381)));
}
 

import com.lambdaworks.redis.RedisURI; //导入依赖的package包/类
@Test
public void isChangedShouldReturnFalse() throws Exception {

    RedisMasterSlaveNode master = new RedisMasterSlaveNode("host", 1234, RedisURI.create("host", 111),
            RedisInstance.Role.MASTER);
    RedisMasterSlaveNode slave = new RedisMasterSlaveNode("host", 234, RedisURI.create("host", 234),
            RedisInstance.Role.SLAVE);

    RedisMasterSlaveNode newmaster = new RedisMasterSlaveNode("host", 1234, RedisURI.create("host", 555),
            RedisInstance.Role.MASTER);
    RedisMasterSlaveNode newslave = new RedisMasterSlaveNode("host", 234, RedisURI.create("host", 666),
            RedisInstance.Role.SLAVE);

    assertThat(MasterSlaveUtils.isChanged(Arrays.asList(master, slave), Arrays.asList(newmaster, newslave))).isFalse();
    assertThat(MasterSlaveUtils.isChanged(Arrays.asList(slave, master), Arrays.asList(newmaster, newslave))).isFalse();

    assertThat(MasterSlaveUtils.isChanged(Arrays.asList(newmaster, newslave), Arrays.asList(master, slave))).isFalse();
    assertThat(MasterSlaveUtils.isChanged(Arrays.asList(newmaster, newslave), Arrays.asList(slave, master))).isFalse();
}
 

import com.lambdaworks.redis.RedisURI; //导入依赖的package包/类
@Test
public void splitUnhealthyNodeViewShouldDecideForHealthyNodes() throws Exception {

    Partitions partitions1 = createPartitions(node1, node2);
    Partitions partitions2 = createPartitions(node2, node3);
    Partitions partitions3 = createPartitions(node3, node4, node5);

    Map<RedisURI, Partitions> map = createMap(partitions1, partitions2, partitions3);

    node2.setFlags(Collections.singleton(RedisClusterNode.NodeFlag.FAIL));
    node3.setFlags(Collections.singleton(RedisClusterNode.NodeFlag.FAIL));
    node4.setFlags(Collections.singleton(RedisClusterNode.NodeFlag.FAIL));

    Partitions result = PartitionsConsensus.HEALTHY_MAJORITY.getPartitions(null, map);

    assertThat(Arrays.asList(partitions1, partitions3)).contains(result);
}
 

import com.lambdaworks.redis.RedisURI; //导入依赖的package包/类
@Test
public void bindDuringClose() {

    sut = new SentinelTopologyRefresh(redisClient, "mymaster", Arrays.asList(host1, host2));

    StatefulRedisPubSubConnection<String, String> connection2 = mock(StatefulRedisPubSubConnection.class);
    RedisPubSubAsyncCommands<String, String> async2 = mock(RedisPubSubAsyncCommands.class);
    when(connection2.async()).thenReturn(async2);

    when(redisClient.connectPubSub(any(StringCodec.class), eq(host2))).thenAnswer(invocation -> {

        sut.close();
        return connection2;
    });

    sut.bind(refreshRunnable);

    verify(redisClient).connectPubSub(any(), eq(host2));
    verify(connection).close();
    verify(connection2).close();

    Map<RedisURI, StatefulRedisPubSubConnection<String, String>> connections = (Map) ReflectionTestUtils.getField(sut,
            "pubSubConnections");

    assertThat(connections).isEmpty();
}