深入剖析Redis客户端Jedis的特性和原理

一、开篇

二、Jedis访问模式对比

三、Jedis的基础用法

// Jedis单机模式的访问public void main(String[] args) {    // 创建Jedis对象    jedis = new Jedis("localhost", 6379);    // 执行hmget操作    jedis.hmget("foobar", "foo");    // 关闭Jedis对象    jedis.close();} // Jedis分片模式的访问public void main(String[] args) {    HostAndPort redis1 = HostAndPortUtil.getRedisServers().get();    HostAndPort redis2 = HostAndPortUtil.getRedisServers().get(1);    List<JedisShardInfo> shards = new ArrayList<JedisShardInfo>(2);    JedisShardInfo shard1 = new JedisShardInfo(redis1);    JedisShardInfo shard2 = new JedisShardInfo(redis2);    // 创建ShardedJedis对象    ShardedJedis shardedJedis = new ShardedJedis(shards);    // 通过ShardedJedis对象执行set操作    shardedJedis.set("a", "bar");} // Jedis集群模式的访问public void main(String[] args) {    // 构建redis的集群池    Set<HostAndPort> nodes = new HashSet<>();    nodes.add(new HostAndPort("127.0.0.1", 7001));    nodes.add(new HostAndPort("127.0.0.1", 7002));    nodes.add(new HostAndPort("127.0.0.1", 7003));     // 创建JedisCluster    JedisCluster cluster = new JedisCluster(nodes);     // 执行JedisCluster对象中的方法    cluster.set("cluster-test", "my jedis cluster test");    String result = cluster.get("cluster-test");}

四、Jedis单机模式的访问

public class Jedis extends BinaryJedis implements JedisCommands, MultiKeyCommands,    AdvancedJedisCommands, ScriptingCommands, BasicCommands, ClusterCommands, SentinelCommands, ModuleCommands {   protected JedisPoolAbstract dataSource = null;   public Jedis(final String host, final int port) {    // 创建父类BinaryJedis对象    super(host, port);  }} public class BinaryJedis implements BasicCommands, BinaryJedisCommands, MultiKeyBinaryCommands,    AdvancedBinaryJedisCommands, BinaryScriptingCommands, Closeable {   // 访问redis的Client对象  protected Client client = null;   public BinaryJedis(final String host, final int port) {    // 创建Client对象访问redis    client = new Client(host, port);  }}

public class Client extends BinaryClient implements Commands {  public Client(final String host, final int port) {    super(host, port);  }} public class BinaryClient extends Connection {  // 存储和Redis连接的相关信息  private boolean isInMulti;  private String user;  private String password;  private int db;  private boolean isInWatch;   public BinaryClient(final String host, final int port) {    super(host, port);  }} public class Connection implements Closeable {  // 管理和Redis连接的socket信息及对应的输入输出流  private JedisSocketFactory jedisSocketFactory;  private Socket socket;  private RedisOutputStream outputStream;  private RedisInputStream inputStream;  private int infiniteSoTimeout = ;  private boolean broken = false;   public Connection(final String host, final int port, final boolean ssl,      SSLSocketFactory sslSocketFactory, SSLParameters sslParameters,      HostnameVerifier hostnameVerifier) {    // 构建DefaultJedisSocketFactory来创建和Redis连接的Socket对象    this(new DefaultJedisSocketFactory(host, port, Protocol.DEFAULT_TIMEOUT,        Protocol.DEFAULT_TIMEOUT, ssl, sslSocketFactory, sslParameters, hostnameVerifier));  }}

public class Jedis extends BinaryJedis implements JedisCommands, MultiKeyCommands,    AdvancedJedisCommands, ScriptingCommands, BasicCommands, ClusterCommands, SentinelCommands, ModuleCommands {  @Override  public String set(final String key, final String value) {    checkIsInMultiOrPipeline();    // client执行set操作    client.set(key, value);    return client.getStatusCodeReply();  }} public class Client extends BinaryClient implements Commands {  @Override  public void set(final String key, final String value) {    // 执行set命令    set(SafeEncoder.encode(key), SafeEncoder.encode(value));  }} public class BinaryClient extends Connection {  public void set(final byte[] key, final byte[] value) {    // 发送set指令    sendCommand(SET, key, value);  }} public class Connection implements Closeable {  public void sendCommand(final ProtocolCommand cmd, final byte[]... args) {    try {      // socket连接redis      connect();      // 按照redis的协议发送命令      Protocol.sendCommand(outputStream, cmd, args);    } catch (JedisConnectionException ex) {    }  }}

五、Jedis分片模式的访问

public class Sharded<R, S extends ShardInfo<R>> {   public static final int DEFAULT_WEIGHT = 1;  // 保存虚拟节点和redis的node节点的映射关系  private TreeMap<Long, S> nodes;  // hash算法  private final Hashing algo;  // 保存redis节点和访问该节点的Jedis的连接信息  private final Map<ShardInfo<R>, R> resources = new LinkedHashMap<>();   public Sharded(List<S> shards, Hashing algo) {    this.algo = algo;    initialize(shards);  }   private void initialize(List<S> shards) {    nodes = new TreeMap<>();    // 遍历每个redis的节点并设置hash值到节点的映射关系    for (int i = ; i != shards.size(); ++i) {      final S shardInfo = shards.get(i);      // 根据权重映射成未160个虚拟节点      int N =  160 * shardInfo.getWeight();      if (shardInfo.getName() == null) for (int n = ; n < N; n++) {        // 构建hash值和节点映射关系        nodes.put(this.algo.hash("SHARD-" + i + "-NODE-" + n), shardInfo);      }      else for (int n = ; n < N; n++) {        nodes.put(this.algo.hash(shardInfo.getName() + "*" + n), shardInfo);      }      // 保存每个节点的访问对象      resources.put(shardInfo, shardInfo.createResource());    }  }}

public class Sharded<R, S extends ShardInfo<R>> {   public static final int DEFAULT_WEIGHT = 1;  private TreeMap<Long, S> nodes;  private final Hashing algo;  // 保存redis节点和访问该节点的Jedis的连接信息  private final Map<ShardInfo<R>, R> resources = new LinkedHashMap<>();   public R getShard(String key) {    // 根据redis节点找到对应的访问对象Jedis    return resources.get(getShardInfo(key));  }   public S getShardInfo(String key) {    return getShardInfo(SafeEncoder.encode(getKeyTag(key)));  }   public S getShardInfo(byte[] key) {    // 针对访问的key生成对应的hash值    // 根据hash值找到对应的redis节点    SortedMap<Long, S> tail = nodes.tailMap(algo.hash(key));    if (tail.isEmpty()) {      return nodes.get(nodes.firstKey());    }    return tail.get(tail.firstKey());  }}

六、Jedis集群模式的访问

127.0.0.1:30001> cluster slots1) 1) (integer) 0 // 开始槽位   2) (integer) 5460 // 结束槽位   3) 1) "127.0.0.1" // master节点的host      2) (integer) 30001 // master节点的port      3) "09dbe9720cda62f7865eabc5fd8857c5d2678366" // 节点的编码   4) 1) "127.0.0.1" // slave节点的host      2) (integer) 30004 // slave节点的port      3) "821d8ca00d7ccf931ed3ffc7e3db0599d2271abf" // 节点的编码2) 1) (integer) 5461   2) (integer) 10922   3) 1) "127.0.0.1"      2) (integer) 30002      3) "c9d93d9f2c0c524ff34cc11838c2003d8c29e013"   4) 1) "127.0.0.1"      2) (integer) 30005      3) "faadb3eb99009de4ab72ad6b6ed87634c7ee410f"3) 1) (integer) 10923   2) (integer) 16383   3) 1) "127.0.0.1"      2) (integer) 30003      3) "044ec91f325b7595e76dbcb18cc688b6a5b434a1"   4) 1) "127.0.0.1"      2) (integer) 30006      3) "58e6e48d41228013e5d9c1c37c5060693925e97e"

public class JedisCluster extends BinaryJedisCluster implements JedisClusterCommands,    MultiKeyJedisClusterCommands, JedisClusterScriptingCommands {  public JedisCluster(Set<HostAndPort> jedisClusterNode, int connectionTimeout, int soTimeout,      int maxAttempts, String password, String clientName, final GenericObjectPoolConfig poolConfig,      boolean ssl, SSLSocketFactory sslSocketFactory, SSLParameters sslParameters,      HostnameVerifier hostnameVerifier, JedisClusterHostAndPortMap hostAndPortMap) {     // 访问父类BinaryJedisCluster    super(jedisClusterNode, connectionTimeout, soTimeout, maxAttempts, password, clientName, poolConfig,        ssl, sslSocketFactory, sslParameters, hostnameVerifier, hostAndPortMap);  }} public class BinaryJedisCluster implements BinaryJedisClusterCommands,    MultiKeyBinaryJedisClusterCommands, JedisClusterBinaryScriptingCommands, Closeable {  public BinaryJedisCluster(Set<HostAndPort> jedisClusterNode, int connectionTimeout, int soTimeout,      int maxAttempts, String user, String password, String clientName, GenericObjectPoolConfig poolConfig,      boolean ssl, SSLSocketFactory sslSocketFactory, SSLParameters sslParameters,      HostnameVerifier hostnameVerifier, JedisClusterHostAndPortMap hostAndPortMap) {     // 创建JedisSlotBasedConnectionHandler对象    this.connectionHandler = new JedisSlotBasedConnectionHandler(jedisClusterNode, poolConfig,        connectionTimeout, soTimeout, user, password, clientName, ssl, sslSocketFactory, sslParameters, hostnameVerifier, hostAndPortMap);     this.maxAttempts = maxAttempts;  }}

public class JedisSlotBasedConnectionHandler extends JedisClusterConnectionHandler {  public JedisSlotBasedConnectionHandler(Set<HostAndPort> nodes, GenericObjectPoolConfig poolConfig,      int connectionTimeout, int soTimeout, String user, String password, String clientName,      boolean ssl, SSLSocketFactory sslSocketFactory, SSLParameters sslParameters,      HostnameVerifier hostnameVerifier, JedisClusterHostAndPortMap portMap) {     super(nodes, poolConfig, connectionTimeout, soTimeout, user, password, clientName,        ssl, sslSocketFactory, sslParameters, hostnameVerifier, portMap);  }} public abstract class JedisClusterConnectionHandler implements Closeable {  public JedisClusterConnectionHandler(Set<HostAndPort> nodes, final GenericObjectPoolConfig poolConfig,      int connectionTimeout, int soTimeout, int infiniteSoTimeout, String user, String password, String clientName,      boolean ssl, SSLSocketFactory sslSocketFactory, SSLParameters sslParameters,      HostnameVerifier hostnameVerifier, JedisClusterHostAndPortMap portMap) {     // 创建JedisClusterInfoCache对象    this.cache = new JedisClusterInfoCache(poolConfig, connectionTimeout, soTimeout, infiniteSoTimeout,        user, password, clientName, ssl, sslSocketFactory, sslParameters, hostnameVerifier, portMap);     // 初始化jedis的Slot信息    initializeSlotsCache(nodes, connectionTimeout, soTimeout, infiniteSoTimeout,        user, password, clientName, ssl, sslSocketFactory, sslParameters, hostnameVerifier);  }    private void initializeSlotsCache(Set<HostAndPort> startNodes,      int connectionTimeout, int soTimeout, int infiniteSoTimeout, String user, String password, String clientName,      boolean ssl, SSLSocketFactory sslSocketFactory, SSLParameters sslParameters, HostnameVerifier hostnameVerifier) {    for (HostAndPort hostAndPort : startNodes) {       try (Jedis jedis = new Jedis(hostAndPort.getHost(), hostAndPort.getPort(), connectionTimeout,          soTimeout, infiniteSoTimeout, ssl, sslSocketFactory, sslParameters, hostnameVerifier)) {         // 通过discoverClusterNodesAndSlots进行集群发现        cache.discoverClusterNodesAndSlots(jedis);        return;      } catch (JedisConnectionException e) {      }    }  }}

public class JedisClusterInfoCache {  // 负责保存redis集群的节点信息  private final Map<String, JedisPool> nodes = new HashMap<>();  // 负责保存redis的槽位和redis节点的映射关系  private final Map<Integer, JedisPool> slots = new HashMap<>();   // 负责集群的发现逻辑  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;        }        // 获取redis节点对应的槽位信息        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.isEmpty()) {            continue;          }           HostAndPort targetNode = generateHostAndPort(hostInfos);          // 负责保存redis节点信息          setupNodeIfNotExist(targetNode);          if (i == MASTER_NODE_INDEX) {            // 负责保存槽位和redis节点的映射关系            assignSlotsToNode(slotNums, targetNode);          }        }      }    } finally {      w.unlock();    }  }   public void assignSlotsToNode(List<Integer> targetSlots, HostAndPort targetNode) {    w.lock();    try {      JedisPool targetPool = setupNodeIfNotExist(targetNode);      // 保存槽位和对应的JedisPool对象      for (Integer slot : targetSlots) {        slots.put(slot, targetPool);      }    } finally {      w.unlock();    }  }   public JedisPool setupNodeIfNotExist(HostAndPort node) {    w.lock();    try {      // 生产redis节点对应的nodeKey      String nodeKey = getNodeKey(node);      JedisPool existingPool = nodes.get(nodeKey);      if (existingPool != null) return existingPool;      // 生产redis节点对应的JedisPool      JedisPool nodePool = new JedisPool(poolConfig, node.getHost(), node.getPort(),          connectionTimeout, soTimeout, infiniteSoTimeout, user, password, , clientName,          ssl, sslSocketFactory, sslParameters, hostnameVerifier);      // 保存redis节点的key和对应的JedisPool对象      nodes.put(nodeKey, nodePool);      return nodePool;    } finally {      w.unlock();    }  }}

public class JedisPool extends JedisPoolAbstract {   public JedisPool() {    this(Protocol.DEFAULT_HOST, Protocol.DEFAULT_PORT);  }} public class JedisPoolAbstract extends Pool<Jedis> {   public JedisPoolAbstract() {    super();  }} public abstract class Pool<T> implements Closeable {  protected GenericObjectPool<T> internalPool;   public void initPool(final GenericObjectPoolConfig poolConfig, PooledObjectFactory<T> factory) {    if (this.internalPool != null) {      try {        closeInternalPool();      } catch (Exception e) {      }    }    this.internalPool = new GenericObjectPool<>(factory, poolConfig);  }} class JedisFactory implements PooledObjectFactory<Jedis> {     @Override  public PooledObject<Jedis> makeObject() throws Exception {    // 创建Jedis对象    final HostAndPort hp = this.hostAndPort.get();    final Jedis jedis = new Jedis(hp.getHost(), hp.getPort(), connectionTimeout, soTimeout,        infiniteSoTimeout, ssl, sslSocketFactory, sslParameters, hostnameVerifier);     try {      // Jedis对象连接      jedis.connect();      if (user != null) {        jedis.auth(user, password);      } else if (password != null) {        jedis.auth(password);      }      if (database != ) {        jedis.select(database);      }      if (clientName != null) {        jedis.clientSetname(clientName);      }    } catch (JedisException je) {      jedis.close();      throw je;    }    // 将Jedis对象包装成DefaultPooledObject进行返回    return new DefaultPooledObject<>(jedis);  }}

public class JedisCluster extends BinaryJedisCluster implements JedisClusterCommands,    MultiKeyJedisClusterCommands, JedisClusterScriptingCommands {   @Override  public String set(final String key, final String value, final SetParams params) {    return new JedisClusterCommand<String>(connectionHandler, maxAttempts) {      @Override      public String execute(Jedis connection) {        return connection.set(key, value, params);      }    }.run(key);  }}

public abstract class JedisClusterCommand<T> {  public T run(String key) {    // 针对key进行槽位的计算    return runWithRetries(JedisClusterCRC16.getSlot(key), this.maxAttempts, false, null);  }     private T runWithRetries(final int slot, int attempts, boolean tryRandomNode, JedisRedirectionException redirect) {     Jedis connection = null;    try {       if (redirect != null) {        connection = this.connectionHandler.getConnectionFromNode(redirect.getTargetNode());        if (redirect instanceof JedisAskDataException) {          connection.asking();        }      } else {        if (tryRandomNode) {          connection = connectionHandler.getConnection();        } else {          // 根据slot去获取Jedis对象          connection = connectionHandler.getConnectionFromSlot(slot);        }      }      // 执行真正的Redis的命令      return execute(connection);    } catch (JedisNoReachableClusterNodeException jnrcne) {      throw jnrcne;    } catch (JedisConnectionException jce) {       releaseConnection(connection);      connection = null;       if (attempts <= 1) {        // 保证后两次机会去重新刷新槽位和节点的对应的信息        this.connectionHandler.renewSlotCache();      }      // 按照重试次数进行重试操作      return runWithRetries(slot, attempts - 1, tryRandomNode, redirect);    } catch (JedisRedirectionException jre) {      // 针对返回Move命令立即触发重新刷新槽位和节点的对应信息      if (jre instanceof JedisMovedDataException) {        // it rebuilds cluster's slot cache recommended by Redis cluster specification        this.connectionHandler.renewSlotCache(connection);      }       releaseConnection(connection);      connection = null;       return runWithRetries(slot, attempts - 1, false, jre);    } finally {      releaseConnection(connection);    }  }}

public class JedisSlotBasedConnectionHandler extends JedisClusterConnectionHandler {   protected final JedisClusterInfoCache cache;   @Override  public Jedis getConnectionFromSlot(int slot) {    // 获取槽位对应的JedisPool对象    JedisPool connectionPool = cache.getSlotPool(slot);    if (connectionPool != null) {      // 从JedisPool对象中获取Jedis对象      return connectionPool.getResource();    } else {      // 获取失败就重新刷新槽位信息      renewSlotCache();      connectionPool = cache.getSlotPool(slot);      if (connectionPool != null) {        return connectionPool.getResource();      } else {        //no choice, fallback to new connection to random node        return getConnection();      }    }  }}

七、Jedis的Pipeline实现

public void pipeline() {    jedis = new Jedis(hnp.getHost(), hnp.getPort(), 500);    Pipeline p = jedis.pipelined();    // 批量发送命令到redis    p.set("foo", "bar");    p.get("foo");    // 同步等待响应结果    List<Object> results = p.syncAndReturnAll();     assertEquals(2, results.size());    assertEquals("OK", results.get());    assertEquals("bar", results.get(1)); }  public abstract class PipelineBase extends Queable implements BinaryRedisPipeline, RedisPipeline {   @Override  public Response<String> set(final String key, final String value) {    // 发送命令    getClient(key).set(key, value);    // pipeline的getResponse只是把待响应的请求聚合到pipelinedResponses对象当中    return getResponse(BuilderFactory.STRING);  }}  public class Queable {   private Queue<Response<?>> pipelinedResponses = new LinkedList<>();  protected <T> Response<T> getResponse(Builder<T> builder) {    Response<T> lr = new Response<>(builder);    // 统一保存到响应队列当中    pipelinedResponses.add(lr);    return lr;  }}  public class Pipeline extends MultiKeyPipelineBase implements Closeable {   public List<Object> syncAndReturnAll() {    if (getPipelinedResponseLength() > ) {      // 根据批量发送命令的个数即需要批量返回命令的个数，通过client对象进行批量读取      List<Object> unformatted = client.getMany(getPipelinedResponseLength());      List<Object> formatted = new ArrayList<>();      for (Object o : unformatted) {        try {          // 格式化每个返回的结果并终保存在列表中进行返回          formatted.add(generateResponse(o).get());        } catch (JedisDataException e) {          formatted.add(e);        }      }      return formatted;    } else {      return java.util.Collections.<Object> emptyList();    }  }}

public class BinaryJedis implements BasicCommands, BinaryJedisCommands, MultiKeyBinaryCommands,    AdvancedBinaryJedisCommands, BinaryScriptingCommands, Closeable {   @Override  public String set(final byte[] key, final byte[] value) {    checkIsInMultiOrPipeline();    // 发送命令    client.set(key, value);    // 等待请求响应    return client.getStatusCodeReply();  }}  public class Connection implements Closeable {  public String getStatusCodeReply() {    // 通过flush立即发送请求    flush();    // 处理响应请求    final byte[] resp = (byte[]) readProtocolWithCheckingBroken();    if (null == resp) {      return null;    } else {      return SafeEncoder.encode(resp);    }  }}  public class Connection implements Closeable {  protected void flush() {    try {      // 针对输出流进行flush操作保证报文的发出      outputStream.flush();    } catch (IOException ex) {      broken = true;      throw new JedisConnectionException(ex);    }  }}

八、结束语