Jedis cluster集群初始化源码剖析

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:p="http://www.springframework.org/schema/p"
	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.2.xsd">

	<!-- 连接池配置. -->
	<bean id="jedisPoolConfig" class="redis.clients.jedis.JedisPoolConfig">
		<!-- 连接池中最大连接数。高版本：maxTotal，低版本：maxActive -->
		<property name="maxTotal" value="8" />
		<!-- 连接池中最大空闲的连接数. -->
		<property name="maxIdle" value="4" />
		<!-- 连接池中最少空闲的连接数. -->
		<property name="minIdle" value="1" />
		<!-- 当连接池资源耗尽时，调用者最大阻塞的时间，超时将跑出异常。单位，毫秒数;默认为-1.表示永不超时。高版本：maxWaitMillis，低版本：maxWait -->
		<property name="maxWaitMillis" value="5000" />
		<!-- 连接空闲的最小时间，达到此值后空闲连接将可能会被移除。负值(-1)表示不移除. -->
		<property name="minEvictableIdleTimeMillis" value="300000" />
		<!-- 对于“空闲链接”检测线程而言，每次检测的链接资源的个数。默认为3 -->
		<property name="numTestsPerEvictionRun" value="3" />
		<!-- “空闲链接”检测线程，检测的周期，毫秒数。如果为负值，表示不运行“检测线程”。默认为-1. -->
		<property name="timeBetweenEvictionRunsMillis" value="60000" />
		<!-- testOnBorrow:向调用者输出“链接”资源时，是否检测是有有效，如果无效则从连接池中移除，并尝试获取继续获取。默认为false。建议保持默认值. -->
		<!-- testOnReturn:向连接池“归还”链接时，是否检测“链接”对象的有效性。默认为false。建议保持默认值. -->
		<!-- testWhileIdle:向调用者输出“链接”对象时，是否检测它的空闲超时；默认为false。如果“链接”空闲超时，将会被移除。建议保持默认值. -->
		<!-- whenExhaustedAction:当“连接池”中active数量达到阀值时，即“链接”资源耗尽时，连接池需要采取的手段, 默认为1(0:抛出异常。1:阻塞，直到有可用链接资源。2:强制创建新的链接资源) -->
	</bean>

	<bean id="n1" class="org.springframework.data.redis.connection.RedisNode">
		<constructor-arg value="127.0.0.1" />
		<constructor-arg value="6379" type="int" />
	</bean>
	<bean id="n2" class="org.springframework.data.redis.connection.RedisNode">
		<constructor-arg value="127.0.0.1" />
		<constructor-arg value="6380" type="int" />
	</bean>
	<bean id="n3" class="org.springframework.data.redis.connection.RedisNode">
		<constructor-arg value="127.0.0.1" />
		<constructor-arg value="6381" type="int" />
	</bean>
	
	<bean id="redisClusterConfiguration"
		class="org.springframework.data.redis.connection.RedisClusterConfiguration">
		<property name="clusterNodes">
			<set>
				<ref bean="n1" />
				<ref bean="n2" />
				<ref bean="n3" />
			</set>
		</property>
		<property name="maxRedirects" value="5" />
	</bean>

	<bean id="jedisConnectionFactory"
		class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory">
		<constructor-arg ref="redisClusterConfiguration" />
		<constructor-arg ref="jedisPoolConfig" />
	</bean>

	<!-- Spring提供的访问Redis类. -->
	<bean id="redisTemplate" class="org.springframework.data.redis.core.RedisTemplate">
		<property name="connectionFactory" ref="jedisConnectionFactory" />
		<property name="KeySerializer">
			<bean
				class="org.springframework.data.redis.serializer.StringRedisSerializer" />
		</property>
		<property name="ValueSerializer">
			<bean
				class="org.springframework.data.redis.serializer.StringRedisSerializer" />
		</property>
		<property name="hashKeySerializer">
			<bean
				class="org.springframework.data.redis.serializer.StringRedisSerializer" />
		</property>
		<property name="hashValueSerializer">
			<bean
				class="org.springframework.data.redis.serializer.StringRedisSerializer" />
		</property>
	</bean>

	<!-- Redis配置结束 -->
</beans>

<bean id="jedisConnectionFactory"
class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory">
		<constructor-arg ref="redisClusterConfiguration" />
		<constructor-arg ref="jedisPoolConfig" />
</bean>

public class JedisConnectionFactory implements InitializingBean, DisposableBean, RedisConnectionFactory {
    
    private JedisPoolConfig poolConfig = new JedisPoolConfig();
    private RedisClusterConfiguration clusterConfig;

	public JedisConnectionFactory(RedisClusterConfiguration clusterConfig, JedisPoolConfig poolConfig) {
		this.clusterConfig = clusterConfig;
		this.poolConfig = poolConfig;
	}
	
	/*
	 * (non-Javadoc)
	 * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
	 */
	public void afterPropertiesSet() {
		if (shardInfo == null) {
			shardInfo = new JedisShardInfo(hostName, port);

			if (StringUtils.hasLength(password)) {
				shardInfo.setPassword(password);
			}

			if (timeout > 0) {
				setTimeoutOn(shardInfo, timeout);
			}
		}

		if (usePool && clusterConfig == null) {
			this.pool = createPool();
		}
      
      //如果集群配置信息不为空，则创建JedisCluster对象
		if (clusterConfig != null) {
			this.cluster = createCluster();
		}
	}
}

private JedisCluster createCluster() {

		JedisCluster cluster = createCluster(this.clusterConfig, this.poolConfig);
		this.clusterCommandExecutor = new ClusterCommandExecutor(
				new JedisClusterConnection.JedisClusterTopologyProvider(cluster),
				new JedisClusterConnection.JedisClusterNodeResourceProvider(cluster), EXCEPTION_TRANSLATION);
		return cluster;
	}

	/**
	 * Creates {@link JedisCluster} for given {@link RedisClusterConfiguration} and {@link GenericObjectPoolConfig}.
	 * 
	 * @param clusterConfig must not be {@literal null}.
	 * @param poolConfig can be {@literal null}.
	 * @return
	 * @since 1.7
	 */
	protected JedisCluster createCluster(RedisClusterConfiguration clusterConfig, GenericObjectPoolConfig poolConfig) {

		Assert.notNull(clusterConfig, "Cluster configuration must not be null!");

		Set<HostAndPort> hostAndPort = new HashSet<HostAndPort>();
		for (RedisNode node : clusterConfig.getClusterNodes()) {
			hostAndPort.add(new HostAndPort(node.getHost(), node.getPort()));
		}

		int redirects = clusterConfig.getMaxRedirects() != null ? clusterConfig.getMaxRedirects().intValue() : 5;

		return StringUtils.hasText(getPassword())
				? new JedisCluster(hostAndPort, timeout, timeout, redirects, password, poolConfig)
				: new JedisCluster(hostAndPort, timeout, redirects, poolConfig);
	}

public JedisCluster(Set<HostAndPort> jedisClusterNode, int timeout, int maxAttempts,
      final GenericObjectPoolConfig poolConfig) {
    super(jedisClusterNode, timeout, maxAttempts, poolConfig);
  }

 public BinaryJedisCluster(Set<HostAndPort> jedisClusterNode, int timeout, int maxAttempts,
      final GenericObjectPoolConfig poolConfig) {
    this.connectionHandler = new JedisSlotBasedConnectionHandler(jedisClusterNode, poolConfig,
        timeout);
    this.maxAttempts = maxAttempts;
  }

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

public Jedis getConnectionFromNode(HostAndPort node) {
    return cache.setupNodeIfNotExist(node).getResource();
  }
  
  public Map<String, JedisPool> getNodes() {
    return cache.getNodes();
  }

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

  public void renewSlotCache() {
    cache.renewClusterSlots(null);
  }

  public void renewSlotCache(Jedis jedis) {
    cache.renewClusterSlots(jedis);
  }

  @Override
  public void close() {
    cache.reset();
  }

127.0.0.1:6379> cluster slots
1) 1) (integer) 12288
   2) (integer) 16383
   3) 1) "127.0.0.1"
      2) (integer) 6382
      3) "65aea5fc4485bc7c0c3c4425fb3f500c562ee243"
   4) 1) "127.0.0.1"
      2) (integer) 6386
      3) "4061e306b094e707b6f4a7c8cd8e82bd61155060"
2) 1) (integer) 4096
   2) (integer) 8191
   3) 1) "127.0.0.1"
      2) (integer) 6380
      3) "c6e1b3691b968b009357dcac3349afbcd557fd8c"
   4) 1) "127.0.0.1"
      2) (integer) 6384
      3) "f915c7e6812a7d8fbe637c782ad261cd453022b2"
3) 1) (integer) 0
   2) (integer) 4095
   3) 1) "127.0.0.1"
      2) (integer) 6379
      3) "91bb43a956a04a9812e4d6950efebbb2e0f646fd"
   4) 1) "127.0.0.1"
      2) (integer) 6383
      3) "c1d9d907f6905dd826dad774d127b75484ef8ea8"
4) 1) (integer) 8192
   2) (integer) 12287
   3) 1) "127.0.0.1"
      2) (integer) 6381
      3) "745936c1192bc1b136fd1f5df842bc1dd517ef36"
   4) 1) "127.0.0.1"
      2) (integer) 6385
      3) "1c07bd8406156122eb4855d2e8b36e785e7901c7"

public class JedisClusterInfoCache {
	// 保存Redis集群节点和节点连接池信息：key为节点地址、value为连接池
	private final Map<String, JedisPool> nodes = new HashMap<String, JedisPool>();
	// 保存Redis集群节点槽和槽所在的主节点连接池信息：key为节点槽、value为连接池
	private final Map<Integer, JedisPool> slots = new HashMap<Integer, JedisPool>();

	// 使用读写锁保证nodes和slots两个map的写安全
	private final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
	private final Lock r = rwl.readLock();
	private final Lock w = rwl.writeLock();
	// 重建缓存的标识变量，false为未进行，true为正在进行
	private volatile boolean rediscovering;
	private final GenericObjectPoolConfig poolConfig;

	private int connectionTimeout;
	private int soTimeout;
	private String password;

	// 主节点索引位置标识，遍历cluster slots结果时使用
	private static final int MASTER_NODE_INDEX = 2;

	public JedisClusterInfoCache(final GenericObjectPoolConfig poolConfig, int timeout) {
		this(poolConfig, timeout, timeout, null);
	}

	public JedisClusterInfoCache(final GenericObjectPoolConfig poolConfig, final int connectionTimeout, final int soTimeout, final String password) {
		this.poolConfig = poolConfig;
		this.connectionTimeout = connectionTimeout;
		this.soTimeout = soTimeout;
		this.password = password;
	}

	/**
	 * 在jedis封装的redis集群节点信息上发送cluster slots命令，获取所有集群节点信息和槽信息
	 * 
	 * @param jedis
	 */
	public void discoverClusterNodesAndSlots(Jedis jedis) {
		w.lock();// 由当前线程获得写锁，在当前线程操作未结束之前，其他线程只能等待

		try {
			reset();// 重置nodes、slots两个Map，释放JedisPool连接池资源
			List<Object> slots = jedis.clusterSlots();// 在redis集群节点信息上发送cluster slots命令，获取所有集群节点信息和槽信息

			// 遍历slots集合，保存Redis集群节点和节点连接池信息到nodes Map中，保存Redis集群节点槽和槽所在的主节点连接池信息到slots Map中
			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();
				// 遍历slots集合元素中的主从节点信息，保存Redis集群节点和节点连接池信息到nodes Map中，保存Redis集群节点槽和槽所在的主节点连接池信息到slots Map中
				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);
					// 保存Redis集群节点和节点连接池信息到nodes Map中
					setupNodeIfNotExist(targetNode);
					// 如果当前遍历的是主节点信息，则保存Redis集群节点槽和槽所在的主节点连接池信息到slots Map中
					if (i == MASTER_NODE_INDEX) {
						assignSlotsToNode(slotNums, targetNode);
					}
				}
			}
		} finally {
			w.unlock();// 释放写锁，使其他线程使用
		}
	}

	/**
	 * 重建Cluster集群节点和Slot槽缓存
	 * 
	 * @param jedis
	 */
	public void renewClusterSlots(Jedis jedis) {
		// 如果重建操作未进行，则开始重建缓存操作
		if (!rediscovering) {
			try {
				w.lock();
				rediscovering = true;// 设重建缓存标识变量的值为true，表示重建操作正在进行

				// 如果封装redis连接信息的jedis对象不为空，则使用该节点进行重建缓存操作并返回
				if (jedis != null) {
					try {
						discoverClusterSlots(jedis);
						return;
					} catch (JedisException e) {
						// try nodes from all pools
					}
				}

				// 如果封装redis连接信息的jedis对象为空，则打乱nodes Map中保存的jedis连接池信息，遍历连接池中的节点进行重建缓存操作并返回
				for (JedisPool jp : getShuffledNodesPool()) {
					try {
						jedis = jp.getResource();
						discoverClusterSlots(jedis);
						return;
					} catch (JedisConnectionException e) {
						// try next nodes
					} finally {
						if (jedis != null) {
							jedis.close();
						}
					}
				}
			} finally {
				rediscovering = false;// 设重建缓存标识变量的值为false，表示重建操作未进行
				w.unlock();
			}
		}
	}

	/**
	 * 逻辑类似discoverClusterNodesAndSlots方法
	 * 
	 * @param jedis
	 */
	private void discoverClusterSlots(Jedis jedis) {
		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);
			assignSlotsToNode(slotNums, targetNode);
		}
	}

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

	/**
	 * 保存Redis集群节点和节点连接池信息到nodes Map中
	 * 
	 * @param node
	 * @return
	 */
	public JedisPool setupNodeIfNotExist(HostAndPort node) {
		w.lock();
		try {
			// 获取节点key，形式为"服务器地址:端口"
			String nodeKey = getNodeKey(node);
			// 如果节点已存在nodes Map中，则直接返回
			JedisPool existingPool = nodes.get(nodeKey);
			if (existingPool != null)
				return existingPool;

			// 创建节点相应的JedisPool连接池对象，并保存到nodes Map中，然后返回JedisPool连接池对象
			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();
		}
	}

	/**
	 * 遍历槽集合，保存Redis集群节点槽和槽所在的主节点连接池信息到slots Map中
	 * 
	 * @param targetSlots
	 * @param 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();
		}
	}

	/**
	 * 根据节点key获取JedisPool连接池对象
	 * 
	 * @param nodeKey
	 * @return
	 */
	public JedisPool getNode(String nodeKey) {
		r.lock();
		try {
			return nodes.get(nodeKey);
		} finally {
			r.unlock();
		}
	}

	/**
	 * 根据slot槽值获取JedisPool连接池对象
	 * 
	 * @param slot
	 * @return
	 */
	public JedisPool getSlotPool(int slot) {
		r.lock();
		try {
			return slots.get(slot);
		} finally {
			r.unlock();
		}
	}

	/**
	 * 获取节点信息和节点对象对应的连接池信息
	 * 
	 * @return
	 */
	public Map<String, JedisPool> getNodes() {
		r.lock();
		try {
			return new HashMap<String, JedisPool>(nodes);
		} finally {
			r.unlock();
		}
	}

	/**
	 * 获取nodes Map打乱顺序后的Redis集群节点连接池信息
	 * 
	 * @return
	 */
	public List<JedisPool> getShuffledNodesPool() {
		r.lock();
		try {
			List<JedisPool> pools = new ArrayList<JedisPool>(nodes.values());
			Collections.shuffle(pools);
			return pools;
		} finally {
			r.unlock();
		}
	}

	/**
	 * 清空集群节点集合和槽集合，释放JedisPool资源
	 */
	public void reset() {
		w.lock();
		try {
			for (JedisPool pool : nodes.values()) {
				try {
					if (pool != null) {
						pool.destroy();
					}
				} catch (Exception e) {
					// pass
				}
			}
			nodes.clear();
			slots.clear();
		} finally {
			w.unlock();
		}
	}

	public static String getNodeKey(HostAndPort hnp) {
		return hnp.getHost() + ":" + hnp.getPort();
	}

	/**
	 * 遍历槽区间，获取槽节点集合
	 * 
	 * @param slotInfo
	 * @return
	 */
	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;
	}
}