最近一個項目用到Jedis客戶端，需要對這個客戶端進行改造。看了一下Jedis Cluster源碼，做個記錄

首先，說核心內容， 在Jedis源碼中，關于cluster有個兩個重要的map。一個是nodes，一個是slots

nodes: ?host:port ?----> ?JedisPool

slots: ?slot ----> JedisPool

nodes存放的是key為host:port到JedisPool的映射

slots存放的 slot到JedisPool的映射

這里，JedisPool是用apache common pool存放jedis對象的pool，slot是通過Crc16算出對16384取余得到

上個Jedis Cluster的Demo吧

1 import redis.clients.jedis.HostAndPort; 2 import redis.clients.jedis.JedisCluster; 3 import java.util.HashSet; 4 import java.util.Set; 5 6 /** 7 * Created by guanxianseng on 2017/8/15. 8 * 9 * nodes: host:port -> JedisPool 10 * slots: slot -> JedisPool 11 */ 12 public class TestCluster { 13 public static void main(String[] args) { 14 Set<HostAndPort> jedisClusterNodes = new HashSet<HostAndPort>(); 15 jedisClusterNodes.add(new HostAndPort("192.168.211.131", 7340)); 16 jedisClusterNodes.add(new HostAndPort("192.168.211.131", 7341)); 17 jedisClusterNodes.add(new HostAndPort("192.168.211.131", 7342)); 18 JedisCluster jc = new JedisCluster(jedisClusterNodes); 19 jc.set("name", "guanxianseng"); 20 System.out.println(jc.get("name")); 21 } 22 }

輸出

guanxiansengProcess finished with exit code 0

這里IP是我的虛擬機的IP，開了兩臺虛擬機，部署的是三主三從的集群

首先，進入JedisCluster的構造函數，一路找下去，我們會看到這樣的代碼

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

進入initializeSlotsCache方法

1 private void initializeSlotsCache(Set<HostAndPort> startNodes, GenericObjectPoolConfig poolConfig, String password) { 2 for (HostAndPort hostAndPort : startNodes) { 3 Jedis jedis = new Jedis(hostAndPort.getHost(), hostAndPort.getPort()); 4 if (password != null) { 5 jedis.auth(password); 6 } 7 try { 8 cache.discoverClusterNodesAndSlots(jedis); 9 break; 10 } catch (JedisConnectionException e) { 11 // try next nodes 12 } finally { 13 if (jedis != null) { 14 jedis.close(); 15 } 16 } 17 } 18 }

這里，獲取集群節點的jedis對象，進入discoverClusterNodesAndSlots(jedis)

1 public void discoverClusterNodesAndSlots(Jedis jedis) { 2 w.lock(); 3 4 try { 5 reset(); 6 List<Object> slots = jedis.clusterSlots(); 7 8 for (Object slotInfoObj : slots) { 9 List<Object> slotInfo = (List<Object>) slotInfoObj; 10 11 if (slotInfo.size() <= MASTER_NODE_INDEX) { 12 continue; 13 } 14 15 List<Integer> slotNums = getAssignedSlotArray(slotInfo); 16 17 // hostInfos 18 int size = slotInfo.size(); 19 for (int i = MASTER_NODE_INDEX; i < size; i++) { 20 List<Object> hostInfos = (List<Object>) slotInfo.get(i); 21 if (hostInfos.size() <= 0) { 22 continue; 23 } 24 25 HostAndPort targetNode = generateHostAndPort(hostInfos); 26 setupNodeIfNotExist(targetNode); 27 if (i == MASTER_NODE_INDEX) { 28 assignSlotsToNode(slotNums, targetNode); 29 } 30 } 31 } 32 } finally { 33 w.unlock(); 34 } 35 }

第6行，其實就是執行slots命令。進入getAssignedSlotArray方法

private List<Integer> getAssignedSlotArray(List<Object> slotInfo) {List<Integer> slotNums = new ArrayList<Integer>();for (int slot = ((Long) slotInfo.get(0)).intValue(); slot <= ((Long) slotInfo.get(1)).intValue(); slot++) {slotNums.add(slot);}return slotNums;}

這里獲取了，節點分配的slots

回到上面，進入generateHostAndPort方法

private HostAndPort generateHostAndPort(List<Object> hostInfos) {return new HostAndPort(SafeEncoder.encode((byte[]) hostInfos.get(0)),((Long) hostInfos.get(1)).intValue());}

這里獲取到節點的host和port

回到上面，進入setupNodeIfNotExist(targetNode);

1 public JedisPool setupNodeIfNotExist(HostAndPort node) { 2 w.lock(); 3 try { 4 String nodeKey = getNodeKey(node); 5 JedisPool existingPool = nodes.get(nodeKey); 6 if (existingPool != null) return existingPool; 7 8 JedisPool nodePool = new JedisPool(poolConfig, node.getHost(), node.getPort(), 9 connectionTimeout, soTimeout, password, 0, null, false, null, null, null); 10 nodes.put(nodeKey, nodePool); 11 return nodePool; 12 } finally { 13 w.unlock(); 14 } 15 }

這里設置我們一開始提到的nodes， host:port ? -------> JedisPool映射

繼續回到上面，進入assignSlotsToNode(slotNums, targetNode);

1 public void assignSlotsToNode(List<Integer> targetSlots, HostAndPort targetNode) { 2 w.lock(); 3 try { 4 JedisPool targetPool = setupNodeIfNotExist(targetNode); 5 for (Integer slot : targetSlots) { 6 slots.put(slot, targetPool); 7 } 8 } finally { 9 w.unlock(); 10 } 11 }

這里設置了前面說的slots， slot ------> JedisPool的映射

這里初始化完成

執行set命令

1 @Override 2 public String set(final String key, final String value) { 3 return new JedisClusterCommand<String>(connectionHandler, maxAttempts) { 4 @Override 5 public String execute(Jedis connection) { 6 return connection.set(key, value); 7 } 8 }.run(key); 9 }

進入run(key);方法

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

進入runWithRetries()

1 private T runWithRetries(byte[] key, int attempts, boolean tryRandomNode, boolean asking) { 2 if (attempts <= 0) { 3 throw new JedisClusterMaxRedirectionsException("Too many Cluster redirections?"); 4 } 5 6 Jedis connection = null; 7 try { 8 9 if (asking) { 10 // TODO: Pipeline asking with the original command to make it 11 // faster.... 12 connection = askConnection.get(); 13 connection.asking(); 14 15 // if asking success, reset asking flag 16 asking = false; 17 } else { 18 if (tryRandomNode) { 19 connection = connectionHandler.getConnection(); 20 } else { 21 connection = connectionHandler.getConnectionFromSlot(JedisClusterCRC16.getSlot(key)); 22 } 23 } 24 25 return execute(connection); 26 27 } catch (JedisNoReachableClusterNodeException jnrcne) {

這里有點長，截取了前面一部分

1 connection = connectionHandler.getConnectionFromSlot(JedisClusterCRC16.getSlot(key));

這里，計算key的slot，從slots獲取Jedis對象

到這，基本已完成

?

總結一下，執行slots命令，緩存host:port --> JedisPool， slot ---->JedisPool映射。執行命令，key ---> slot ----> JedisPool ? ------->Jedis

?

?

轉載于:https://www.cnblogs.com/luckygxf/p/7368493.html

總結

以上是生活随笔為你收集整理的Jedis Cluster源码分析的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。