Jetbrains全系列IDE稳定放心使用

Pre

Redis进阶-Redis集群原理剖析及gossip协议初探 集群原理部分 简单的提了下Jest是如何实现Redis Cluster 的 ，这里我们再来梳理一下

Code

import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.JedisPoolConfig;

import java.io.IOException;
import java.util.HashSet;
import java.util.Set;


public class JedisClusterDemo { 
   
    public static void main(String[] args) throws IOException { 
   

        JedisPoolConfig config = new JedisPoolConfig();
        config.setMaxTotal(20);
        config.setMaxIdle(10);
        config.setMinIdle(5);

        Set<HostAndPort> jedisClusterNode = new HashSet<HostAndPort>();
        jedisClusterNode.add(new HostAndPort("192.168.18.131", 8001));
        jedisClusterNode.add(new HostAndPort("192.168.18.131", 8004));
        jedisClusterNode.add(new HostAndPort("192.168.18.132", 8002));
        jedisClusterNode.add(new HostAndPort("192.168.18.132", 8005));
        jedisClusterNode.add(new HostAndPort("192.168.18.133", 8003));
        jedisClusterNode.add(new HostAndPort("192.168.18.133", 8006));

        JedisCluster jedisCluster = null;
        try { 
   
            //connectionTimeout：指的是连接一个url的连接等待时间
            //soTimeout：指的是连接上一个url，获取response的返回等待时间
            jedisCluster = new JedisCluster(jedisClusterNode, 6000, 5000, 10, "artisan", config);
            System.out.println(jedisCluster.set("clusterArtisan", "artisanValue"));
            System.out.println(jedisCluster.get("clusterArtisan"));
        } catch (Exception e) { 
   
            e.printStackTrace();
        } finally { 
   
            if (jedisCluster != null)
                jedisCluster.close();
        }
    }
}

这里是个简单的demo， 生产中用的话，需要确保jedisCluster是单例的，并且无需手工调用close，不然的话 这个连接池就关闭了，你就无法获取到连接了。

初始化

当 Redis Cluster 的客户端来连接集群时，它也会得到一份集群的槽位配置信息并将其缓存在客户端本地。这样当客户端要查找某个 key 时，可以直接定位到目标节点。

我们来看下jedis的实现

 jedisCluster = new JedisCluster(jedisClusterNode, 6000, 5000, 10, "artisan", config);

跟下源码

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

继续 cache.discoverClusterNodesAndSlots(jedis); cache为 JedisClusterInfoCache 对象。

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

槽计算

set --------> run  ----> runWithRetries ----> connectionHandler.getConnectionFromSlot(JedisClusterCRC16.getSlot(key))

CRC16算法，计算key对应的slot connectionHandler.getConnectionFromSlot(JedisClusterCRC16.getSlot(key))

  public static int getSlot(byte[] key) { 
   
    int s = -1;
    int e = -1;
    boolean sFound = false;
    for (int i = 0; i < key.length; i++) { 
   
      if (key[i] == '{' && !sFound) { 
   
        s = i;
        sFound = true;
      }
      if (key[i] == '}' && sFound) { 
   
        e = i;
        break;
      }
    }
    if (s > -1 && e > -1 && e != s + 1) { 
   
      return getCRC16(key, s + 1, e) & (16384 - 1);
    }
    return getCRC16(key) & (16384 - 1);
  }

无需手工调用close方法

进入到set方法中看下源码

jedisCluster.set("clusterArtisan", "artisanValue")

如下：

  @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方法

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

    Jedis connection = null;
    try { 
   

      if (asking) { 
   
       ......
      } else { 
   
        if (tryRandomNode) { 
   
          connection = connectionHandler.getConnection();
        } else { 
   
          connection = connectionHandler.getConnectionFromSlot(JedisClusterCRC16.getSlot(key));
        }
      }

      return execute(connection);
    } finally { 
   
      releaseConnection(connection);
    }
  }

关注点

• CRC16算法，计算key对应的slot connectionHandler.getConnectionFromSlot(JedisClusterCRC16.getSlot(key))

  public static int getSlot(byte[] key) { 
   
    int s = -1;
    int e = -1;
    boolean sFound = false;
    for (int i = 0; i < key.length; i++) { 
   
      if (key[i] == '{' && !sFound) { 
   
        s = i;
        sFound = true;
      }
      if (key[i] == '}' && sFound) { 
   
        e = i;
        break;
      }
    }
    if (s > -1 && e > -1 && e != s + 1) { 
   
      return getCRC16(key, s + 1, e) & (16384 - 1);
    }
    return getCRC16(key) & (16384 - 1);
  }

• getConnectionFromSlot 通过 JedisPool 获取连接

关注下 JedisCluster是如何获取连接的 getConnectionFromSlot 方法

  @Override
  public Jedis getConnectionFromSlot(int slot) { 
   
    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) { 
   
        return connectionPool.getResource();
      } else { 
   
        //no choice, fallback to new connection to random node
        return getConnection();
      }
    }
  }

本质上还是通过 JedisPool 来获取一个getResource ，跟我们使用Sentinel 啊 单节点获取方法是一样的

• finally 语句中的 releaseConnection(connection); ,自动释放连接

看下该方法

  private void releaseConnection(Jedis connection) { 
   
    if (connection != null) { 
   
      connection.close();
    }
  }

说白了，JedisCluster set后会自动释放连接，调用的是jedis 的close方法，所以我们无需手工关闭，否则你这个jedis的连接池就挂逼了…