Redis-Jedis的MOVED_ASK实现

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

  private T runWithRetries(byte[] key, int attempts, boolean tryRandomNode, boolean asking) {
    //2.尝试完attempts则抛出异常
    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();
        //7.先发送ASKING命令，之后发送Redis的命令(如Get/Set)
        connection.asking();

        // if asking success, reset asking flag
        asking = false;
      } else {
        if (tryRandomNode) {
          connection = connectionHandler.getConnection();
        } else {
          //3.通过CRC16计算出槽位对应的节点
          connection = connectionHandler.getConnectionFromSlot(JedisClusterCRC16.getSlot(key));
        }
      }
			//请求对应的Redis目标节点
      return execute(connection);
    } catch (JedisNoReachableClusterNodeException jnrcne) {
      throw jnrcne;
    } catch (JedisConnectionException jce) {
      // release current connection before recursion
      releaseConnection(connection);
      connection = null;

      if (attempts <= 1) {
        //We need this because if node is not reachable anymore - we need to finally initiate slots renewing,
        //or we can stuck with cluster state without one node in opposite case.
        //But now if maxAttempts = 1 or 2 we will do it too often. For each time-outed request.
        //TODO make tracking of successful/unsuccessful operations for node - do renewing only
        //if there were no successful responses from this node last few seconds
        this.connectionHandler.renewSlotCache();
        //no more redirections left, throw original exception, not JedisClusterMaxRedirectionsException, because it's not MOVED situation
        throw jce;
      }

      return runWithRetries(key, attempts - 1, tryRandomNode, asking);
    } catch (JedisRedirectionException jre) {
      // if MOVED redirection occurred,
      //4.如果Redis服务端返回MOVED命令，则刷新本地的槽位Slot信息
      if (jre instanceof JedisMovedDataException) {
        // it rebuilds cluster's slot cache
        // recommended by Redis cluster specification
        //发送slots命令给Redis节点获取槽位信息
        this.connectionHandler.renewSlotCache(connection);
      }

      // release current connection before recursion or renewing
      releaseConnection(connection);
      connection = null;
			//5.如果Redis服务端返回ASK命令
      if (jre instanceof JedisAskDataException) {
        asking = true;
        //6.请求ASK命令携带的目标节点
        askConnection.set(this.connectionHandler.getConnectionFromNode(jre.getTargetNode()));
      } else if (jre instanceof JedisMovedDataException) {
      } else {
        throw new JedisClusterException(jre);
      }
			
      return runWithRetries(key, attempts - 1, false, asking);
    } finally {
      releaseConnection(connection);
    }
  }
...

  private static Object process(final RedisInputStream is) {
    final byte b = is.readByte();
    if (b == PLUS_BYTE) {
      return processStatusCodeReply(is);
    } else if (b == DOLLAR_BYTE) {
      return processBulkReply(is);
    } else if (b == ASTERISK_BYTE) {
      return processMultiBulkReply(is);
    } else if (b == COLON_BYTE) {
      return processInteger(is);
    } else if (b == MINUS_BYTE) {
    	//响应命令以-开头则表示错误
      processError(is);
      return null;
    } else {
      throw new JedisConnectionException("Unknown reply: " + (char) b);
    }
  }
  ...
  private static void processError(final RedisInputStream is) {
    String message = is.readLine();
    //是MOVED命令，则抛出JedisMovedDataException命令
    if (message.startsWith(MOVED_RESPONSE)) {
      String[] movedInfo = parseTargetHostAndSlot(message);
      throw new JedisMovedDataException(message, new HostAndPort(movedInfo[1],
          Integer.valueOf(movedInfo[2])), Integer.valueOf(movedInfo[0]));
    //是ASK命令，则抛出JedisAskDataException命令      
    } else if (message.startsWith(ASK_RESPONSE)) {
      String[] askInfo = parseTargetHostAndSlot(message);
      throw new JedisAskDataException(message, new HostAndPort(askInfo[1],
          Integer.valueOf(askInfo[2])), Integer.valueOf(askInfo[0]));
    } else if (message.startsWith(CLUSTERDOWN_RESPONSE)) {
      throw new JedisClusterException(message);
    } else if (message.startsWith(BUSY_RESPONSE)) {
      throw new JedisBusyException(message);
    } else if (message.startsWith(NOSCRIPT_RESPONSE) ) {
      throw new JedisNoScriptException(message);
    }
    throw new JedisDataException(message);
  }

public void renewClusterSlots(Jedis jedis) {
  //If rediscovering is already in process - no need to start one more same rediscovering, just return
  if (!rediscovering) {
    try {
      w.lock();
      rediscovering = true;

      if (jedis != null) {
        try {
          discoverClusterSlots(jedis);
          return;
        } catch (JedisException e) {
          //try nodes from all pools
        }
      }

			//1.getShuffledNodesPool()获取当前持有的Pool中的Redis节点
      for (JedisPool jp : getShuffledNodesPool()) {
        try {
          jedis = jp.getResource();
          //2.获取池Pool对应的Redis节点信息
          discoverClusterSlots(jedis);
          return;
        } catch (JedisConnectionException e) {
          // try next nodes
        } finally {
          if (jedis != null) {
            jedis.close();
          }
        }
      }
    } finally {
      rediscovering = false;
      w.unlock();
    }
  }
}

private void discoverClusterSlots(Jedis jedis) {
	//3.发送SLOTS命令给节点，获取当前集群的状态信息，里面含槽位信息和对应的节点
  List<Object> slots = jedis.clusterSlots();
  this.slots.clear();

  for (Object slotInfoObj : slots) {
    List<Object> slotInfo = (List<Object>) slotInfoObj;

    if (slotInfo.size() <= MASTER_NODE_INDEX) {
      continue;
    }

    List<Integer> slotNums = getAssignedSlotArray(slotInfo);

    // hostInfos
    List<Object> hostInfos = (List<Object>) slotInfo.get(MASTER_NODE_INDEX);
    if (hostInfos.isEmpty()) {
      continue;
    }

    // at this time, we just use master, discard slave information
    HostAndPort targetNode = generateHostAndPort(hostInfos);
    //4.将槽位slot-节点映射写入Map中
    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();
    }
}