Redis的Java客户端源码解读

Redis的Java客户端源码解读

       Redis的Java客户端对应的类叫做Jedis。客户端调用redis可以直接利用IP端口建立一条连接，即创建一个Jedis，之后就可以用这个Jedis对象执行命令。也可以做一个连接池，大家不用排队使用一条连接。

像下面这样：

JedisPool pool = new JedisPool(new JedisPoolConfig(), "127.0.0.1", 2100);
// 从池中获取一个Jedis对象
Jedis jedis = pool.getResource();

连接池创建对象的代码如下，JedisPool使用的工厂类为JedisFactory

public PooledObject<Jedis> makeObject() throws Exception {
final HostAndPort hostAndPort = this.hostAndPort.get();
final Jedis jedis = new Jedis(hostAndPort.getHost(), hostAndPort.getPort(), this.timeout);

jedis.connect();
if (null != this.password) {
jedis.auth(this.password);
}
if (database != 0) {
jedis.select(database);
}
if (clientName != null) {
jedis.clientSetname(clientName);
}

return new DefaultPooledObject<Jedis>(jedis);
}

  类似于memchache，redis也可以使用集群模式，把数据分散到不同的服务器上存储。这时候客户端就要与所有的服务器建立连接，并且在执行命令的时候需要先判断数据在哪个服务器上。下面来看看集群模式下，redis客户端的执行逻辑。
  集群模式下，客户端的使用方法如下：

JedisShardInfo jedisShardInfo1 = new JedisShardInfo("127.0.0.1", 2100);
JedisShardInfo jedisShardInfo2 = new JedisShardInfo("127.0.0.1", 2200);
List<JedisShardInfo> list = new LinkedList<JedisShardInfo>();
list.add(jedisShardInfo1);
list.add(jedisShardInfo2);
// 初始化ShardedJedisPool代替JedisPool
ShardedJedisPool pool = new ShardedJedisPool(new JedisPoolConfig(), list);
// 从池中获取一个Jedis对象
ShardedJedis jedis = pool.getResource();

   此时创建的对象池不再是JedisPool而是叫做ShardedJedisPool，它返回的对象也变成了ShardedJedis。这个ShardedJedis的连接是一条虚拟的连接，你不知道命令最终会发到哪台服务器上。那么它是怎么根据key来查找到对应的服务器的呢？继续往下看，先来了解一下ShardedJedis的创建过程，它由对象工厂创建，代码如下。

public PooledObject<ShardedJedis> makeObject() throws Exception {
ShardedJedis jedis = new ShardedJedis(shards, algo, keyTagPattern);
return new DefaultPooledObject<ShardedJedis>(jedis);
}

  创建的过程只是直接new了一个对象，那么继续看看它的初始化操作，父类Shared的代码如下：

public Sharded(List<S> shards, Hashing algo, Pattern tagPattern) {
this.algo = algo;
this.tagPattern = tagPattern;
initialize(shards);
}

private void initialize(List<S> shards) {
nodes = new TreeMap<Long, S>();

for (int i = 0; i != shards.size(); ++i) {
final S shardInfo = shards.get(i);
if (shardInfo.getName() == null) for (int n = 0; n < 160 * shardInfo.getWeight(); n++) {
nodes.put(this.algo.hash("SHARD-" + i + "-NODE-" + n), shardInfo);
}
else for (int n = 0; n < 160 * shardInfo.getWeight(); n++) {
nodes.put(this.algo.hash(shardInfo.getName() + "*" + shardInfo.getWeight() + n), shardInfo);
}
resources.put(shardInfo, shardInfo.createResource());
}
}

  这是一个一致性哈希的算法，将每台服务器节点采用hash算法划分为160个虚拟节点，利用TreeMap存储起来。对Key采用同样的hash算法，然后从TreeMap获取大于等于键hash值得节点，取最邻近节点存储（后面具体分析过程）。对象池中每次创建一个ShardedJedis都会创建这样的160个虚拟节点，它们之间互不影响。接下来了解一下resources保存的是什么内容。主要看JedisSharedInfo的createResource（）方法：

public Jedis createResource() {
return new Jedis(this);
}

   一路跟下去

public Jedis(JedisShardInfo shardInfo) {
super(shardInfo);
}

public BinaryJedis(final JedisShardInfo shardInfo) {
client = new Client(shardInfo.getHost(), shardInfo.getPort());
client.setConnectionTimeout(shardInfo.getConnectionTimeout());
client.setSoTimeout(shardInfo.getSoTimeout());
client.setPassword(shardInfo.getPassword());
client.setDb(shardInfo.getDb());
}

   其实，这里的client才是一条真正的物理连接，它是redis的Connection的子类。Connection在发送命令的时候会先检查是不是已经建立了连接，如果没有则先建立连接，所以此处没有主动的建立连接而是直接执行命令。Jedis利用它来执行命令，如

public String set(final byte[] key, final byte[] value) {
checkIsInMultiOrPipeline();
client.set(key, value);
return client.getStatusCodeReply();
}

  此刻，已经了解到Sharded的resources其实就是服务器与连接（Jedis）的映射关系，每个服务器建立一条连接，保存它的数据结构是LinkedHashMap。
 下面该轮到分析SharededJedis是如何把命令发出去的。以一个简单的命令为例，

public String set(byte[] key, byte[] value) {
Jedis j = getShard(key);
return j.set(key, value);
}

  它会通过getShare（key）方法获取到Jedis对象，继续分析Jedis的获取过程：

public R getShard(byte[] key) {
return resources.get(getShardInfo(key));
}

public S getShardInfo(byte[] key) {
SortedMap<Long, S> tail = nodes.tailMap(algo.hash(key));
if (tail.isEmpty()) {
return nodes.get(nodes.firstKey());
}
return tail.get(tail.firstKey());
}

 上面的代码验证了之前所说的一致性哈希算法。
 如果只把redis作为缓存，服务器的增减不会对业务造成太大影响。若把redis作为持久化存储，则服务器发生变化的时候即使是采用了一致性哈希，数据的重新分配也会导致丢失一部分数据。