剖析更高级的Redis客户端Lettuce

前言

Lettuce是一个Redis的Java驱动包，初识她的时候是使用RedisTemplate的时候遇到点问题Debug到底层的一些源码，发现spring-data-redis的驱动包在某个版本之后替换为Lettuce。Lettuce翻译为生菜，没错，就是吃的那种生菜，所以它的Logo长这样：

既然能被Spring生态所认可，Lettuce想必有过人之处，于是笔者花时间阅读她的官方文档，整理测试示例，写下这篇文章。编写本文时所使用的版本为Lettuce 5.1.8.RELEASE，SpringBoot 2.1.8.RELEASE，JDK [8,11]。

Lettuce简介

Lettuce是一个高性能基于Java编写的Redis驱动框架，底层集成了Project Reactor提供天然的反应式编程，通信框架集成了Netty使用了非阻塞IO，5.x版本之后融合了JDK1.8的异步编程特性，在保证高性能的同时提供了十分丰富易用的API，5.1版本的新特性如下：

• 支持Redis的新增命令ZPOPMIN, ZPOPMAX, BZPOPMIN, BZPOPMAX。

• 支持通过Brave模块跟踪Redis命令执行。

• 支持Redis Streams。

• 支持异步的主从连接。

• 支持异步连接池。

• 新增命令最多执行一次模式（禁止自动重连）。

• 全局命令超时设置（对异步和反应式命令也有效）。

• ......等等

注意一点：Redis的版本至少需要2.6，当然越高越好，API的兼容性比较强大。

只需要引入单个依赖就可以开始愉快地使用Lettuce：

• Maven

<dependency><groupId>io.lettuce</groupId><artifactId>lettuce-core</artifactId><version>5.1.8.RELEASE</version>
</dependency>

• Gradle

dependencies {compile 'io.lettuce:lettuce-core:5.1.8.RELEASE'
}

连接Redis

单机、哨兵、集群模式下连接Redis需要一个统一的标准去表示连接的细节信息，在Lettuce中这个统一的标准是RedisURI。可以通过三种方式构造一个RedisURI实例：

• 定制的字符串URI语法：

RedisURI uri = RedisURI.create("redis://localhost/");

• 使用建造器（RedisURI.Builder）：

RedisURI uri = RedisURI.builder().withHost("localhost").withPort(6379).build();

• 直接通过构造函数实例化：

RedisURI uri = new RedisURI("localhost", 6379, 60, TimeUnit.SECONDS);

定制的连接URI语法

• 单机（前缀为redis://）

格式：redis://[password@]host[:port][/databaseNumber][?[timeout=timeout[d|h|m|s|ms|us|ns]]
完整：redis://mypassword@127.0.0.1:6379/0?timeout=10s
简单：redis://localhost

• 单机并且使用SSL（前缀为rediss://） <== 注意后面多了个s

格式：rediss://[password@]host[:port][/databaseNumber][?[timeout=timeout[d|h|m|s|ms|us|ns]]
完整：rediss://mypassword@127.0.0.1:6379/0?timeout=10s
简单：rediss://localhost

• 单机Unix Domain Sockets模式（前缀为redis-socket://）

格式：redis-socket://path[?[timeout=timeout[d|h|m|s|ms|us|ns]][&_database=database_]]
完整：redis-socket:///tmp/redis?timeout=10s&_database=0

• 哨兵（前缀为redis-sentinel://）

格式：redis-sentinel://[password@]host[:port][,host2[:port2]][/databaseNumber][?[timeout=timeout[d|h|m|s|ms|us|ns]]#sentinelMasterId
完整：redis-sentinel://mypassword@127.0.0.1:6379,127.0.0.1:6380/0?timeout=10s#mymaster

超时时间单位：

• d 天

• h 小时

• m 分钟

• s 秒钟

• ms 毫秒

• us 微秒

• ns 纳秒

个人建议使用RedisURI提供的建造器，毕竟定制的URI虽然简洁，但是比较容易出现人为错误。鉴于笔者没有SSL和Unix Domain Socket的使用场景，下面不对这两种连接方式进行列举。

基本使用

Lettuce使用的时候依赖于四个主要组件：

• RedisURI：连接信息。

• RedisClient：Redis客户端，特殊地，集群连接有一个定制的RedisClusterClient。

• Connection：Redis连接，主要是StatefulConnection或者StatefulRedisConnection的子类，连接的类型主要由连接的具体方式（单机、哨兵、集群、订阅发布等等）选定，比较重要。

• RedisCommands：Redis命令API接口，基本上覆盖了Redis发行版本的所有命令，提供了同步（sync）、异步（async）、反应式（reative）的调用方式，对于使用者而言，会经常跟RedisCommands系列接口打交道。

一个基本使用例子如下：

@Test
public void testSetGet() throws Exception {RedisURI redisUri = RedisURI.builder()                    // <1> 创建单机连接的连接信息.withHost("localhost").withPort(6379).withTimeout(Duration.of(10, ChronoUnit.SECONDS)).build();RedisClient redisClient = RedisClient.create(redisUri);   // <2> 创建客户端StatefulRedisConnection<String, String> connection = redisClient.connect();     // <3> 创建线程安全的连接RedisCommands<String, String> redisCommands = connection.sync();                // <4> 创建同步命令SetArgs setArgs = SetArgs.Builder.nx().ex(5);String result = redisCommands.set("name", "throwable", setArgs);Assertions.assertThat(result).isEqualToIgnoringCase("OK");result = redisCommands.get("name");Assertions.assertThat(result).isEqualTo("throwable");// ... 其他操作connection.close();   // <5> 关闭连接redisClient.shutdown();  // <6> 关闭客户端
}

注意：

• <5>：关闭连接一般在应用程序停止之前操作，一个应用程序中的一个Redis驱动实例不需要太多的连接（一般情况下只需要一个连接实例就可以，如果有多个连接的需要可以考虑使用连接池，其实Redis目前处理命令的模块是单线程，在客户端多个连接多线程调用理论上没有效果）。

• <6>：关闭客户端一般应用程序停止之前操作，如果条件允许的话，基于后开先闭原则，客户端关闭应该在连接关闭之后操作。

API

Lettuce主要提供三种API：

• 同步（sync）：RedisCommands。

• 异步（async）：RedisAsyncCommands。

• 反应式（reactive）：RedisReactiveCommands。

先准备好一个单机Redis连接备用：

private static StatefulRedisConnection<String, String> CONNECTION;
private static RedisClient CLIENT;@BeforeClass
public static void beforeClass() {RedisURI redisUri = RedisURI.builder().withHost("localhost").withPort(6379).withTimeout(Duration.of(10, ChronoUnit.SECONDS)).build();CLIENT = RedisClient.create(redisUri);CONNECTION = CLIENT.connect();
}@AfterClass
public static void afterClass() throws Exception {CONNECTION.close();CLIENT.shutdown();
}

Redis命令API的具体实现可以直接从StatefulRedisConnection实例获取，见其接口定义：

public interface StatefulRedisConnection<K, V> extends StatefulConnection<K, V> {boolean isMulti();RedisCommands<K, V> sync();RedisAsyncCommands<K, V> async();RedisReactiveCommands<K, V> reactive();
}

值得注意的是，在不指定编码解码器RedisCodec的前提下，RedisClient创建的StatefulRedisConnection实例一般是泛型实例StatefulRedisConnection<String,String>，也就是所有命令API的KEY和VALUE都是String类型，这种使用方式能满足大部分的使用场景。当然，必要的时候可以定制编码解码器RedisCodec<K,V>。

同步API

先构建RedisCommands实例：

private static RedisCommands<String, String> COMMAND;@BeforeClass
public static void beforeClass() {COMMAND = CONNECTION.sync();
}

基本使用：

@Test
public void testSyncPing() throws Exception {String pong = COMMAND.ping();Assertions.assertThat(pong).isEqualToIgnoringCase("PONG");
}@Test
public void testSyncSetAndGet() throws Exception {SetArgs setArgs = SetArgs.Builder.nx().ex(5);COMMAND.set("name", "throwable", setArgs);String value = COMMAND.get("name");log.info("Get value: {}", value);
}// Get value: throwable

同步API在所有命令调用之后会立即返回结果。如果熟悉Jedis的话，RedisCommands的用法其实和它相差不大。

异步API

先构建RedisAsyncCommands实例：

private static RedisAsyncCommands<String, String> ASYNC_COMMAND;@BeforeClass
public static void beforeClass() {ASYNC_COMMAND = CONNECTION.async();
}

基本使用：

@Test
public void testAsyncPing() throws Exception {RedisFuture<String> redisFuture = ASYNC_COMMAND.ping();log.info("Ping result:{}", redisFuture.get());
}
// Ping result:PONG

RedisAsyncCommands所有方法执行返回结果都是RedisFuture实例，而RedisFuture接口的定义如下：

public interface RedisFuture<V> extends CompletionStage<V>, Future<V> {String getError();boolean await(long timeout, TimeUnit unit) throws InterruptedException;
}

也就是，RedisFuture可以无缝使用Future或者JDK1.8中引入的CompletableFuture提供的方法。举个例子：

@Test
public void testAsyncSetAndGet1() throws Exception {SetArgs setArgs = SetArgs.Builder.nx().ex(5);RedisFuture<String> future = ASYNC_COMMAND.set("name", "throwable", setArgs);// CompletableFuture#thenAccept()future.thenAccept(value -> log.info("Set命令返回:{}", value));// Future#get()future.get();
}
// Set命令返回:OK@Test
public void testAsyncSetAndGet2() throws Exception {SetArgs setArgs = SetArgs.Builder.nx().ex(5);CompletableFuture<Void> result =(CompletableFuture<Void>) ASYNC_COMMAND.set("name", "throwable", setArgs).thenAcceptBoth(ASYNC_COMMAND.get("name"),(s, g) -> {log.info("Set命令返回:{}", s);log.info("Get命令返回:{}", g);});result.get();
}
// Set命令返回:OK
// Get命令返回:throwable

如果能熟练使用CompletableFuture和函数式编程技巧，可以组合多个RedisFuture完成一些列复杂的操作。

反应式API

Lettuce引入的反应式编程框架是Project Reactor，如果没有反应式编程经验可以先自行了解一下Project Reactor。

构建RedisReactiveCommands实例：

private static RedisReactiveCommands<String, String> REACTIVE_COMMAND;@BeforeClass
public static void beforeClass() {REACTIVE_COMMAND = CONNECTION.reactive();
}

根据Project Reactor，RedisReactiveCommands的方法如果返回的结果只包含0或1个元素，那么返回值类型是Mono，如果返回的结果包含0到N（N大于0）个元素，那么返回值是Flux。举个例子：

@Test
public void testReactivePing() throws Exception {Mono<String> ping = REACTIVE_COMMAND.ping();ping.subscribe(v -> log.info("Ping result:{}", v));Thread.sleep(1000);
}
// Ping result:PONG@Test
public void testReactiveSetAndGet() throws Exception {SetArgs setArgs = SetArgs.Builder.nx().ex(5);REACTIVE_COMMAND.set("name", "throwable", setArgs).block();REACTIVE_COMMAND.get("name").subscribe(value -> log.info("Get命令返回:{}", value));Thread.sleep(1000);
}
// Get命令返回:throwable@Test
public void testReactiveSet() throws Exception {REACTIVE_COMMAND.sadd("food", "bread", "meat", "fish").block();Flux<String> flux = REACTIVE_COMMAND.smembers("food");flux.subscribe(log::info);REACTIVE_COMMAND.srem("food", "bread", "meat", "fish").block();Thread.sleep(1000);
}
// meat
// bread
// fish

举个更加复杂的例子，包含了事务、函数转换等：

@Test
public void testReactiveFunctional() throws Exception {REACTIVE_COMMAND.multi().doOnSuccess(r -> {REACTIVE_COMMAND.set("counter", "1").doOnNext(log::info).subscribe();REACTIVE_COMMAND.incr("counter").doOnNext(c -> log.info(String.valueOf(c))).subscribe();}).flatMap(s -> REACTIVE_COMMAND.exec()).doOnNext(transactionResult -> log.info("Discarded:{}", transactionResult.wasDiscarded())).subscribe();Thread.sleep(1000);
}
// OK
// 2
// Discarded:false

这个方法开启一个事务，先把counter设置为1，再将counter自增1。

发布和订阅

非集群模式下的发布订阅依赖于定制的连接StatefulRedisPubSubConnection，集群模式下的发布订阅依赖于定制的连接StatefulRedisClusterPubSubConnection，两者分别来源于RedisClient#connectPubSub()系列方法和RedisClusterClient#connectPubSub()：

• 非集群模式：

// 可能是单机、普通主从、哨兵等非集群模式的客户端
RedisClient client = ...
StatefulRedisPubSubConnection<String, String> connection = client.connectPubSub();
connection.addListener(new RedisPubSubListener<String, String>() { ... });// 同步命令
RedisPubSubCommands<String, String> sync = connection.sync();
sync.subscribe("channel");// 异步命令
RedisPubSubAsyncCommands<String, String> async = connection.async();
RedisFuture<Void> future = async.subscribe("channel");// 反应式命令
RedisPubSubReactiveCommands<String, String> reactive = connection.reactive();
reactive.subscribe("channel").subscribe();reactive.observeChannels().doOnNext(patternMessage -> {...}).subscribe()

• 集群模式：

// 使用方式其实和非集群模式基本一致
RedisClusterClient clusterClient = ...
StatefulRedisClusterPubSubConnection<String, String> connection = clusterClient.connectPubSub();
connection.addListener(new RedisPubSubListener<String, String>() { ... });
RedisPubSubCommands<String, String> sync = connection.sync();
sync.subscribe("channel");
// ...

这里用单机同步命令的模式举一个Redis键空间通知（Redis Keyspace Notifications）的例子：

@Test
public void testSyncKeyspaceNotification() throws Exception {RedisURI redisUri = RedisURI.builder().withHost("localhost").withPort(6379)// 注意这里只能是0号库.withDatabase(0).withTimeout(Duration.of(10, ChronoUnit.SECONDS)).build();RedisClient redisClient = RedisClient.create(redisUri);StatefulRedisConnection<String, String> redisConnection = redisClient.connect();RedisCommands<String, String> redisCommands = redisConnection.sync();// 只接收键过期的事件redisCommands.configSet("notify-keyspace-events", "Ex");StatefulRedisPubSubConnection<String, String> connection = redisClient.connectPubSub();connection.addListener(new RedisPubSubAdapter<>() {@Overridepublic void psubscribed(String pattern, long count) {log.info("pattern:{},count:{}", pattern, count);}@Overridepublic void message(String pattern, String channel, String message) {log.info("pattern:{},channel:{},message:{}", pattern, channel, message);}});RedisPubSubCommands<String, String> commands = connection.sync();commands.psubscribe("__keyevent@0__:expired");redisCommands.setex("name", 2, "throwable");Thread.sleep(10000);redisConnection.close();connection.close();redisClient.shutdown();
}
// pattern:__keyevent@0__:expired,count:1
// pattern:__keyevent@0__:expired,channel:__keyevent@0__:expired,message:name

实际上，在实现RedisPubSubListener的时候可以单独抽离，尽量不要设计成匿名内部类的形式。

事务和批量命令执行

事务相关的命令就是WATCH、UNWATCH、EXEC、MULTI和DISCARD，在RedisCommands系列接口中有对应的方法。举个例子：

// 同步模式
@Test
public void testSyncMulti() throws Exception {COMMAND.multi();COMMAND.setex("name-1", 2, "throwable");COMMAND.setex("name-2", 2, "doge");TransactionResult result = COMMAND.exec();int index = 0;for (Object r : result) {log.info("Result-{}:{}", index, r);index++;}
}
// Result-0:OK
// Result-1:OK

Redis的Pipeline也就是管道机制可以理解为把多个命令打包在一次请求发送到Redis服务端，然后Redis服务端把所有的响应结果打包好一次性返回，从而节省不必要的网络资源（最主要是减少网络请求次数）。Redis对于Pipeline机制如何实现并没有明确的规定，也没有提供特殊的命令支持Pipeline机制。Jedis中底层采用BIO（阻塞IO）通讯，所以它的做法是客户端缓存将要发送的命令，最后需要触发然后同步发送一个巨大的命令列表包，再接收和解析一个巨大的响应列表包。Pipeline在Lettuce中对使用者是透明的，由于底层的通讯框架是Netty，所以网络通讯层面的优化Lettuce不需要过多干预，换言之可以这样理解：Netty帮Lettuce从底层实现了Redis的Pipeline机制。但是，Lettuce的异步API也提供了手动Flush的方法：

@Test
public void testAsyncManualFlush() {// 取消自动flushASYNC_COMMAND.setAutoFlushCommands(false);List<RedisFuture<?>> redisFutures = Lists.newArrayList();int count = 5000;for (int i = 0; i < count; i++) {String key = "key-" + (i + 1);String value = "value-" + (i + 1);redisFutures.add(ASYNC_COMMAND.set(key, value));redisFutures.add(ASYNC_COMMAND.expire(key, 2));}long start = System.currentTimeMillis();ASYNC_COMMAND.flushCommands();boolean result = LettuceFutures.awaitAll(10, TimeUnit.SECONDS, redisFutures.toArray(new RedisFuture[0]));Assertions.assertThat(result).isTrue();log.info("Lettuce cost:{} ms", System.currentTimeMillis() - start);
}
// Lettuce cost:1302 ms

上面只是从文档看到的一些理论术语，但是现实是骨感的，对比了下Jedis的Pipeline提供的方法，发现了Jedis的Pipeline执行耗时比较低：

@Test
public void testJedisPipeline() throws Exception {Jedis jedis = new Jedis();Pipeline pipeline = jedis.pipelined();int count = 5000;for (int i = 0; i < count; i++) {String key = "key-" + (i + 1);String value = "value-" + (i + 1);pipeline.set(key, value);pipeline.expire(key, 2);}long start = System.currentTimeMillis();pipeline.syncAndReturnAll();log.info("Jedis cost:{} ms", System.currentTimeMillis()  - start);
}
// Jedis cost:9 ms

个人猜测Lettuce可能底层并非合并所有命令一次发送（甚至可能是单条发送），具体可能需要抓包才能定位。依此来看，如果真的有大量执行Redis命令的场景，不妨可以使用Jedis的Pipeline。

注意：由上面的测试推断RedisTemplate的executePipelined()方法是假的Pipeline执行方法，使用RedisTemplate的时候请务必注意这一点。

Lua脚本执行

Lettuce中执行Redis的Lua命令的同步接口如下：

public interface RedisScriptingCommands<K, V> {<T> T eval(String var1, ScriptOutputType var2, K... var3);<T> T eval(String var1, ScriptOutputType var2, K[] var3, V... var4);<T> T evalsha(String var1, ScriptOutputType var2, K... var3);<T> T evalsha(String var1, ScriptOutputType var2, K[] var3, V... var4);List<Boolean> scriptExists(String... var1);String scriptFlush();String scriptKill();String scriptLoad(V var1);String digest(V var1);
}

异步和反应式的接口方法定义差不多，不同的地方就是返回值类型，一般我们常用的是eval()、evalsha()和scriptLoad()方法。举个简单的例子：

private static RedisCommands<String, String> COMMANDS;
private static String RAW_LUA = "local key = KEYS[1]\n" +"local value = ARGV[1]\n" +"local timeout = ARGV[2]\n" +"redis.call('SETEX', key, tonumber(timeout), value)\n" +"local result = redis.call('GET', key)\n" +"return result;";
private static AtomicReference<String> LUA_SHA = new AtomicReference<>();@Test
public void testLua() throws Exception {LUA_SHA.compareAndSet(null, COMMANDS.scriptLoad(RAW_LUA));String[] keys = new String[]{"name"};String[] args = new String[]{"throwable", "5000"};String result = COMMANDS.evalsha(LUA_SHA.get(), ScriptOutputType.VALUE, keys, args);log.info("Get value:{}", result);
}
// Get value:throwable

高可用和分片

为了Redis的高可用，一般会采用普通主从（Master/Replica，这里笔者称为普通主从模式，也就是仅仅做了主从复制，故障需要手动切换）、哨兵和集群。普通主从模式可以独立运行，也可以配合哨兵运行，只是哨兵提供自动故障转移和主节点提升功能。普通主从和哨兵都可以使用MasterSlave，通过入参包括RedisClient、编码解码器以及一个或者多个RedisURI获取对应的Connection实例。

这里注意一点，MasterSlave中提供的方法如果只要求传入一个RedisURI实例，那么Lettuce会进行拓扑发现机制，自动获取Redis主从节点信息；如果要求传入一个RedisURI集合，那么对于普通主从模式来说所有节点信息是静态的，不会进行发现和更新。

拓扑发现的规则如下：

• 对于普通主从（Master/Replica）模式，不需要感知RedisURI指向从节点还是主节点，只会进行一次性的拓扑查找所有节点信息，此后节点信息会保存在静态缓存中，不会更新。

• 对于哨兵模式，会订阅所有哨兵实例并侦听订阅/发布消息以触发拓扑刷新机制，更新缓存的节点信息，也就是哨兵天然就是动态发现节点信息，不支持静态配置。

拓扑发现机制的提供API为TopologyProvider，需要了解其原理的可以参考具体的实现。

对于集群（Cluster）模式，Lettuce提供了一套独立的API。

另外，如果Lettuce连接面向的是非单个Redis节点，连接实例提供了数据读取节点偏好（ReadFrom）设置，可选值有：

• MASTER：只从Master节点中读取。

• MASTER_PREFERRED：优先从Master节点中读取。

• SLAVE_PREFERRED：优先从Slavor节点中读取。

• SLAVE：只从Slavor节点中读取。

• NEAREST：使用最近一次连接的Redis实例读取。

普通主从模式

假设现在有三个Redis服务形成树状主从关系如下：

• 节点一：localhost:6379，角色为Master。

• 节点二：localhost:6380，角色为Slavor，节点一的从节点。

• 节点三：localhost:6381，角色为Slavor，节点二的从节点。

首次动态节点发现主从模式的节点信息需要如下构建连接：

@Test
public void testDynamicReplica() throws Exception {// 这里只需要配置一个节点的连接信息，不一定需要是主节点的信息，从节点也可以RedisURI uri = RedisURI.builder().withHost("localhost").withPort(6379).build();RedisClient redisClient = RedisClient.create(uri);StatefulRedisMasterSlaveConnection<String, String> connection = MasterSlave.connect(redisClient, new Utf8StringCodec(), uri);// 只从从节点读取数据connection.setReadFrom(ReadFrom.SLAVE);// 执行其他Redis命令connection.close();redisClient.shutdown();
}

如果需要指定静态的Redis主从节点连接属性，那么可以这样构建连接：

@Test
public void testStaticReplica() throws Exception {List<RedisURI> uris = new ArrayList<>();RedisURI uri1 = RedisURI.builder().withHost("localhost").withPort(6379).build();RedisURI uri2 = RedisURI.builder().withHost("localhost").withPort(6380).build();RedisURI uri3 = RedisURI.builder().withHost("localhost").withPort(6381).build();uris.add(uri1);uris.add(uri2);uris.add(uri3);RedisClient redisClient = RedisClient.create();StatefulRedisMasterSlaveConnection<String, String> connection = MasterSlave.connect(redisClient,new Utf8StringCodec(), uris);// 只从主节点读取数据connection.setReadFrom(ReadFrom.MASTER);// 执行其他Redis命令connection.close();redisClient.shutdown();
}

哨兵模式

由于Lettuce自身提供了哨兵的拓扑发现机制，所以只需要随便配置一个哨兵节点的RedisURI实例即可：

@Test
public void testDynamicSentinel() throws Exception {RedisURI redisUri = RedisURI.builder().withPassword("你的密码").withSentinel("localhost", 26379).withSentinelMasterId("哨兵Master的ID").build();RedisClient redisClient = RedisClient.create();StatefulRedisMasterSlaveConnection<String, String> connection = MasterSlave.connect(redisClient, new Utf8StringCodec(), redisUri);// 只允许从从节点读取数据connection.setReadFrom(ReadFrom.SLAVE);RedisCommands<String, String> command = connection.sync();SetArgs setArgs = SetArgs.Builder.nx().ex(5);command.set("name", "throwable", setArgs);String value = command.get("name");log.info("Get value:{}", value);
}
// Get value:throwable

集群模式

鉴于笔者对Redis集群模式并不熟悉，Cluster模式下的API使用本身就有比较多的限制，所以这里只简单介绍一下怎么用。先说几个特性：

下面的API提供跨槽位（Slot）调用的功能：

• RedisAdvancedClusterCommands。

• RedisAdvancedClusterAsyncCommands。

• RedisAdvancedClusterReactiveCommands。

静态节点选择功能：

• masters：选择所有主节点执行命令。

• slaves：选择所有从节点执行命令，其实就是只读模式。

• all nodes：命令可以在所有节点执行。

集群拓扑视图动态更新功能：

• 手动更新，主动调用RedisClusterClient#reloadPartitions()。

• 后台定时更新。

• 自适应更新，基于连接断开和MOVED/ASK命令重定向自动更新。

Redis集群搭建详细过程可以参考官方文档，假设已经搭建好集群如下（192.168.56.200是笔者的虚拟机Host）：

• 192.168.56.200:7001 => 主节点，槽位0-5460。

• 192.168.56.200:7002 => 主节点，槽位5461-10922。

• 192.168.56.200:7003 => 主节点，槽位10923-16383。

• 192.168.56.200:7004 => 7001的从节点。

• 192.168.56.200:7005 => 7002的从节点。

• 192.168.56.200:7006 => 7003的从节点。

简单的集群连接和使用方式如下：

@Test
public void testSyncCluster(){RedisURI uri = RedisURI.builder().withHost("192.168.56.200").build();RedisClusterClient redisClusterClient = RedisClusterClient.create(uri);StatefulRedisClusterConnection<String, String> connection = redisClusterClient.connect();RedisAdvancedClusterCommands<String, String> commands = connection.sync();commands.setex("name",10, "throwable");String value = commands.get("name");log.info("Get value:{}", value);
}
// Get value:throwable

节点选择：

@Test
public void testSyncNodeSelection() {RedisURI uri = RedisURI.builder().withHost("192.168.56.200").withPort(7001).build();RedisClusterClient redisClusterClient = RedisClusterClient.create(uri);StatefulRedisClusterConnection<String, String> connection = redisClusterClient.connect();RedisAdvancedClusterCommands<String, String> commands = connection.sync();
//  commands.all();  // 所有节点
//  commands.masters();  // 主节点// 从节点只读NodeSelection<String, String> replicas = commands.slaves();NodeSelectionCommands<String, String> nodeSelectionCommands = replicas.commands();// 这里只是演示,一般应该禁用keys *命令Executions<List<String>> keys = nodeSelectionCommands.keys("*");keys.forEach(key -> log.info("key: {}", key));connection.close();redisClusterClient.shutdown();
}

定时更新集群拓扑视图（每隔十分钟更新一次，这个时间自行考量，不能太频繁）：

@Test
public void testPeriodicClusterTopology() throws Exception {RedisURI uri = RedisURI.builder().withHost("192.168.56.200").withPort(7001).build();RedisClusterClient redisClusterClient = RedisClusterClient.create(uri);ClusterTopologyRefreshOptions options = ClusterTopologyRefreshOptions.builder().enablePeriodicRefresh(Duration.of(10, ChronoUnit.MINUTES)).build();redisClusterClient.setOptions(ClusterClientOptions.builder().topologyRefreshOptions(options).build());StatefulRedisClusterConnection<String, String> connection = redisClusterClient.connect();RedisAdvancedClusterCommands<String, String> commands = connection.sync();commands.setex("name", 10, "throwable");String value = commands.get("name");log.info("Get value:{}", value);Thread.sleep(Integer.MAX_VALUE);connection.close();redisClusterClient.shutdown();
}

自适应更新集群拓扑视图：

@Test
public void testAdaptiveClusterTopology() throws Exception {RedisURI uri = RedisURI.builder().withHost("192.168.56.200").withPort(7001).build();RedisClusterClient redisClusterClient = RedisClusterClient.create(uri);ClusterTopologyRefreshOptions options = ClusterTopologyRefreshOptions.builder().enableAdaptiveRefreshTrigger(ClusterTopologyRefreshOptions.RefreshTrigger.MOVED_REDIRECT,ClusterTopologyRefreshOptions.RefreshTrigger.PERSISTENT_RECONNECTS).adaptiveRefreshTriggersTimeout(Duration.of(30, ChronoUnit.SECONDS)).build();redisClusterClient.setOptions(ClusterClientOptions.builder().topologyRefreshOptions(options).build());StatefulRedisClusterConnection<String, String> connection = redisClusterClient.connect();RedisAdvancedClusterCommands<String, String> commands = connection.sync();commands.setex("name", 10, "throwable");String value = commands.get("name");log.info("Get value:{}", value);Thread.sleep(Integer.MAX_VALUE);connection.close();redisClusterClient.shutdown();
}

动态命令和自定义命令

自定义命令是Redis命令有限集，不过可以更细粒度指定KEY、ARGV、命令类型、编码解码器和返回值类型，依赖于dispatch()方法：

// 自定义实现PING方法
@Test
public void testCustomPing() throws Exception {RedisURI redisUri = RedisURI.builder().withHost("localhost").withPort(6379).withTimeout(Duration.of(10, ChronoUnit.SECONDS)).build();RedisClient redisClient = RedisClient.create(redisUri);StatefulRedisConnection<String, String> connect = redisClient.connect();RedisCommands<String, String> sync = connect.sync();RedisCodec<String, String> codec = StringCodec.UTF8;String result = sync.dispatch(CommandType.PING, new StatusOutput<>(codec));log.info("PING:{}", result);connect.close();redisClient.shutdown();
}
// PING:PONG// 自定义实现Set方法
@Test
public void testCustomSet() throws Exception {RedisURI redisUri = RedisURI.builder().withHost("localhost").withPort(6379).withTimeout(Duration.of(10, ChronoUnit.SECONDS)).build();RedisClient redisClient = RedisClient.create(redisUri);StatefulRedisConnection<String, String> connect = redisClient.connect();RedisCommands<String, String> sync = connect.sync();RedisCodec<String, String> codec = StringCodec.UTF8;sync.dispatch(CommandType.SETEX, new StatusOutput<>(codec),new CommandArgs<>(codec).addKey("name").add(5).addValue("throwable"));String result = sync.get("name");log.info("Get value:{}", result);connect.close();redisClient.shutdown();
}
// Get value:throwable

动态命令是基于Redis命令有限集，并且通过注解和动态代理完成一些复杂命令组合的实现。主要注解在io.lettuce.core.dynamic.annotation包路径下。简单举个例子：

public interface CustomCommand extends Commands {// SET [key] [value]@Command("SET ?0 ?1")String setKey(String key, String value);// SET [key] [value]@Command("SET :key :value")String setKeyNamed(@Param("key") String key, @Param("value") String value);// MGET [key1] [key2]@Command("MGET ?0 ?1")List<String> mGet(String key1, String key2);/*** 方法名作为命令*/@CommandNaming(strategy = CommandNaming.Strategy.METHOD_NAME)String mSet(String key1, String value1, String key2, String value2);
}@Test
public void testCustomDynamicSet() throws Exception {RedisURI redisUri = RedisURI.builder().withHost("localhost").withPort(6379).withTimeout(Duration.of(10, ChronoUnit.SECONDS)).build();RedisClient redisClient = RedisClient.create(redisUri);StatefulRedisConnection<String, String> connect = redisClient.connect();RedisCommandFactory commandFactory = new RedisCommandFactory(connect);CustomCommand commands = commandFactory.getCommands(CustomCommand.class);commands.setKey("name", "throwable");commands.setKeyNamed("throwable", "doge");log.info("MGET ===> " + commands.mGet("name", "throwable"));commands.mSet("key1", "value1","key2", "value2");log.info("MGET ===> " + commands.mGet("key1", "key2"));connect.close();redisClient.shutdown();
}
// MGET ===> [throwable, doge]
// MGET ===> [value1, value2]

高阶特性

Lettuce有很多高阶使用特性，这里只列举个人认为常用的两点：

• 配置客户端资源。

• 使用连接池。

更多其他特性可以自行参看官方文档。

配置客户端资源

客户端资源的设置与Lettuce的性能、并发和事件处理相关。线程池或者线程组相关配置占据客户端资源配置的大部分（EventLoopGroups和EventExecutorGroup），这些线程池或者线程组是连接程序的基础组件。一般情况下，客户端资源应该在多个Redis客户端之间共享，并且在不再使用的时候需要自行关闭。笔者认为，客户端资源是面向Netty的。注意：除非特别熟悉或者花长时间去测试调整下面提到的参数，否则在没有经验的前提下凭直觉修改默认值，有可能会踩坑。

客户端资源接口是ClientResources，实现类是DefaultClientResources。

构建DefaultClientResources实例：

// 默认
ClientResources resources = DefaultClientResources.create();// 建造器
ClientResources resources = DefaultClientResources.builder().ioThreadPoolSize(4).computationThreadPoolSize(4).build()

使用：

ClientResources resources = DefaultClientResources.create();
// 非集群
RedisClient client = RedisClient.create(resources, uri);
// 集群
RedisClusterClient clusterClient = RedisClusterClient.create(resources, uris);
// ......
client.shutdown();
clusterClient.shutdown();
// 关闭资源
resources.shutdown();

客户端资源基本配置：

属性	描述	默认值
ioThreadPoolSize	I/O线程数	Runtime.getRuntime().availableProcessors()
computationThreadPoolSize	任务线程数	Runtime.getRuntime().availableProcessors()

客户端资源高级配置：

属性	描述	默认值
eventLoopGroupProvider	EventLoopGroup提供商	-
eventExecutorGroupProvider	EventExecutorGroup提供商	-
eventBus	事件总线	DefaultEventBus
commandLatencyCollectorOptions	命令延时收集器配置	DefaultCommandLatencyCollectorOptions
commandLatencyCollector	命令延时收集器	DefaultCommandLatencyCollector
commandLatencyPublisherOptions	命令延时发布器配置	DefaultEventPublisherOptions
dnsResolver	DNS处理器	JDK或者Netty提供
reconnectDelay	重连延时配置	Delay.exponential()
nettyCustomizer	Netty自定义配置器	-
tracing	轨迹记录器	-

非集群客户端RedisClient的属性配置：

Redis非集群客户端RedisClient本身提供了配置属性方法：

RedisClient client = RedisClient.create(uri);
client.setOptions(ClientOptions.builder().autoReconnect(false).pingBeforeActivateConnection(true).build());

非集群客户端的配置属性列表：

属性	描述	默认值
pingBeforeActivateConnection	连接激活之前是否执行PING命令	false
autoReconnect	是否自动重连	true
cancelCommandsOnReconnectFailure	重连失败是否拒绝命令执行	false
suspendReconnectOnProtocolFailure	底层协议失败是否挂起重连操作	false
requestQueueSize	请求队列容量	2147483647(Integer#MAX_VALUE)
disconnectedBehavior	失去连接时候的行为	DEFAULT
sslOptions	SSL配置	-
socketOptions	Socket配置	10 seconds Connection-Timeout, no keep-alive, no TCP noDelay
timeoutOptions	超时配置	-
publishOnScheduler	发布反应式信号数据的调度器	使用I/O线程

集群客户端属性配置：

Redis集群客户端RedisClusterClient本身提供了配置属性方法：

RedisClusterClient client = RedisClusterClient.create(uri);
ClusterTopologyRefreshOptions topologyRefreshOptions = ClusterTopologyRefreshOptions.builder().enablePeriodicRefresh(refreshPeriod(10, TimeUnit.MINUTES)).enableAllAdaptiveRefreshTriggers().build();client.setOptions(ClusterClientOptions.builder().topologyRefreshOptions(topologyRefreshOptions).build());

集群客户端的配置属性列表：

属性	描述	默认值
enablePeriodicRefresh	是否允许周期性更新集群拓扑视图	false
refreshPeriod	更新集群拓扑视图周期	60秒
enableAdaptiveRefreshTrigger	设置自适应更新集群拓扑视图触发器RefreshTrigger	-
adaptiveRefreshTriggersTimeout	自适应更新集群拓扑视图触发器超时设置	30秒
refreshTriggersReconnectAttempts	自适应更新集群拓扑视图触发重连次数	5
dynamicRefreshSources	是否允许动态刷新拓扑资源	true
closeStaleConnections	是否允许关闭陈旧的连接	true
maxRedirects	集群重定向次数上限	5
validateClusterNodeMembership	是否校验集群节点的成员关系	true

使用连接池

引入连接池依赖commons-pool2：

<dependency><groupId>org.apache.commons</groupId><artifactId>commons-pool2</artifactId><version>2.7.0</version>
</dependency

基本使用如下：

@Test
public void testUseConnectionPool() throws Exception {RedisURI redisUri = RedisURI.builder().withHost("localhost").withPort(6379).withTimeout(Duration.of(10, ChronoUnit.SECONDS)).build();RedisClient redisClient = RedisClient.create(redisUri);GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();GenericObjectPool<StatefulRedisConnection<String, String>> pool= ConnectionPoolSupport.createGenericObjectPool(redisClient::connect, poolConfig);try (StatefulRedisConnection<String, String> connection = pool.borrowObject()) {RedisCommands<String, String> command = connection.sync();SetArgs setArgs = SetArgs.Builder.nx().ex(5);command.set("name", "throwable", setArgs);String n = command.get("name");log.info("Get value:{}", n);}pool.close();redisClient.shutdown();
}

其中，同步连接的池化支持需要用ConnectionPoolSupport，异步连接的池化支持需要用AsyncConnectionPoolSupport（Lettuce5.1之后才支持）。

几个常见的渐进式删除例子

渐进式删除Hash中的域-属性：

@Test
public void testDelBigHashKey() throws Exception {// SCAN参数ScanArgs scanArgs = ScanArgs.Builder.limit(2);// TEMP游标ScanCursor cursor = ScanCursor.INITIAL;// 目标KEYString key = "BIG_HASH_KEY";prepareHashTestData(key);log.info("开始渐进式删除Hash的元素...");int counter = 0;do {MapScanCursor<String, String> result = COMMAND.hscan(key, cursor, scanArgs);// 重置TEMP游标cursor = ScanCursor.of(result.getCursor());cursor.setFinished(result.isFinished());Collection<String> fields = result.getMap().values();if (!fields.isEmpty()) {COMMAND.hdel(key, fields.toArray(new String[0]));}counter++;} while (!(ScanCursor.FINISHED.getCursor().equals(cursor.getCursor()) && ScanCursor.FINISHED.isFinished() == cursor.isFinished()));log.info("渐进式删除Hash的元素完毕,迭代次数:{} ...", counter);
}private void prepareHashTestData(String key) throws Exception {COMMAND.hset(key, "1", "1");COMMAND.hset(key, "2", "2");COMMAND.hset(key, "3", "3");COMMAND.hset(key, "4", "4");COMMAND.hset(key, "5", "5");
}

渐进式删除集合中的元素：

@Test
public void testDelBigSetKey() throws Exception {String key = "BIG_SET_KEY";prepareSetTestData(key);// SCAN参数ScanArgs scanArgs = ScanArgs.Builder.limit(2);// TEMP游标ScanCursor cursor = ScanCursor.INITIAL;log.info("开始渐进式删除Set的元素...");int counter = 0;do {ValueScanCursor<String> result = COMMAND.sscan(key, cursor, scanArgs);// 重置TEMP游标cursor = ScanCursor.of(result.getCursor());cursor.setFinished(result.isFinished());List<String> values = result.getValues();if (!values.isEmpty()) {COMMAND.srem(key, values.toArray(new String[0]));}counter++;} while (!(ScanCursor.FINISHED.getCursor().equals(cursor.getCursor()) && ScanCursor.FINISHED.isFinished() == cursor.isFinished()));log.info("渐进式删除Set的元素完毕,迭代次数:{} ...", counter);
}private void prepareSetTestData(String key) throws Exception {COMMAND.sadd(key, "1", "2", "3", "4", "5");
}

渐进式删除有序集合中的元素：

@Test
public void testDelBigZSetKey() throws Exception {// SCAN参数ScanArgs scanArgs = ScanArgs.Builder.limit(2);// TEMP游标ScanCursor cursor = ScanCursor.INITIAL;// 目标KEYString key = "BIG_ZSET_KEY";prepareZSetTestData(key);log.info("开始渐进式删除ZSet的元素...");int counter = 0;do {ScoredValueScanCursor<String> result = COMMAND.zscan(key, cursor, scanArgs);// 重置TEMP游标cursor = ScanCursor.of(result.getCursor());cursor.setFinished(result.isFinished());List<ScoredValue<String>> scoredValues = result.getValues();if (!scoredValues.isEmpty()) {COMMAND.zrem(key, scoredValues.stream().map(ScoredValue<String>::getValue).toArray(String[]::new));}counter++;} while (!(ScanCursor.FINISHED.getCursor().equals(cursor.getCursor()) && ScanCursor.FINISHED.isFinished() == cursor.isFinished()));log.info("渐进式删除ZSet的元素完毕,迭代次数:{} ...", counter);
}private void prepareZSetTestData(String key) throws Exception {COMMAND.zadd(key, 0, "1");COMMAND.zadd(key, 0, "2");COMMAND.zadd(key, 0, "3");COMMAND.zadd(key, 0, "4");COMMAND.zadd(key, 0, "5");
}

在SpringBoot中使用Lettuce

个人认为，spring-data-redis中的API封装并不是很优秀，用起来比较重，不够灵活，这里结合前面的例子和代码，在SpringBoot脚手架项目中配置和整合Lettuce。先引入依赖：

<dependencyManagement><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-dependencies</artifactId><version>2.1.8.RELEASE</version><type>pom</type><scope>import</scope></dependency></dependencies>
</dependencyManagement>
<dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>io.lettuce</groupId><artifactId>lettuce-core</artifactId><version>5.1.8.RELEASE</version></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.18.10</version><scope>provided</scope></dependency>
</dependencies>        

一般情况下，每个应用应该使用单个Redis客户端实例和单个连接实例，这里设计一个脚手架，适配单机、普通主从、哨兵和集群四种使用场景。对于客户端资源，采用默认的实现即可。对于Redis的连接属性，比较主要的有Host、Port和Password，其他可以暂时忽略。基于约定大于配置的原则，先定制一系列属性配置类（其实有些配置是可以完全共用，但是考虑到要清晰描述类之间的关系，这里拆分多个配置属性类和多个配置方法）：

@Data
@ConfigurationProperties(prefix = "lettuce")
public class LettuceProperties {private LettuceSingleProperties single;private LettuceReplicaProperties replica;private LettuceSentinelProperties sentinel;private LettuceClusterProperties cluster;}@Data
public class LettuceSingleProperties {private String host;private Integer port;private String password;
}@EqualsAndHashCode(callSuper = true)
@Data
public class LettuceReplicaProperties extends LettuceSingleProperties {}@EqualsAndHashCode(callSuper = true)
@Data
public class LettuceSentinelProperties extends LettuceSingleProperties {private String masterId;
}@EqualsAndHashCode(callSuper = true)
@Data
public class LettuceClusterProperties extends LettuceSingleProperties {}

配置类如下，主要使用@ConditionalOnProperty做隔离，一般情况下，很少有人会在一个应用使用一种以上的Redis连接场景：

@RequiredArgsConstructor
@Configuration
@ConditionalOnClass(name = "io.lettuce.core.RedisURI")
@EnableConfigurationProperties(value = LettuceProperties.class)
public class LettuceAutoConfiguration {private final LettuceProperties lettuceProperties;@Bean(destroyMethod = "shutdown")public ClientResources clientResources() {return DefaultClientResources.create();}@Bean(destroyMethod = "shutdown")@ConditionalOnProperty(name = "lettuce.single.host")public RedisClient singleRedisClient(ClientResources clientResources) {LettuceSingleProperties singleProperties = lettuceProperties.getSingle();RedisURI uri = RedisURI.builder().withHost(singleProperties.getHost()).withPort(singleProperties.getPort()).withPassword(singleProperties.getPassword()).build();return RedisClient.create(clientResources, uri);}@Bean(destroyMethod = "close")@ConditionalOnProperty(name = "lettuce.single.host")public StatefulRedisConnection<String, String> singleRedisConnection(@Qualifier("singleRedisClient") RedisClient singleRedisClient) {return singleRedisClient.connect();}@Bean(destroyMethod = "shutdown")@ConditionalOnProperty(name = "lettuce.replica.host")public RedisClient replicaRedisClient(ClientResources clientResources) {LettuceReplicaProperties replicaProperties = lettuceProperties.getReplica();RedisURI uri = RedisURI.builder().withHost(replicaProperties.getHost()).withPort(replicaProperties.getPort()).withPassword(replicaProperties.getPassword()).build();return RedisClient.create(clientResources, uri);}@Bean(destroyMethod = "close")@ConditionalOnProperty(name = "lettuce.replica.host")public StatefulRedisConnection<String, String> replicaRedisConnection(@Qualifier("replicaRedisClient") RedisClient replicaRedisClient) {return replicaRedisClient.connect();}@Bean(destroyMethod = "shutdown")@ConditionalOnProperty(name = "lettuce.sentinel.host")public RedisClient sentinelRedisClient(ClientResources clientResources) {LettuceSentinelProperties sentinelProperties = lettuceProperties.getSentinel();RedisURI uri = RedisURI.builder().withHost(sentinelProperties.getHost()).withPort(sentinelProperties.getPort()).withPassword(sentinelProperties.getPassword()).withSentinel(sentinelProperties.getHost(), sentinelProperties.getPort()).withSentinelMasterId(sentinelProperties.getMasterId()).build();return RedisClient.create(clientResources, uri);}@Bean(destroyMethod = "close")@ConditionalOnProperty(name = "lettuce.sentinel.host")public StatefulRedisConnection<String, String> sentinelRedisConnection(@Qualifier("sentinelRedisClient") RedisClient sentinelRedisClient) {return sentinelRedisClient.connect();}@Bean(destroyMethod = "shutdown")@ConditionalOnProperty(name = "lettuce.cluster.host")public RedisClusterClient redisClusterClient(ClientResources clientResources) {LettuceClusterProperties clusterProperties = lettuceProperties.getCluster();RedisURI uri = RedisURI.builder().withHost(clusterProperties.getHost()).withPort(clusterProperties.getPort()).withPassword(clusterProperties.getPassword()).build();return RedisClusterClient.create(clientResources, uri);}@Bean(destroyMethod = "close")@ConditionalOnProperty(name = "lettuce.cluster")public StatefulRedisClusterConnection<String, String> clusterConnection(RedisClusterClient clusterClient) {return clusterClient.connect();}
}

最后为了让IDE识别我们的配置，可以添加IDE亲缘性，/META-INF文件夹下新增一个文件spring-configuration-metadata.json，内容如下：

{"properties": [{"name": "lettuce.single","type": "club.throwable.spring.lettuce.LettuceSingleProperties","description": "单机配置","sourceType": "club.throwable.spring.lettuce.LettuceProperties"},{"name": "lettuce.replica","type": "club.throwable.spring.lettuce.LettuceReplicaProperties","description": "主从配置","sourceType": "club.throwable.spring.lettuce.LettuceProperties"},{"name": "lettuce.sentinel","type": "club.throwable.spring.lettuce.LettuceSentinelProperties","description": "哨兵配置","sourceType": "club.throwable.spring.lettuce.LettuceProperties"},{"name": "lettuce.single","type": "club.throwable.spring.lettuce.LettuceClusterProperties","description": "集群配置","sourceType": "club.throwable.spring.lettuce.LettuceProperties"}]
}

如果想IDE亲缘性做得更好，可以添加/META-INF/additional-spring-configuration-metadata.json进行更多细节定义。简单使用如下：

@Slf4j
@Component
public class RedisCommandLineRunner implements CommandLineRunner {@Autowired@Qualifier("singleRedisConnection")private StatefulRedisConnection<String, String> connection;@Overridepublic void run(String... args) throws Exception {RedisCommands<String, String> redisCommands = connection.sync();redisCommands.setex("name", 5, "throwable");log.info("Get value:{}", redisCommands.get("name"));}
}
// Get value:throwable

小结

本文算是基于Lettuce的官方文档，对它的使用进行全方位的分析，包括主要功能、配置都做了一些示例，限于篇幅部分特性和配置细节没有分析。Lettuce已经被spring-data-redis接纳作为官方的Redis客户端驱动，所以值得信赖，它的一些API设计确实比较合理，扩展性高的同时灵活性也高。个人建议，基于Lettuce包自行添加配置到SpringBoot应用用起来会得心应手，毕竟RedisTemplate实在太笨重，而且还屏蔽了Lettuce一些高级特性和灵活的API。

参考资料：

• Lettuce Reference Guide

其他文章

你真的会高效的在GitHub搜索开源项目吗?

如何给Spring Boot 的嵌入式 Tomcat 部署多个应用?

 Java虚拟机的显微镜 Serviceability Agent

怎样回答技术面试题？

深入Tomcat源码分析Session到底是个啥!

支付宝54.4万笔/秒支付的架构是如何设计的？

大佬分享：API网关在微服务架构中的应用

常用的 19 条 MySQL 优化

微服务之间的最佳调用方式

更多常见问题，请关注公众号，在菜单「常见问题」中查看。

源码|实战|成长|职场

这里是「Tomcat那些事儿」

请留下你的足迹

我们一起「终身成长」