文章目录

Pre
Code
初始化
槽计算
无需手工调用close方法

Pre

Redis进阶-Redis集群原理剖析及gossip协议初探集群原理部分简单的提了下Jest是如何实现Redis Cluster 的，这里我们再来梳理一下

Code


import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.JedisPoolConfig;import java.io.IOException;
import java.util.HashSet;
import java.util.Set;public class JedisClusterDemo {public static void main(String[] args) throws IOException {JedisPoolConfig config = new JedisPoolConfig();config.setMaxTotal(20);config.setMaxIdle(10);config.setMinIdle(5);Set<HostAndPort> jedisClusterNode = new HashSet<HostAndPort>();jedisClusterNode.add(new HostAndPort("192.168.18.131", 8001));jedisClusterNode.add(new HostAndPort("192.168.18.131", 8004));jedisClusterNode.add(new HostAndPort("192.168.18.132", 8002));jedisClusterNode.add(new HostAndPort("192.168.18.132", 8005));jedisClusterNode.add(new HostAndPort("192.168.18.133", 8003));jedisClusterNode.add(new HostAndPort("192.168.18.133", 8006));JedisCluster jedisCluster = null;try {//connectionTimeout：指的是连接一个url的连接等待时间//soTimeout：指的是连接上一个url，获取response的返回等待时间jedisCluster = new JedisCluster(jedisClusterNode, 6000, 5000, 10, "artisan", config);System.out.println(jedisCluster.set("clusterArtisan", "artisanValue"));System.out.println(jedisCluster.get("clusterArtisan"));} catch (Exception e) {e.printStackTrace();} finally {if (jedisCluster != null)jedisCluster.close();}}
}

这里是个简单的demo，生产中用的话，需要确保jedisCluster是单例的，并且无需手工调用close，不然的话这个连接池就关闭了，你就无法获取到连接了。

初始化

当 Redis Cluster 的客户端来连接集群时，它也会得到一份集群的槽位配置信息并将其缓存在客户端本地。这样当客户端要查找某个 key 时，可以直接定位到目标节点。

我们来看下jedis的实现

 jedisCluster = new JedisCluster(jedisClusterNode, 6000, 5000, 10, "artisan", config);

跟下源码

public JedisClusterConnectionHandler(Set<HostAndPort> nodes,final GenericObjectPoolConfig poolConfig, int connectionTimeout, int soTimeout, String password) {this.cache = new JedisClusterInfoCache(poolConfig, connectionTimeout, soTimeout, password);initializeSlotsCache(nodes, poolConfig, password);}

重点看下 initializeSlotsCache

  private void initializeSlotsCache(Set<HostAndPort> startNodes, GenericObjectPoolConfig poolConfig, String password) {for (HostAndPort hostAndPort : startNodes) {Jedis jedis = new Jedis(hostAndPort.getHost(), hostAndPort.getPort());if (password != null) {jedis.auth(password);}try {cache.discoverClusterNodesAndSlots(jedis);break;} catch (JedisConnectionException e) {// try next nodes} finally {if (jedis != null) {jedis.close();}}}}

继续 cache.discoverClusterNodesAndSlots(jedis); cache为 JedisClusterInfoCache 对象。

  public void discoverClusterNodesAndSlots(Jedis jedis) {w.lock();try {reset();List<Object> slots = jedis.clusterSlots();for (Object slotInfoObj : slots) {List<Object> slotInfo = (List<Object>) slotInfoObj;if (slotInfo.size() <= MASTER_NODE_INDEX) {continue;}List<Integer> slotNums = getAssignedSlotArray(slotInfo);// hostInfosint size = slotInfo.size();for (int i = MASTER_NODE_INDEX; i < size; i++) {List<Object> hostInfos = (List<Object>) slotInfo.get(i);if (hostInfos.size() <= 0) {continue;}HostAndPort targetNode = generateHostAndPort(hostInfos);setupNodeIfNotExist(targetNode);if (i == MASTER_NODE_INDEX) {assignSlotsToNode(slotNums, targetNode);}}}} finally {w.unlock();}}

槽计算

set --------> run  ----> runWithRetries ----> connectionHandler.getConnectionFromSlot(JedisClusterCRC16.getSlot(key))

CRC16算法，计算key对应的slot connectionHandler.getConnectionFromSlot(JedisClusterCRC16.getSlot(key))

  public static int getSlot(byte[] key) {int s = -1;int e = -1;boolean sFound = false;for (int i = 0; i < key.length; i++) {if (key[i] == '{' && !sFound) {s = i;sFound = true;}if (key[i] == '}' && sFound) {e = i;break;}}if (s > -1 && e > -1 && e != s + 1) {return getCRC16(key, s + 1, e) & (16384 - 1);}return getCRC16(key) & (16384 - 1);}

无需手工调用close方法

进入到set方法中看下源码

jedisCluster.set("clusterArtisan", "artisanValue")

如下：

  @Overridepublic String set(final String key, final String value) {return new JedisClusterCommand<String>(connectionHandler, maxAttempts) {@Overridepublic String execute(Jedis connection) {return connection.set(key, value);}}.run(key);}

命令模式，关注 run方法

  public T run(String key) {if (key == null) {throw new JedisClusterException("No way to dispatch this command to Redis Cluster.");}return runWithRetries(SafeEncoder.encode(key), this.maxAttempts, false, false);}

继续 runWithRetries ，截取核心逻辑

private T runWithRetries(byte[] key, int attempts, boolean tryRandomNode, boolean asking) {Jedis connection = null;try {if (asking) {......} else {if (tryRandomNode) {connection = connectionHandler.getConnection();} else {connection = connectionHandler.getConnectionFromSlot(JedisClusterCRC16.getSlot(key));}}return execute(connection);} finally {releaseConnection(connection);}}

关注点

CRC16算法，计算key对应的slot connectionHandler.getConnectionFromSlot(JedisClusterCRC16.getSlot(key))

  public static int getSlot(byte[] key) {int s = -1;int e = -1;boolean sFound = false;for (int i = 0; i < key.length; i++) {if (key[i] == '{' && !sFound) {s = i;sFound = true;}if (key[i] == '}' && sFound) {e = i;break;}}if (s > -1 && e > -1 && e != s + 1) {return getCRC16(key, s + 1, e) & (16384 - 1);}return getCRC16(key) & (16384 - 1);}

getConnectionFromSlot 通过 JedisPool 获取连接

关注下 JedisCluster是如何获取连接的 getConnectionFromSlot 方法

  @Overridepublic Jedis getConnectionFromSlot(int slot) {JedisPool connectionPool = cache.getSlotPool(slot);if (connectionPool != null) {// It can't guaranteed to get valid connection because of node// assignmentreturn connectionPool.getResource();} else {renewSlotCache(); //It's abnormal situation for cluster mode, that we have just nothing for slot, try to rediscover stateconnectionPool = cache.getSlotPool(slot);if (connectionPool != null) {return connectionPool.getResource();} else {//no choice, fallback to new connection to random nodereturn getConnection();}}}

本质上还是通过 JedisPool 来获取一个getResource ，跟我们使用Sentinel 啊单节点获取方法是一样的

finally 语句中的 releaseConnection(connection); ,自动释放连接

看下该方法

  private void releaseConnection(Jedis connection) {if (connection != null) {connection.close();}}

说白了，JedisCluster set后会自动释放连接，调用的是jedis 的close方法，所以我们无需手工关闭，否则你这个jedis的连接池就挂逼了…

Redis进阶-JedisCluster初始化自动管理连接池中的连接 _ 源码分析相关推荐

聊聊 SQLAlchemy 连接池中的连接失效问题
最近项目中事情比较多,也遇到了一些问题,其中有一个是关于连接池的,比较有意思,这里分享下. 一天早上,进入业务系统,点击了一个功能按钮,页面上突然弹出个 MySQL gone away 的错误,我擦, ...
mosquitto客户端对象“struct mosquitto *mosq”管理下篇（mosquitto2.0.15客户端源码分析之四）
文章目录前言 5 设置网络参数 5.1 客户端连接服务器使用的端口号 `mosq->port` 5.2 指定绑定的网络地址 `mosq->bind_address` 5.3 客户端连接服 ...
Flutter路由管理和页面参数的传递（源码分析）
前言上一篇 Flutter路由管理和页面参数的传递(获取&返回) 文章中我们讲述了这么用代码实现 Flutter 中页面参数的传递,这一篇我们用源码分析一下 Navigator 为什么可以进 ...
用 subsetting 限制连接池中的连接数量
内网使用服务发现后,服务与其它服务的实例之间使用一条 TCP 长连接进行通信.这种情况下常见的做法是按照 registry 下发的 host:port 列表来直接建连. 简单来说就是下图这样: 每一个 ...
千千静听　窗口自动粘合，java　代码实现　源码分析
笔者这几天在写基于 jmf 的音乐播放器. 今天无意中想要加上,"窗口自动粘合".在这里把所有问题全写出来. 在这里说明一下,我使用的是两个窗体,MainView(主界面) ...
java redis释放连接池_Java 使用连接池操作redis
构建连接池对象JedisPool JedisPool jedisPool = new JedisPool(jedisPoolConfig, "127.0.0.1", 6379); ...
jsp获取连接池的实时连接数_一篇看懂数据库连接池概念、原理、运行机制
概述数据库连接池是负责分配.管理和释放数据库连接,它允许应用程序重复使用一个现有的数据库连接,而不是再重新建立一个.那么其中的运行机制又是怎样的呢?今天主要介绍一下数据库连接池原理和常用的连接池. ...
jsp获取连接池的实时连接数_数据库连接池原理分析及模拟实现
数据库访问访问数据库主要有以下几个步骤: 加载数据库驱动创建数据库连接执行访问操作并处理执行结果关闭连接,释放资源在每一次请求数据库都要经历上述过程,创建连接和释放资源也都是些重复性的动作, ...
JDBC、封装JDBC连接池、第三方连接池工具
主要内容: JDBC简介 JDBC来源通过代码实现JDBC JDBC的改进需求 JDBC改进的代码实现 JDBC使用的设计模式封装连接池封装JDBC连接池 ThreadLoacl的使用 Thre ...

Redis进阶-JedisCluster初始化自动管理连接池中的连接 _ 源码分析

文章目录

Pre

Code

初始化

槽计算

无需手工调用close方法

Redis进阶-JedisCluster初始化自动管理连接池中的连接 _ 源码分析相关推荐

最新文章

热门文章

Redis进阶-JedisCluster初始化 自动管理连接池中的连接 _ 源码分析

文章目录

Pre

Code

初始化

槽计算

无需手工调用close方法

Redis进阶-JedisCluster初始化 自动管理连接池中的连接 _ 源码分析相关推荐

最新文章

热门文章

Redis进阶-JedisCluster初始化自动管理连接池中的连接 _ 源码分析

Redis进阶-JedisCluster初始化自动管理连接池中的连接 _ 源码分析相关推荐