文章目錄

• Pre
• Code
• 初始化
• 槽計(jì)算
• 無需手工調(diào)用close方法

---

Pre

Redis進(jìn)階-Redis集群原理剖析及gossip協(xié)議初探 集群原理部分 簡單的提了下Jest是如何實(shí)現(xiàn)Redis Cluster 的 ，這里我們?cè)賮硎崂硪幌?/p>

---

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：指的是連接一個(gè)url的連接等待時(shí)間//soTimeout：指的是連接上一個(gè)url，獲取response的返回等待時(shí)間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();}} }

這里是個(gè)簡單的demo， 生產(chǎn)中用的話，需要確保jedisCluster是單例的，并且無需手工調(diào)用close，不然的話 這個(gè)連接池就關(guān)閉了，你就無法獲取到連接了。

---

初始化

當(dāng) Redis Cluster 的客戶端來連接集群時(shí)，它也會(huì)得到一份集群的槽位配置信息并將其緩存在客戶端本地。這樣當(dāng)客戶端要查找某個(gè) key 時(shí)，可以直接定位到目標(biāo)節(jié)點(diǎn)。

我們來看下jedis的實(shí)現(xiàn)

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);}

重點(diǎn)看下 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();}}}}

繼續(xù) cache.discoverClusterNodesAndSlots(jedis); cache為 JedisClusterInfoCache 對(duì)象。

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();}}

---

槽計(jì)算

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

CRC16算法，計(jì)算key對(duì)應(yīng)的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);}

---

無需手工調(diào)用close方法

進(jìn)入到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);}

命令模式， 關(guān)注 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);}

繼續(xù) 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);}}

關(guān)注點(diǎn)

• CRC16算法，計(jì)算key對(duì)應(yīng)的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 獲取連接

關(guān)注下 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();}}}

本質(zhì)上還是通過 JedisPool 來獲取一個(gè)getResource ，跟我們使用Sentinel 啊 單節(jié)點(diǎn)獲取方法是一樣的

---

• finally 語句中的 releaseConnection(connection); ,自動(dòng)釋放連接

看下該方法

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

說白了，JedisCluster set后會(huì)自動(dòng)釋放連接，調(diào)用的是jedis 的close方法，所以我們無需手工關(guān)閉，否則你這個(gè)jedis的連接池就掛逼了…

總結(jié)

以上是生活随笔為你收集整理的Redis进阶-JedisCluster初始化 自动管理连接池中的连接 _ 源码分析的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。