Jedis Cluster源码分析

最近一个项目用到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吧

import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.JedisCluster;
import java.util.HashSet;
import java.util.Set;

/**
 * Created by guanxianseng on 2017/8/15.
 *
 * nodes: host:port -> JedisPool
 * slots: slot -> JedisPool
 */
public class TestCluster {
    public static void main(String[] args) {
        Set<HostAndPort> jedisClusterNodes = new HashSet<HostAndPort>();
        jedisClusterNodes.add(new HostAndPort("192.168.211.131", 7340));
        jedisClusterNodes.add(new HostAndPort("192.168.211.131", 7341));
        jedisClusterNodes.add(new HostAndPort("192.168.211.131", 7342));
        JedisCluster jc = new JedisCluster(jedisClusterNodes);
        jc.set("name", "guanxianseng");
        System.out.println(jc.get("name"));
    }
}

输出

guanxianseng

Process finished with exit code 0

这里IP是我的虚拟机的IP，开了两台虚拟机，部署的是三主三从的集群

首先，进入JedisCluster的构造函数，一路找下去，我们会看到这样的代码

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

这里，获取集群节点的jedis对象，进入discoverClusterNodesAndSlots(jedis)

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

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

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

public JedisPool setupNodeIfNotExist(HostAndPort node) {
    w.lock();
    try {
      String nodeKey = getNodeKey(node);
      JedisPool existingPool = nodes.get(nodeKey);
      if (existingPool != null) return existingPool;

      JedisPool nodePool = new JedisPool(poolConfig, node.getHost(), node.getPort(),
          connectionTimeout, soTimeout, password, 0, null, false, null, null, null);
      nodes.put(nodeKey, nodePool);
      return nodePool;
    } finally {
      w.unlock();
    }
  }

这里设置我们一开始提到的nodes， host:port   -------> JedisPool映射

继续回到上面，进入assignSlotsToNode(slotNums, targetNode);

public void assignSlotsToNode(List<Integer> targetSlots, HostAndPort targetNode) {
    w.lock();
    try {
      JedisPool targetPool = setupNodeIfNotExist(targetNode);
      for (Integer slot : targetSlots) {
        slots.put(slot, targetPool);
      }
    } finally {
      w.unlock();
    }
  }

这里设置了前面说的slots， slot ------> JedisPool的映射

这里初始化完成

执行set命令

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

进入run(key);方法

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) {
    if (attempts <= 0) {
      throw new JedisClusterMaxRedirectionsException("Too many Cluster redirections?");
    }

    Jedis connection = null;
    try {

      if (asking) {
        // TODO: Pipeline asking with the original command to make it
        // faster....
        connection = askConnection.get();
        connection.asking();

        // if asking success, reset asking flag
        asking = false;
      } else {
        if (tryRandomNode) {
          connection = connectionHandler.getConnection();
        } else {
          connection = connectionHandler.getConnectionFromSlot(JedisClusterCRC16.getSlot(key));
        }
      }

      return execute(connection);

    } catch (JedisNoReachableClusterNodeException jnrcne) {

这里有点长，截取了前面一部分

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

这里，计算key的slot，从slots获取Jedis对象

到这，基本已完成

总结一下，执行slots命令，缓存host:port --> JedisPool， slot ---->JedisPool映射。执行命令，key ---> slot ----> JedisPool   ------->Jedis