jinfeng_wang

G-G-S,D-D-U!

BlogJava

管理

400 Posts :: 0 Stories :: 296 Comments :: 0 Trackbacks

JedisCluster 源碼分析

http://m635674608.iteye.com/blog/2297558

Java代碼  
BinaryJedisCluster   
public String set(final byte[] key, final byte[] value) {  
  return new JedisClusterCommand<String>(connectionHandler, maxRedirections) {  
    @Override  
    public String execute(Jedis connection) {  
      return connection.set(key, value);  
    }   
  }.runBinary(key);  
}  

Java代碼  
JedisClusterCommand  
 public T runBinary(byte[] key) {  
    if (key == null) {  
      throw new JedisClusterException("No way to dispatch this command to Redis Cluster.");  
    }  
  
    return runWithRetries(key, this.redirections, false, false);  
  }  
  
 private T runWithRetries(byte[] key, int redirections, boolean tryRandomNode, boolean asking) {  
    if (redirections <= 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 (JedisConnectionException jce) {  
      if (tryRandomNode) {  
        // maybe all connection is down  
        throw jce;  
      }  
  
      // release current connection before recursion  
      releaseConnection(connection);  
      connection = null;  
  
      // retry with random connection  
      return runWithRetries(key, redirections - 1, true, asking);  
    } catch (JedisRedirectionException jre) {  
      // if MOVED redirection occurred,  
      if (jre instanceof JedisMovedDataException) {  
        // it rebuilds cluster's slot cache  
        // recommended by Redis cluster specification  
        this.connectionHandler.renewSlotCache(connection);  
      }  
  
      // release current connection before recursion or renewing  
      releaseConnection(connection);  
      connection = null;  
  
      if (jre instanceof JedisAskDataException) {  
        asking = true;  
      askConnection.set(this.connectionHandler.getConnectionFromNode(jre.getTargetNode()));  
      } else if (jre instanceof JedisMovedDataException) {  
      } else {  
        throw new JedisClusterException(jre);  
      }  
  
      return runWithRetries(key, redirections - 1, false, asking);  
    } finally {  
      releaseConnection(connection);  
    }  
  }  
  
 @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 {  
      return getConnection();  
    }  
  }  

Java代碼  
public abstract class JedisClusterConnectionHandler {  
  protected final JedisClusterInfoCache cache;  
  
  public JedisClusterConnectionHandler(Set<HostAndPort> nodes,  
                                       final GenericObjectPoolConfig poolConfig, int connectionTimeout, int soTimeout) {  
    this.cache = new JedisClusterInfoCache(poolConfig, connectionTimeout, soTimeout);  
    //通過slot 初始化集群信息  
    initializeSlotsCache(nodes, poolConfig);  
  }  
  
  abstract Jedis getConnection();  
  
  abstract Jedis getConnectionFromSlot(int slot);  
  
  public Jedis getConnectionFromNode(HostAndPort node) {  
    cache.setNodeIfNotExist(node);  
    return cache.getNode(JedisClusterInfoCache.getNodeKey(node)).getResource();  
  }  
  
public class JedisClusterInfoCache {  
  private Map<String, JedisPool> nodes = new HashMap<String, JedisPool>();  
  private Map<Integer, JedisPool> slots = new HashMap<Integer, JedisPool>();  
  
  private final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();  
  private final Lock r = rwl.readLock();  
  private final Lock w = rwl.writeLock();  
  private final GenericObjectPoolConfig poolConfig;  
  
  private int connectionTimeout;  
  private int soTimeout;  
  
  private static final int MASTER_NODE_INDEX = 2;  
  
  public JedisClusterInfoCache(final GenericObjectPoolConfig poolConfig, int timeout) {  
    this(poolConfig, timeout, timeout);  
  }  
  
  public JedisClusterInfoCache(final GenericObjectPoolConfig poolConfig,  
      final int connectionTimeout, final int soTimeout) {  
    this.poolConfig = poolConfig;  
    this.connectionTimeout = connectionTimeout;  
    this.soTimeout = soTimeout;  
  }  
  
  public void discoverClusterNodesAndSlots(Jedis jedis) {  
    w.lock();  
  
    try {  
      this.nodes.clear();  
      this.slots.clear();  
  
      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);  
          setNodeIfNotExist(targetNode);  
          if (i == MASTER_NODE_INDEX) {  
            assignSlotsToNode(slotNums, targetNode);  
          }  
        }  
      }  
    } finally {  
      w.unlock();  
    }  
  }  
  //初始化集群信息   
  public void discoverClusterSlots(Jedis jedis) {  
    w.lock();  
  
    try {  
      this.slots.clear();  
      //通過 slots 命令獲取集群信息  
      List<Object> slots = jedis.clusterSlots();  
  
      for (Object slotInfoObj : slots) {  
        List<Object> slotInfo = (List<Object>) slotInfoObj;  
  
        if (slotInfo.size() <= 2) {  
          continue;  
        }  
  
        List<Integer> slotNums = getAssignedSlotArray(slotInfo);  
  
        // hostInfos  
        List<Object> hostInfos = (List<Object>) slotInfo.get(2);  
        if (hostInfos.size() <= 0) {  
          continue;  
        }  
  
        // at this time, we just use master, discard slave information  
        HostAndPort targetNode = generateHostAndPort(hostInfos);  
  
        setNodeIfNotExist(targetNode);  
        assignSlotsToNode(slotNums, targetNode);  
      }  
    } finally {  
      w.unlock();  
    }  
  }  
  
  private HostAndPort generateHostAndPort(List<Object> hostInfos) {  
    return new HostAndPort(SafeEncoder.encode((byte[]) hostInfos.get(0)),  
        ((Long) hostInfos.get(1)).intValue());  
  }  
  
  public void setNodeIfNotExist(HostAndPort node) {  
    w.lock();  
    try {  
      String nodeKey = getNodeKey(node);  
      if (nodes.containsKey(nodeKey)) return;  
  
      JedisPool nodePool = new JedisPool(poolConfig, node.getHost(), node.getPort(),  
              connectionTimeout, soTimeout, null, 0, null);  
      nodes.put(nodeKey, nodePool);  
    } finally {  
      w.unlock();  
    }  
  }  
  
  public void assignSlotToNode(int slot, HostAndPort targetNode) {  
    w.lock();  
    try {  
      JedisPool targetPool = nodes.get(getNodeKey(targetNode));  
  
      if (targetPool == null) {  
        setNodeIfNotExist(targetNode);  
        targetPool = nodes.get(getNodeKey(targetNode));  
      }  
      slots.put(slot, targetPool);  
    } finally {  
      w.unlock();  
    }  
  }  
  
  public void assignSlotsToNode(List<Integer> targetSlots, HostAndPort targetNode) {  
    w.lock();  
    try {  
      JedisPool targetPool = nodes.get(getNodeKey(targetNode));  
  
      if (targetPool == null) {  
        setNodeIfNotExist(targetNode);  
        targetPool = nodes.get(getNodeKey(targetNode));  
      }  
  
      for (Integer slot : targetSlots) {  
        slots.put(slot, targetPool);  
      }  
    } finally {  
      w.unlock();  
    }  
  }  
  
  public JedisPool getNode(String nodeKey) {  
    r.lock();  
    try {  
      return nodes.get(nodeKey);  
    } finally {  
      r.unlock();  
    }  
  }  
  
  public JedisPool getSlotPool(int slot) {  
    r.lock();  
    try {  
      return slots.get(slot);  
    } finally {  
      r.unlock();  
    }  
  }  
  
  public Map<String, JedisPool> getNodes() {  
    r.lock();  
    try {  
      return new HashMap<String, JedisPool>(nodes);  
    } finally {  
      r.unlock();  
    }  
  }  
  
  public static String getNodeKey(HostAndPort hnp) {  
    return hnp.getHost() + ":" + hnp.getPort();  
  }  
  
  public static String getNodeKey(Client client) {  
    return client.getHost() + ":" + client.getPort();  
  }  
  
  public static String getNodeKey(Jedis jedis) {  
    return getNodeKey(jedis.getClient());  
  }  
  
  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;  
  }  
  
}  
  
    
  public Map<String, JedisPool> getNodes() {  
    return cache.getNodes();  
  }  
  
  private void initializeSlotsCache(Set<HostAndPort> startNodes, GenericObjectPoolConfig poolConfig) {  
    for (HostAndPort hostAndPort : startNodes) {  
      Jedis jedis = new Jedis(hostAndPort.getHost(), hostAndPort.getPort());  
      try {  
        cache.discoverClusterNodesAndSlots(jedis);  
        break;  
      } catch (JedisConnectionException e) {  
        // try next nodes  
      } finally {  
        if (jedis != null) {  
          jedis.close();  
        }  
      }  
    }  
  
    for (HostAndPort node : startNodes) {  
      cache.setNodeIfNotExist(node);  
    }  
  }  
  
  public void renewSlotCache() {  
    for (JedisPool jp : getShuffledNodesPool()) {  
      Jedis jedis = null;  
      try {  
        jedis = jp.getResource();  
        cache.discoverClusterSlots(jedis);  
        break;  
      } catch (JedisConnectionException e) {  
        // try next nodes  
      } finally {  
        if (jedis != null) {  
          jedis.close();  
        }  
      }  
    }  
  }  
  
  public void renewSlotCache(Jedis jedis) {  
    try {  
      cache.discoverClusterSlots(jedis);  
    } catch (JedisConnectionException e) {  
      renewSlotCache();  
    }  
  }  
  
  protected List<JedisPool> getShuffledNodesPool() {  
    List<JedisPool> pools = new ArrayList<JedisPool>();  
    pools.addAll(cache.getNodes().values());  
    Collections.shuffle(pools);  
    return pools;  
  }  
}  

總結：

1.JedisCluster 會初始化一個連接獲取集群信息通過 solts 命令。（JedisClusterInfoCache 構造方法初始化）

2.get ,set 的時候。會通過key JedisClusterCRC16.getSlot(key) 定位到solt

3. 然后根據solt獲取 jedis

public Jedis getConnectionFromSlot(int slot) {
JedisPool connectionPool = cache.getSlotPool(slot);

4.執行操作

讀操作：主庫，從庫都會讀

寫操作：主庫寫

posted on 2016-12-20 16:29 jinfeng_wang 閱讀(821) 評論(0) 編輯收藏所屬分類: 2016-REDIS

新用戶注冊刷新評論列表


只有注冊用戶登錄后才能發表評論。




網站導航: 博客園 IT新聞 Chat2DB C++博客博問管理
相關文章: 緩存系列文章--7.無底洞問題(multiget hole) 緩存系列文章--6.緩存雪崩問題緩存系列文章--5.緩存穿透問題緩存系列文章--4.緩存的粒度控制緩存系列文章--2.是否真的需要緩存？緩存系列文章--3.緩存常用更新策略對比(一致性)。緩存系列文章--1.緩存的一些基本常識 JedisCluster 源碼分析 redis cluster使用經驗深入淺出Redis（三）高級特性：管道

jinfeng_wang

公告

常用鏈接

留言簿(40)

隨筆分類(592)

隨筆檔案(400)

Domestic

Foreign

搜索

積分與排名

最新評論

閱讀排行榜

評論排行榜