redis集群-管道

造个轮子
轮子

用轮子

一个意外的发现ClusterPipeline

一个坑

造个轮子

轮子

jedis自带的cluster并不支持管道。所以得自己造轮子。

ShardedJedisPipeline

先找一下类似的轮子。jedis-3.0.0.jar里有一个切片（Sharded）相关的类。它有一个管道实现：ShardedJedisPipeline。它继承自PipelineBase类。重载了 getClient方法，实现了sync方法。实际上管道先从连接池里获取一个jedispool，然后获取一个client，所有的操作均调用这个client，然后通过sync方法将client中的指令一次性发送给server。先在的问题编程了如何获取client。

获取client。先查看一下JedisCluster中源码。发现它用JedisSlotBasedConnectionHandler extends JedisClusterConnectionHandler类来持有连接。这个类有一个Jedis getConnectionFromSlot(int slot);方法可以通过槽位来获取jedis连接。3.0的redis都使用CRC16算法来计算hash，可以使用JedisClusterCRC16的getSlot方法，将key转化成槽位即可。至于sync方法，照抄ShardedJedisPipeline即可，因为它也是使用的多client（其实不是因为懒，恩）。

JClusterPipeline（自己的轮子，其实基本照抄ShardedJedisPipeline）

public class JClusterPipeline extends PipelineBase{

private JedisSlotBasedConnectionHandler connection;

private Queue<Client> clients = new LinkedList<Client>();

@Override
protected Client getClient(byte[] key) {
//这个地方可以使用map<solt,client> 来记录已经申请到的连接，这样每次需要连接的时候，现在map里找，这样就不会那么容易连接超限。但是懒得写……
Jedis jedis = connection.getConnectionFromSlot(JedisClusterCRC16.getSlot(key));
clients.add(jedis.getClient());
return jedis.getClient();
}

@Override
protected Client getClient(String key) {
byte[] bKey = SafeEncoder.encode(key);

return getClient(bKey);
}

public void clear() {
sync();
}

public void sync() {
for (Client client : clients) {
generateResponse(client.getOne());
}
}

public List<Object> syncAndReturnAll() {
List<Object> formatted = new ArrayList<Object>();
for (Client client : clients) {
formatted.add(generateResponse(client.getOne()).get());
}
return formatted;
}
@Override
public void close() {
clean();

for (Client client : clients) {
client.close();
}

clients.clear();
}

public void setConnectionHandler(JedisSlotBasedConnectionHandler basedConnectionHandler) {
this.connection = basedConnectionHandler;
}

public void renewSlotCache() {
connection.renewSlotCache();
}
}

用轮子

要获取JedisCluster中的JedisSlotBasedConnectionHandler，要么反射，要么继承。我选择继承。

public class MyJedisCluster extends JedisCluster{

private JedisSlotBasedConnectionHandler basedConnectionHandler;

public MyJedisCluster(Set<HostAndPort> nodes) {
super(nodes);
//如果用JedisClusterConnectionHandler，没法直接调Jedis getConnectionFromSlot(int slot);方法
basedConnectionHandler = (JedisSlotBasedConnectionHandler) connectionHandler;
}

public Pipeline pipeline() {
JClusterPipeline pipeline = new JClusterPipeline();
pipeline.setConnectionHandler(basedConnectionHandler);
return pipeline;
}

@Override
public void close() {
super.close();
}
}

一个意外的发现ClusterPipeline

翻了翻jar，发现有一个ClusterPipeline，里面声明了一些有关cluster的方法，实现一下。这个接口被MultiKeyPipelineBase实现了，所以照抄一下。举个栗子。

MultiKeyPipelineBase.java 的实现
public Response<String> clusterNodes() {
client.clusterNodes();
return getResponse(BuilderFactory.STRING);
}
这里需要一个client。考虑到连接是可贵的，可以先看一下clients里有没有已经获取到的client，
如果有直接使用，如果没有，就使用connection.getConnection().getClient();方法获取一个client。
JClusterPipeline的实现
public Response<String> clusterNodes() {
Client client = null;
if(clients.isEmpty()){
client = connection.getConnection().getClient();
}else {
client = clients.peek();
}
client.clusterNodes();
clients.add(client);
return getResponse(BuilderFactory.STRING);
}

一个坑

获取连接的时候，一定要注意不要超过连接池的数量，不然会一直锁住，直到获取到连接。假如有十个连接，如果操作的11个key都在不同的槽里，这样一次操作就会申请11个连接，这就超了……

由于Cluster使用的是去中心化设计，所以客户端无法感知server节点的变化，所以得手动meet。不过这个可以通过一个代理来实现，要么自己写，要么用twitter 开源的 twemproxy。知乎上说性能会下降18%。