Lettuce客户端连接 Redis 服务器的三种方式你真的懂吗？RedisURI 揭秘URI 语法RedisURI 剖析同步方式异步方式Reactive 模式总结Lettuce系列文章

连接 Redis 的方式居然不止一种？你似乎从来没有关注过这个问题，为什么需要这么多种方式呢？也许你真的不懂，或者你知道有几种方式，但是到底是怎么实现的呢 ？

无论 Redis 服务器是，单机版本，哨兵模式还是集群模式，想要和 Redis 服务器建立连接，首选需要指定的 Redis服务器的地址等信息，Lettuce 对与Redis Standalone，Sentinel或Cluster的连接进行了规范连接详细信息。 统一的形式是RedisURI。

RedisURI 揭秘

URI,统一资源标识符（Uniform Resource Identifier，URI)是一个用于标识某一互联网资源名称的字符串。 该种标识允许用户对任何（包括本地和互联网的资源通过特定的协议进行交互操作。URI由包括确定语法和相关协议的方案所定义。顾名思义，RedisURI 是访问 Redis 资源的标识，Lettuce对 URI 进行了封装和抽象，以便能够快速，高效使用。可以在RedisURI中提供数据库，密码和超时信息。 可以通过以下三种方式创建RedisURI：

1. 使用一个 URI:RedisURI.create("redis://localhost/");
2. 使用 BuilderRedisURI.Builder.redis("localhost", 6379).auth("password").database(1).build();
3. 直接给 RedisURI实例进行赋值new RedisURI("localhost", 6379, 60, TimeUnit.SECONDS);

以上三种方式都可以创建要给 RedisURI 实例，三种方式都是为了给属性字段赋值。我们看看关键的属性，核心字段就三个，host，port 和 授权信息（password),RedisURI支持带有明文，SSL，TLS和unix域套接字连接的Redis Standalone，Redis Sentinel和Redis Cluster。以下 Redis 连接的语法来自于官网。

URI 语法

Redis Standalone

redis / [: password@] host [: port] [/ database][? [timeout=timeout[d|h|m|s|ms|us|ns]] [&database=database]]

Redis Standalone (SSL)

rediss / [: password@] host [: port] [/ database][? [timeout=timeout[d|h|m|s|ms|us|ns]] [&database=database]]

Redis Standalone (Unix Domain Sockets)

redis-socket / path [?[timeout=timeout[d|h|m|s|ms|us|ns]][&database=database]]

Redis Sentinel

redis-sentinel / [: password@] host1[: port1] [, host2[: port2]] [, hostN[: portN]] [/ database][?[timeout=timeout[d|h|m|s|ms|us|ns]] [&sentinelMasterId=sentinelMasterId] [&database=database]]

Schemes

• redis Redis Standalone
• rediss Redis Standalone SSL
• redis-socket Redis Standalone Unix Domain Socket
• redis-sentinel Redis Sentinel

Timeout units

• d Days
• h Hours
• m Minutes
• s Seconds
• ms Milliseconds
• us Microseconds
• ns Nanoseconds

RedisURI 剖析

RedisURI 类的主要职责：Redis URI。 包含Redis / Sentinel连接的连接详细信息。 我们可以在RedisURI中提供数据库，客户端名称，密码和超时。 您可以通过上述的三种方式创建RedisURI，类图如下。

顺便我们探一探 RedisURI 的源码，主要结合源码和文档进行分析，相关地址如下：

• 源码文件地址：https://github.com/paodingjiejiagou/lettuce-core/blob/master/src/main/java/io/lettuce/core/RedisURI.java
• JavaDoc地址 ： https://lettuce.io/core/release/api/index.html

我们先看看 RedisURI 实现的接口ConnectionPoint，连接点，节点信息的封装，我们可以通过安全或者不安全的方式到达连接点，之后再访问 Redis 服务器的数据库。

public interface ConnectionPoint {

    /**
     * Returns the host that should represent the hostname or IPv4/IPv6 literal.
     *
     * @return the hostname/IP address
     */
    String getHost();

    /**
     * Get the current port number.
     *
     * @return the port number
     */
    int getPort();

    /**
     * Get the socket path.
     *
     * @return path to a Unix Domain Socket
     */
    String getSocket();
}

这种多个参数和模式情况下，创建对象实例最适合 Builder 模式，RedisURI 也不例外，代码片段如下，大家可以复习下 Builder 模式，完整代码地址为：https://github.com/paodingjiejiagou/lettuce-core/blob/master/src/main/java/io/lettuce/core/RedisURI.java

public static class Builder {

    private String host;
    private String socket;
    private String sentinelMasterId;
    private int port = DEFAULT_REDIS_PORT;
    private int database;
    private String clientName;
    private String username;
    private char[] password;
    private char[] sentinelPassword;
    private boolean ssl = false;
    private boolean verifyPeer = true;
    private boolean startTls = false;
    private Duration timeout = DEFAULT_TIMEOUT_DURATION;
    private final List<RedisURI> sentinels = new ArrayList<>();

    private Builder() {
    }
    /**
     * Set Redis host. Creates a new builder.
     *
     * @param host the host name
     * @return new builder with Redis host/port.
     */
    public static Builder redis(String host) {
        return redis(host, DEFAULT_REDIS_PORT);
    }

    /**
     * Set Redis host and port. Creates a new builder
     *
     * @param host the host name
     * @param port the port
     * @return new builder with Redis host/port.
     */
    public static Builder redis(String host, int port) {

        LettuceAssert.notEmpty(host, "Host must not be empty");
        LettuceAssert.isTrue(isValidPort(port), () -> String.format("Port out of range: %s", port));

        Builder builder = RedisURI.builder();
        return builder.withHost(host).withPort(port);
    }

  
    /**
     * Configures authentication.
     *
     * @param password the password
     * @return the builder
     * @since 4.4
     */
    public Builder withPassword(char[] password) {

        LettuceAssert.notNull(password, "Password must not be null");

        this.password = Arrays.copyOf(password, password.length);
        return this;
    }

    /**
     * Configures a timeout.
     *
     * @param timeout must not be {@literal null} or negative.
     * @return the builder
     */
    public Builder withTimeout(Duration timeout) {

        LettuceAssert.notNull(timeout, "Timeout must not be null");
        LettuceAssert.notNull(!timeout.isNegative(), "Timeout must be greater or equal 0");

        this.timeout = timeout;
        return this;
    }
    /**
     * @return the RedisURI.
     */
    public RedisURI build() {

        if (sentinels.isEmpty() && LettuceStrings.isEmpty(host) && LettuceStrings.isEmpty(socket)) {
            throw new IllegalStateException(
                    "Cannot build a RedisURI. One of the following must be provided Host, Socket or Sentinel");
        }

        RedisURI redisURI = new RedisURI();
        redisURI.setHost(host);
        redisURI.setPort(port);

        if (username != null) {
            redisURI.setUsername(username);
        }

        if (password != null) {
            redisURI.setPassword(password);
        }

        redisURI.setDatabase(database);
        redisURI.setClientName(clientName);

        redisURI.setSentinelMasterId(sentinelMasterId);

        for (RedisURI sentinel : sentinels) {

            sentinel.setTimeout(timeout);
            redisURI.getSentinels().add(sentinel);
        }

        redisURI.setSocket(socket);
        redisURI.setSsl(ssl);
        redisURI.setStartTls(startTls);
        redisURI.setVerifyPeer(verifyPeer);
        redisURI.setTimeout(timeout);

        return redisURI;
    }
}

分析源码的目的除了欣赏，最主要的要学以致用，在后续的研发过程中，类似的功能，我们可以借鉴 RedisURI 的抽象思路和 Builder 设计模式，书写专业，优雅的代码。

完成了对连接信息的抽象和封装之后，我们该关注连接方式了。

同步方式

同步方式是我们最常使用的方式，也是最容易理解的方式，我们在文章Redis Java 客户端 Lettuce(生菜)食用指南代码示例就是同步方式。再温习下代码示例：

public class LettuceStarted {
    public static void main(String[] args) {
        RedisClient redisClient = RedisClient.create("redis://localhost:6379/0");
        StatefulRedisConnection<String, String> connection = redisClient.connect();
        RedisCommands<String, String> syncCommands = connection.sync();

        syncCommands.set("key", "Hello, Redis!");
        String value = syncCommands.get("key");
        System.out.println("value = " + value);
        connection.close();
        redisClient.shutdown();
    }
}

其中最关键的代码connection.sync()，指明了我们将使用同步方式连接的 Redis 服务器，进行相关数据库的操作。

异步方式

为什么我们需要异步方式，显然是因为当前调用的结果对我们来说，重要而不紧急。异步方法使我们可以利用更好的系统资源，而不是浪费线程等待网络或磁盘I / O。 可以充分利用线程来执行其他工作。 lettuce通过在Netty之上构建客户端来促进异步，netty是一个多线程，事件驱动的I / O框架。 所有通信都是异步处理的。 一旦基础设施能够同时处理命令，便可以方便地利用异步性。 然而将阻塞和同步的工作软件转换为并发处理系统要困难得多。一般来说，处理异步响应结果有三种方式。

• 主动轮询，查询处理结果
• 监听异步通知，进行处理
• 采用异步回调模式，异步处理完成回调指定的方法或者指令

我们看看 Lettuce 是如何处理的呢 ？异步API上的每个命令调用都会创建一个RedisFuture ，可以将其取消，等待和订阅（侦听器）。 CompleteableFuture <T>或RedisFuture <T>是指向结果的指针，该结果最初未知，因为其值的计算尚未完成。 RedisFuture <T>提供用于同步和和链式的操作。

庖丁先解剖下CompleteableFuture <T> ，这个是JDK 1.8引入异步处理类，虽然说 Future 接口在 Java 1.5 就引入了，但是由于其过于简单，不能满足异步场景的要求，很多开源组件实现了自己的异步处理类，如 Google 的 Guava库。CompleteableFuture的类图如下：

代码示例如下：

package com.lerith.anatomy.lettuce.started;

import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisFuture;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.async.RedisAsyncCommands;
import io.lettuce.core.api.sync.RedisCommands;

import java.util.concurrent.ExecutionException;

/**
 *
 */
public class AsynchronousConnected {
    public static void main(String[] args) {
        RedisClient redisClient = RedisClient.create("redis://localhost:6379/0");
        StatefulRedisConnection<String, String> connection = redisClient.connect();
        RedisAsyncCommands<String, String> asyncCommands = connection.async();

        RedisFuture<String> result = asyncCommands.set("key", "Hello, Redis!");
        RedisFuture<String> value = asyncCommands.get("key");
        try {
            System.out.println("value = " + value.get());
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
        connection.close();
        redisClient.shutdown();
    }
}

其中最关键的代码connection.async()，指明了我们将使用异步方式连接 Redis 服务器，进行相关数据库的操作。

Reactive 模式

异步和响应式方法使您可以利用更好的系统资源，而不是浪费线程等待网络或磁盘I / O。 可以充分利用线程来执行其他工作。

存在广泛的技术来促进这种编程风格，范围从非常有限且不易使用的java.util.concurrent.Future到完整的库和运行时（如Akka）。 Project Reactor具有非常丰富的运算符组合，可以组成异步工作流，它对其他框架没有进一步的依赖关系，并且支持非常成熟的Reactive Streams模型。Reactive Streams是为无阻塞背压的异步流处理提供标准的一项举措。 这包括针对运行时环境（JVM和JavaScript）以及网络协议的工作。它是一种超越编程语言的编程范式。

Lettuce 使用 Project Reactor （https://projectreactor.io/） 来实现响应式模式 ，这个模式是个比较大的话题，后面庖丁专门写一篇文章来阐述它。下图是 Project Reactor项目的主页。可以看出它的使命是创建高效的响应式系统。

代码示例如下：

package com.lerith.anatomy.lettuce.started;

import io.lettuce.core.RedisClient;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.reactive.RedisStringReactiveCommands;
import reactor.core.publisher.Mono;

public class ReactiveConnected {
    public static void main(String[] args) {
        RedisClient redisClient = RedisClient.create("redis://localhost:6379/0");
        StatefulRedisConnection<String, String> connection = redisClient.connect();
        RedisStringReactiveCommands<String, String> reactive = connection.reactive();
        Mono<String> set = reactive.set("key", "value");
        Mono<String> get = reactive.get("key");
        set.subscribe();
        System.out.println("get.block() = " + get.block());
    }
}

其中最关键的代码connection.reactive()，指明了我们将使用异步方式连接 Redis 服务器，进行相关数据库的操作。

总结

本文主要阐述了 Lettuce 连接 Redis 的三种方式，也是三种不同的编程风格，适用于不同的场景。示例代码：https://github.com/paodingjiejiagou/anatomy-lettuce 大家可以 clone 下来自己运行。三种连接方式主要定义在StatefulRedisConnection.java 接口中。代码片段如下：

/*
 ---
 */
package io.lettuce.core.api;

import io.lettuce.core.api.async.RedisAsyncCommands;
import io.lettuce.core.api.reactive.RedisReactiveCommands;
import io.lettuce.core.api.sync.RedisCommands;
import io.lettuce.core.protocol.ConnectionWatchdog;

/**
 * A thread-safe connection to a redis server. Multiple threads may share one {@link StatefulRedisConnection}.
 *
 * A {@link ConnectionWatchdog} monitors each connection and reconnects automatically until {@link #close} is called. All
 * pending commands will be (re)sent after successful reconnection.
 *
 * @param <K> Key type.
 * @param <V> Value type.
 * @author Mark Paluch
 * @since 4.0
 */
public interface StatefulRedisConnection<K, V> extends StatefulConnection<K, V> {

    /**
     *
     * @return true, if the connection is within a transaction.
     */
    boolean isMulti();

    /**
     * Returns the {@link RedisCommands} API for the current connection. Does not create a new connection.
     *
     * @return the synchronous API for the underlying connection.
     */
    RedisCommands<K, V> sync();

    /**
     * Returns the {@link RedisAsyncCommands} API for the current connection. Does not create a new connection.
     *
     * @return the asynchronous API for the underlying connection.
     */
    RedisAsyncCommands<K, V> async();

    /**
     * Returns the {@link RedisReactiveCommands} API for the current connection. Does not create a new connection.
     *
     * @return the reactive API for the underlying connection.
     */
    RedisReactiveCommands<K, V> reactive();
}

仔细观察StatefulRedisConnection接口定义的三种连接方式，发现最明显的就是命令操作方式各不相同。lettuce 通过对处理命令的抽象和封装来实现和应对不同的连接方式。下图是三种不同处理命令的类图，比较复杂，但 Redis 支持的操作命令，确实挺多，后续庖丁会继续解剖。

• RedisCommands 接口类图

• RedisAsyncCommands接口类图

• RedisReactiveCommands接口类图

Lettuce系列文章

1. 【Lettuce 架构解剖&源码精读】https://www.toutiao.com/i6817805765684757003/
2. 【Redis Java 客户端 Lettuce(生菜)食用指南】https://www.toutiao.com/i6818921648201138699/