[TOC] # 简介 Lettuce 是一个可伸缩的线程安全的 Redis 客户端，支持同步、异步和响应式模式。多个线程可以共享一个连接实例，而不必担心多线程并发问题。它基于优秀 netty NIO 框架构建，支持 Redis 的高级功能，如 Sentinel，集群，流水线，自动重新连接和 Redis 数据模型。 # redis单机情况 目前，Lettuce 官方发布的最新的版本为[5.0.4](https://lettuce.io/core/5.0.4.RELEASE/api/)，自 5.X 开始，Lettuce 进行了全面重构，与之前的版本相差较大，甚至连包名都全然不同（点击可查看[5.0.4](https://lettuce.io/core/5.0.4.RELEASE/api/)和[4.4.5](https://lettuce.io/lettuce-4/4.4.5.Final/api/)版本），本文基于最新的版本 5.0.4 介绍 Lettuce 的用法，pom 文件中添加 Lettuce 依赖如下： ~~~ <dependency> <groupId>io.lettuce</groupId> <artifactId>lettuce-core</artifactId> <version>5.0.4.RELEASE</version> </dependency> ~~~ ~~~ import io.lettuce.core.RedisClient; import io.lettuce.core.RedisURI; import io.lettuce.core.api.StatefulRedisConnection; import io.lettuce.core.api.sync.RedisCommands; public class Single { public static void main(String[] args) { // 利用redis-server所绑定的IP和Port创建URI， RedisURI redisURI = RedisURI.create("127.0.0.1", 6379); // 创建集Redis单机模式客户端 RedisClient redisClient = RedisClient.create(redisURI); // 开启连接 StatefulRedisConnection<String, String> connect = redisClient.connect(); RedisCommands<String, String> cmd = connect.sync(); // set操作，成功则返回OK cmd.set("key", "value-test"); // get操作，成功命中则返回对应的value，否则返回null cmd.get("key"); // 删除指定的key cmd.del("key"); // 获取redis-server信息，内容极为丰富 cmd.info(); // 列表操作 String[] valuelist = {"China","Americal","England"}; // 将一个或多个值插入到列表头部，此处插入多个 cmd.lpush("listName", valuelist); // 移出并获取列表的第一个元素 System.out.println(cmd.lpop("listName")); // 获取列表长度 System.out.println(cmd.llen("listName")); // 通过索引获取列表中的元素 System.out.println(cmd.lindex("listName", 1)); } } ~~~ 注意 如果 redis-server 设置了访问密码，在进行缓存读写操作之前需要进行鉴权，代码片段如下： ~~~ // 开启连接 StatefulRedisConnection<String, String> connect = redisClient.connect(); RedisCommands<String, String> cmd = connect.sync(); // 如果redis-server设置了访问密码，则需鉴权，否则不可访问 cmd.auth("my-password"); // set操作，成功则返回OK cmd.set("key", "value-test"); ~~~ # redis集群模式 首先介绍一个集群模式下的实例，对比单机模式，读者不难发现，除了创建客户端差别明显外，其它部分几无差别 ~~~ import java.util.ArrayList; import java.util.List; import io.lettuce.core.RedisURI; import io.lettuce.core.cluster.RedisClusterClient; import io.lettuce.core.cluster.api.StatefulRedisClusterConnection; import io.lettuce.core.cluster.api.sync.RedisClusterCommands; public class Cluster { public static void main(String[] args) { // 利用redis-server所绑定的IP和Port创建URI， List<RedisURI> redisURIList = new ArrayList<RedisURI>(); String[] ipSet = {"100.x.x.152","100.x.x.153","100.x.x.154"}; int port = 6379; for (int i=0; i<3; i++) { RedisURI temp = RedisURI.create(ipSet[i], port); redisURIList.add(temp); } // 创建集Redis集群模式客户端 RedisClusterClient redisClusterClient = RedisClusterClient.create(redisURIList); // 连接到Redis集群 StatefulRedisClusterConnection<String, String> clusterCon = redisClusterClient.connect(); // 获取集群同步命令对象 RedisClusterCommands<String, String> commands = clusterCon.sync(); // set操作，成功则返回OK commands.set("key", "value-test"); // get操作，成功命中则返回对应的value，否则返回null commands.get("key"); // 删除指定的key commands.del("key"); // 获取redis-server信息，内容极为丰富 commands.info(); // 列表操作 String[] valuelist = {"China","Americal","England"}; // 将一个或多个值插入到列表头部，此处插入多个 commands.lpush("listName", valuelist); // 移出并获取列表的第一个元素 commands.lpop("listName"); // 获取列表长度 commands.llen("listName"); // 通过索引获取列表中的元素 commands.lindex("listName", 1); } } ~~~ ## 重要接口说明 与单机模式相比，集群模式下命令集要丰富得多，如下图所示， Lettuce 提供的方法可支持集群模式下的所有命令。其中，有几个重要的方法读者需要掌握：clusterAddSlots, clusterFailover, clusterForget, clusterInfo, clusterMeet, clusterNodes 及clusterReplicate  1. clusterMeet(String ip, int port)：以当前节点为基准，将 ip 和 port 所对应的节点纳入集群； 2. clusterAddSlots(int ...slots)：为当前节点指派 slot，只有被指派 slot 的节点才是真正意义上的 master； 3. clusterReplicate(String nodeId)：将当前节点设置为 nodeId 所对应的主节点的从； 4. clusterFailover(boolean force)：发起故障倒换，将当前节点升为主节点，当前节点原本对应的主节点则降为从节点； 5. clusterForget(String nodeId)：将 nodeId 所对应的节点从集群中删除； 6. clusterInfo()：获取集群运行状态信息； 7. clusterNodes()：获取集群节点的详细信息； ## 小技巧 使用 Lettuce时，创建客户端之后还需连接到集群方可，分别调用了 create() 方法和 connect() 方法，如下代码片段所示： ~~~ // 创建集Redis集群模式客户端 RedisClusterClient redisClusterClient = RedisClusterClient.create(redisURIList); // 连接到Redis集群 StatefulRedisClusterConnection<String, String> clusterCon = redisClusterClient.connect(); ~~~ 不知读者是否思考过一个问题：集群连接和单机连接到底有什么区别？为什么一个集群连接就可以操作集群？事实上，所谓集群连接本质上就是一个单机连接的集合，即集群连接包含了到集群中所有节点的连接（单机连接）。既然如此，在集群模式下，当我们需要用到单机连接时，就不必再创建连接了，而是直接从集群连接中“取”出需要的单机连接，这是非常有益的，可以极大的减少资源的消耗，提升性能。如下实例： ~~~ // 创建集Redis集群模式客户端 RedisClusterClient redisClusterClient = RedisClusterClient.create(redisURIList); // 连接到Redis集群 StatefulRedisClusterConnection<String, String> clusterCon = redisClusterClient.connect(); // 从集群连接中取出单机连接 // 方式1:根据ip和端口获取单机连接 StatefulRedisConnection<String, String> conn1 = clusterCon.getConnection(host, port); // 方式2:根据nodeId获取单机连接 StatefulRedisConnection<String, String> conn2 = clusterCon.getConnection(nodeId); ~~~ # Lettuce 创建 Redis 集群 Redis 集群模式至少需要 3 个主节点，作为举例，本文搭建一个 3 主 3 从的精简集群，麻雀虽小，五脏俱全。主从关系如下图所示，其中 M 代码 Master 节点，S 代表 Slave 节点，A-M 和 A-S 为一对主从节点。  由于笔者只有一台物理机，因此在同一台机器上分别启动 6 个 redis-server 进程以创建 3 主 3 从Redis集群，6 个 redis-server 进程分别绑定端口号为 6379，6380，6381，6382，6383，6384 ## Redis 集群创建的步骤 **（1）相互感知，初步形成集群** 在上文中，我们已经成功拉起了 6 个 redis-server 进程，每个进程视为一个节点，这些节点仍处于孤立状态，它们相互之间无法感知对方的存在，既然要创建集群，首先需要让这些孤立的节点相互感知，形成一个集群； **（2）分配 Slot 给期望的主节点** 形成集群之后，仍然无法提供服务，Redis 集群模式下，数据存储于 16384 个 Slot 中，我们需要将这些 Slot 指派给期望的主节点。何为期望呢？我们有 6 个节点，3 主 3 备，我们只能将 Slot 指派给 3 个主节点，至于哪些节点为主节点，我们可以自定义。 **（3）设置从节点** Slot 分配完成后，被分配 Slot 的节点将成为真正可用的主节点，剩下的没有分到 Slot 的节点，即便状态标志为 Master，实际上也不能提供服务。接下来，出于可靠性的考量，我们需要将这些没有被指派 Slot 的节点指定为可用主节点的从节点（Slave）。 经过上述三个步骤，一个精简的 3 主 3 从 Redis 集群就搭建完成了。 ## 基于 Lettuce 的创建集群代码 根据上述步骤，基于 Lettuce 创建集群的代码如下（仅供入门参考）： ~~~ import java.util.ArrayList; import java.util.List; import io.lettuce.core.RedisClient; import io.lettuce.core.RedisCommandTimeoutException; import io.lettuce.core.RedisConnectionException; import io.lettuce.core.RedisException; import io.lettuce.core.RedisURI; import io.lettuce.core.api.StatefulRedisConnection; public class CreateCluster { public static void main(String[] args) throws InterruptedException { createCluster(); } private static void createCluster() throws InterruptedException { // 初始化集群节点列表，并指定主节点列表和从节点列表 List<ClusterNode> clusterNodeList = new ArrayList<ClusterNode>(); List<ClusterNode> masterNodeList = new ArrayList<ClusterNode>(); List<ClusterNode> slaveNodeList = new ArrayList<ClusterNode>(); String[] endpoints = {"127.0.0.1:6379", "127.0.0.1:6380", "127.0.0.1:6381" , "127.0.0.1:6382", "127.0.0.1:6383", "127.0.0.1:6384"}; int index = 0; for (String endpoint : endpoints) { String[] ipAndPort = endpoint.split(":"); ClusterNode node = new ClusterNode(ipAndPort[0], Integer.parseInt(ipAndPort[1])); clusterNodeList.add(node); // 将6379，6380，6381设置为主节点，其余为从节点 if (index < 3) { masterNodeList.add(node); } else { slaveNodeList.add(node); } index++; } // 分别与各个Redis节点建立通信连接 for (ClusterNode node : clusterNodeList) { RedisURI redisUri = RedisURI.Builder.redis(node.getHost(), node.getPort()).build(); RedisClient redisClient = RedisClient.create(redisUri); try { StatefulRedisConnection<String, String> connection = redisClient.connect(); node.setConnection(connection); } catch (RedisException e) { System.out.println("connection failed-->" + node.getHost() + ":" + node.getPort()); } } // 执行cluster meet命令是各个孤立的节点相互感知，初步形成集群。 // 只需以一个节点为基准，让所有节点与之meet即可 ClusterNode firstNode = null; for (ClusterNode node : clusterNodeList) { if (firstNode == null) { firstNode = node; } else { try { node.getConnection().sync().clusterMeet(firstNode.getHost(), firstNode.getPort()); } catch (RedisCommandTimeoutException | RedisConnectionException e) { System.out.println("meet failed-->" + node.getHost() + ":" + node.getPort()); } } } // 为主节点指派slot,将16384个slot分成三份：5461，5461，5462 int[] slots = {0, 5460, 5461, 10921, 10922, 16383}; index = 0; for (ClusterNode node : masterNodeList) { node.setSlotsBegin(slots[index]); index++; node.setSlotsEnd(slots[index]); index++; } // 通过与各个主节点的连接，执行addSlots命令为主节点指派slot System.out.println("Start to set slots..."); for (ClusterNode node : masterNodeList) { try { node.getConnection().sync().clusterAddSlots(createSlots(node.getSlotsBegin(), node.getSlotsEnd())); } catch (RedisCommandTimeoutException | RedisConnectionException e) { System.out.println("add slots failed-->" + node.getHost() + ":" + node.getPort()); } } // 延时5s，等待slot指派完成 sleep(5000); // 为已经指派slot的主节点设置从节点,6379,6380,6381分别对应6382，6383，6384 index = 0; for (ClusterNode node : slaveNodeList) { try { node.getConnection().sync().clusterReplicate(masterNodeList.get(index).getMyId()); } catch (RedisCommandTimeoutException | RedisConnectionException e) { System.out.println("replicate failed-->" + node.getHost() + ":" + node.getPort()); } } // 关闭连接,销毁客户端，释放资源 for (ClusterNode node : clusterNodeList) { node.getConnection().close(); node.getClient().shutdown(); } } public static int[] createSlots(int from, int to) { int[] result = new int[to - from + 1]; int counter = 0; for (int i = from; i <= to; i++) { result[counter++] = i; } return result; } } ~~~ ~~~ //定义集群节点描述类 class ClusterNode { private String host; private int port; private int slotsBegin; private int slotsEnd; private String myId; private String masterId; private StatefulRedisConnection<String, String> connection; private RedisClient redisClient; public ClusterNode(String host, int port) { this.host = host; this.port = port; this.slotsBegin = 0; this.slotsEnd = 0; this.myId = null; this.masterId = null; } public String getHost() { return host; } public int getPort() { return port; } public void setMaster(String masterId) { this.masterId = masterId; } public String getMaster() { return masterId; } public void setMyId(String myId) { this.myId = myId; } public String getMyId() { return myId; } public void setSlotsBegin(int first) { this.slotsBegin = first; } public void setSlotsEnd(int last) { this.slotsEnd = last; } public int getSlotsBegin() { return slotsBegin; } public int getSlotsEnd() { return slotsEnd; } public void setConnection(StatefulRedisConnection<String, String> connection) { this.connection = connection; } public void setClient(RedisClient client) { this.redisClient = client; } public StatefulRedisConnection<String, String> getConnection() { return connection; } public RedisClient getClient() { return redisClient; } } ~~~ ## Lettuce方法获取并解析集群状态信息 Lettuce 提供了与 Redis 的 cluster info，cluster nodes 及 info 命令对应的方法，分别为：clusterInfo(), clusterNodes()和info()，是不是觉得很亲切？ 仅仅只是获取信息还不够，如此复杂的信息，虽然人可以一眼看出要点，但用程序来解析却是一件很麻烦的事情。Lettuce 已经考虑到了这一点，为此提供了专门的方法来解析获取到的集群节点信息，以 clusterNodes() 为例：  **例子** 一个可用的 Redis 集群，其 16384 个 slot 必须全部处于正常工作状态，换句话说，这些 slots 对应的 master 必须是正常的。以下我们通过解析 clusterNodes() 方法获取的信息来判断集群状态是否正常，如果不正常，还可以进一步识别出不正常的节点。 **注意** 下面的程序仅仅是举例，事实上，通过解析 clusterNodes() 方法获取的信息可以获取集群节点的运行状态，主从关系，slot 分布等重要信息。  **例子的完整程序** ~~~ import java.util.ArrayList; import java.util.List; import io.lettuce.core.RedisURI; import io.lettuce.core.cluster.RedisClusterClient; import io.lettuce.core.cluster.api.StatefulRedisClusterConnection; import io.lettuce.core.cluster.api.sync.RedisClusterCommands; import io.lettuce.core.cluster.models.partitions.ClusterPartitionParser; import io.lettuce.core.cluster.models.partitions.Partitions; import io.lettuce.core.cluster.models.partitions.RedisClusterNode; public class ClusterState { public static void main(String[] args) { // 利用redis-server所绑定的IP和Port创建URI， List<RedisURI> redisURIList = new ArrayList<RedisURI>(); // 笔者在一台物理机上启动6个redis-server进程，ip均为127.X，端口为6379～6384 String ip = "127.0.0.1"; int port = 6379; for (int i = 0; i < 6; i++) { RedisURI temp = RedisURI.create(ip, port + i); redisURIList.add(temp); } // 创建集Redis集群模式客户端 RedisClusterClient redisClusterClient = RedisClusterClient.create(redisURIList); // 连接到Redis集群 StatefulRedisClusterConnection<String, String> clusterCon = redisClusterClient.connect(); // 获取集群同步命令对象 RedisClusterCommands<String, String> commands = clusterCon.sync(); // 获取集群节点信息并解析 Partitions partitions = ClusterPartitionParser.parse(commands.clusterNodes()); int slotsSize = 0; for (RedisClusterNode partition : partitions) { if (partition.getFlags().contains(RedisClusterNode.NodeFlag.FAIL) || partition.getFlags().contains(RedisClusterNode.NodeFlag.EVENTUAL_FAIL) || partition.getFlags().contains(RedisClusterNode.NodeFlag.NOADDR)) { System.out.println("The node's state is not normal：" + partition.getUri()); continue; } slotsSize += partition.getSlots().size(); } if (slotsSize < 16384) { System.out.println("Cluster_slots_assigned is:" + slotsSize); } else { System.out.println("Cluster_state is OK."); } } } ~~~ # 遇到的问题 **堆内存溢出事件** 在实际应用场景下，Redis 集群可能出现节点故障下线、新节点加入、主从倒换等事件，这些事件都会导致 Redis 集群拓扑结构改变，作为客户端的 Lettuce 自然也需要刷新保存的拓扑结构甚至重新建立连接，否则，客户端与服务端之间的通道可能无法工作。 出于对上述原因考虑，为提高可用性，笔者曾经主导过的一个项目通过一个线程来定时检测连接是否可用，如果不可用便重建连接。但是，当时犯了一个错误：重建连接时，仅仅关闭了旧的连接，却没有销毁客户端，而客户端是极为占用资源的。 由于连接不可用的场景并不多，上述问题一直处于潜伏状态，直到有一天网络出现问题，因连接不可用而一次次重建连接，同时重建了客户端。一段时间后，Lettuce 相关的线程竟堆积了近 300 个，而相关进程预设的内存不过 2G，进而出现了内存溢出。 **规避方法：** 简而言之，对于不再使用的客户端和连接一定要显示的关闭，如下代码所示：