java延迟加载 dbutils_Lettuce「建议收藏」

[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 依赖如下：

~~~

io.lettuce

lettuce-core

5.0.4.RELEASE

~~~

~~~

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 connect = redisClient.connect();

RedisCommands 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 connect = redisClient.connect();

RedisCommands 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 redisURIList = new ArrayList();

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 clusterCon = redisClusterClient.connect();

// 获取集群同步命令对象

RedisClusterCommands 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 clusterCon = redisClusterClient.connect();

~~~

不知读者是否思考过一个问题：集群连接和单机连接到底有什么区别？为什么一个集群连接就可以操作集群？事实上，所谓集群连接本质上就是一个单机连接的集合，即集群连接包含了到集群中所有节点的连接(单机连接)。既然如此，在集群模式下，当我们需要用到单机连接时，就不必再创建连接了，而是直接从集群连接中“取”出需要的单机连接，这是非常有益的，可以极大的减少资源的消耗，提升性能。如下实例：

~~~

// 创建集Redis集群模式客户端

RedisClusterClient redisClusterClient = RedisClusterClient.create(redisURIList);

// 连接到Redis集群

StatefulRedisClusterConnection clusterCon = redisClusterClient.connect();

// 从集群连接中取出单机连接

// 方式1:根据ip和端口获取单机连接

StatefulRedisConnection conn1 = clusterCon.getConnection(host, port);

// 方式2:根据nodeId获取单机连接

StatefulRedisConnection 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 clusterNodeList = new ArrayList();

List masterNodeList = new ArrayList();

List slaveNodeList = new ArrayList();

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 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 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 connection) {

this.connection = connection;

}

public void setClient(RedisClient client) {

this.redisClient = client;

}

public StatefulRedisConnection 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 redisURIList = new ArrayList();

// 笔者在一台物理机上启动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 clusterCon = redisClusterClient.connect();

// 获取集群同步命令对象

RedisClusterCommands 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，进而出现了内存溢出。

**规避方法：**

简而言之，对于不再使用的客户端和连接一定要显示的关闭，如下代码所示：