Jedis源码分析:JedisClusterConnectionHandler

JedisClusterConnectionHandler

   JedisClusterConnectionHandler提供了JedisCluster接口获取资源池中Jedis连接对象的一个门面类,JedisClusterConnectionHandler提供了初始化集群,获取资源池中连接对象,刷新资源池等各种方法。JedisClusterConnectionHandler必须依赖于JedisClusterInfoCache类,因为 JedisClusterConnectionHandler本身的方法都是基于JedisClusterInfoCache类实现的,下列所讲解的方法的底层实现会在JedisClusterInfoCache进行讲述。

   JedisClusterConnectionHandler实例化时必须提供以下的参数(见代码1-1),这些参数本身对于JedisClusterConnectionHandler没什么用处,完全就是在该类的构造器中走了一遭,因为JedisClusterConnectionHandler内部维护着JedisClusterInfoCache实例。JedisClusterConnectionHandler将这些参数传递给cache实例后,通过initializeSlotsCache()方法获取集群信息,并调用cache.discoverClusterNodesAndSlots()将获取到的集群信息分别放入nodes缓存和slots缓存中。这两个缓存的存储形式为 Map<String, JedisPool>,Map<Integer, JedisPool> ,虽然都是Map,但是存储的数据内容完全不同,nodes存储着集群所有节点信息,存储形式为<"host:port",JedisPool>,而slots存储的信息为<slot,JedisPool>,并且slots只存储主节点信息。

   从代码1-2中看出,通过遍历节点集合,实例化Jedis对象,再调用cache.discoverClusterNodesAndSlots(jedis)方法将节点放入缓存Map中,获取集群信息的方法也很简单,通过发送'cluster slots'便知道槽和节点的对应情况,再将Redis服务端返回来的输入流解析成Java对象即可。这里注明下,发现集群状况只需要一个可用的Jedis实例即可,但是集群节点总会发生"挂掉"的情况,建议在使用JedisCluster时配上所有的节点信息。

代码1-1

public JedisClusterConnectionHandler(Set<HostAndPort> nodes, final GenericObjectPoolConfig poolConfig, int connectionTimeout, int soTimeout, String password, String clientName, boolean ssl, SSLSocketFactory sslSocketFactory, SSLParameters sslParameters, HostnameVerifier hostnameVerifier, JedisClusterHostAndPortMap portMap) { //实例化集群信息缓存
    this.cache = new JedisClusterInfoCache(poolConfig, connectionTimeout, soTimeout, password, clientName, ssl, sslSocketFactory, sslParameters, hostnameVerifier, portMap); initializeSlotsCache(nodes, poolConfig, connectionTimeout, soTimeout, password, clientName, ssl, sslSocketFactory, sslParameters, hostnameVerifier); }

 

代码1-2 集群发现及初始化缓存池

 private void initializeSlotsCache(Set<HostAndPort> startNodes, GenericObjectPoolConfig poolConfig, int connectionTimeout, int soTimeout, String password, String clientName, boolean ssl, SSLSocketFactory sslSocketFactory, SSLParameters sslParameters, HostnameVerifier hostnameVerifier) { //遍历节点集合,拿出一个可用的节点获取集群信息
      for (HostAndPort hostAndPort : startNodes) { Jedis jedis = null; try { jedis = new Jedis(hostAndPort.getHost(), hostAndPort.getPort(), connectionTimeout, soTimeout, ssl, sslSocketFactory, sslParameters, hostnameVerifier); //添加密码认证
        if (password != null) { jedis.auth(password); } //设置客户端名称
        if (clientName != null) { jedis.clientSetname(clientName); } //TODO 根据节点发现集群信息
 cache.discoverClusterNodesAndSlots(jedis); break; } catch (JedisConnectionException e) { // try next nodes
      } finally { if (jedis != null) { jedis.close(); } } } }

 

集群节点的获取

   JedisClusterConnectionHandler提供了四种获取可用节点的方法,其中 getConnection()和 getConnectionFromSlot(int slot);被定义为抽象方法,交由JedisSlotBasedConnectionHandler子类进行实现。值得注意的是,在获取Jedis连接时,有两个比较重要的点需要说明一下。前面说过JedisClusterInfoCache存放着nodes和slots两个map对象,通常集群模式都会做主从分离,即master节点负责读写操作,slave节点负责读和备份.如果尝试对slave节点进行写操作,那么会发生写入错误。所以提供了getConnectionFromSlot(int slot) 方法来根据槽点值获取对应的主节点信息,如果此时后端集群发生主从切换,该方法会尝试刷新集群状态来保证获取可用的Jedis对象。如果集群频繁的出现切换,那么即使刷新了集群信息也不能保证获取到的Jedis对象一定是有效的,代码中也做了说明"It can't guaranteed to get valid connection because of node assignment(因为节点的分配问题并不能获取到有效连接)",所以代码中采用了重试机制和随机数来尽量保证获取到的是有效的Jedis连接。第二个点是Jedis包依赖于Apache-CommonsPool2来做Jedis的二级缓存,为了保证Jedis是有效的,Jedis包下的JedisFactory实现了PooledObjectFactory相关方法,通过配置DEFAULT_TEST_ON_CREATE,DEFAULT_TEST_ON_BORROW,DEFAULT_TEST_ON_RETURN,DEFAULT_NUM_TESTS_PER_EVICTION_RUN 参数保证在创建阶段,借用阶段,归还阶段时Jedis对象都是可用的。

• Jedis getConnectionFromNode(HostAndPort node)  使用二级缓存,先从JedisClusterInfoCache中获取缓存对象,再去commons-pool包下获取Jedis对象,如果commons-pool2中没有存储该对象,那么先将该对象放入通用缓存池中再进行获取。
• Map<String, JedisPool> getNodes() 直接从JedisClusterInfoCache 的nodes缓存中获取所有JedisPool对象
• Jedis getConnection() 遍历nodes中的节点,如果存在一个jedis对象能够成功执行ping命令,那么返回该Jedis对象。如果遍历完所有的node节点仍然不能找到一个有效的节点,那么抛出JedisNoReachableClusterNodeException异常
• Jedis getConnectionFromSlot(int slot) 根据槽点值获取对应的主节点信息,slots缓存map中存放着槽点值->master node的对应关系。

代码2-1:从缓存池中获取Jedis对象

public Jedis getConnectionFromNode(HostAndPort node) { //使用二级缓存,先从JedisClusterInfoCache中获取缓存对象,再去commons-pool包下获取Jedis对象
    return cache.setupNodeIfNotExist(node).getResource(); } public Map<String, JedisPool> getNodes() { //直接从JedisClusterInfoCache 的nodes缓存中获取所有JedisPool对象
    return cache.getNodes(); } public Jedis getConnection() { // In antirez's redis-rb-cluster implementation, // getRandomConnection always return valid connection (able to // ping-pong) // or exception if all connections are invalid //随机获取
    List<JedisPool> pools = cache.getShuffledNodesPool(); /** * 遍历缓存池中的每个池对象 */
    for (JedisPool pool : pools) { Jedis jedis = null; try { jedis = pool.getResource(); if (jedis == null) { continue; } String result = jedis.ping();//尝试进行ping
        if (result.equalsIgnoreCase("pong")) return jedis;//返回可用连接
 jedis.close(); } catch (JedisException ex) { if (jedis != null) { jedis.close(); } } } //遍历完成后仍然无法获取有效实例
    throw new JedisNoReachableClusterNodeException("No reachable node in cluster"); } public Jedis getConnectionFromSlot(int slot) { //从slots中获取
    JedisPool connectionPool = cache.getSlotPool(slot); if (connectionPool != null) { // It can't guaranteed to get valid connection because of node // assignment
      return connectionPool.getResource(); } else { //刷新集群信息
      renewSlotCache(); //It's abnormal situation for cluster mode, that we have just nothing for slot, try to rediscover state
      connectionPool = cache.getSlotPool(slot); if (connectionPool != null) { //TODO
        return connectionPool.getResource(); } else { //no choice, fallback to new connection to random node //回到随机获取
        return getConnection(); } } }

 

集群信息的重置与关闭

JedisClusterConnectionHandler提供了三个方法来进行集群信息的刷新和关闭操作,详细的说明见《Jedis源码分析:JedisClusterInfoCache》