Redis管道的使用
原文介绍：http://redis.cn/topics/pipelining.html
- Redis Pipelining
客户端可以向服务器发送多个请求而无需等待回复, 最后只需一步即可读取回复。
- RTT(Round Trip Time)
往返时间。
- 为什么需要通道
Redis是使用client-server模型和Request/Response协议的TCP服务器. 这意味着通常通过以下步骤完成请求：
客户端向服务器发送查询, 并通常以阻塞方式从套接字读取服务器响应.
服务器处理该命令并将响应发送回客户端。
应用程序与Redis通过网络进行连接, 可能非常快(本地回环), 也可能很慢. 但无论网络延迟是多少, 数据包都需要时间从客户端传输到服务器, 然后从服务器返回到客户端以进行回复(此时间称为RTT). 当客户端需要连续执行许多请求时(例如, 将多个元素添加到同一列表或使用多个键填充数据库), 很容易发现这种频繁操作很影响性能. 使用管道将多次操作通过一次IO发送给Redis服务器, 然后一次性获取每一条指令的结果, 以减少网络上的开销。
未使用管道的情景：
屏幕快照 2021-01-24 下午7.07.13.png
使用管道的情景：
屏幕快照 2021-01-24 下午7.08.35.png
- 代码示例
Jedis
/** jedis pool */
private final Logger LOGGER = LoggerFactory.getLogger(getClass());
private static final JedisPool POOL = new JedisPool(new JedisPoolConfig(), "test-redis-server", 6379);
/**
* test pipelining with Jedis
*/
@Test
public void testPipelining() {
try (Jedis jedis = POOL.getResource()) {
Pipeline pipelined = jedis.pipelined(); // PS1
Response<String> response1 = pipelined.set("mykey1", "myvalue1");
Response<String> response2 = pipelined.set("mykey2", "myvalue2");
Response<String> response3 = pipelined.set("mykey3", "myvalue3");
pipelined.sync(); // PS2
LOGGER.info("cmd: SET mykey1 myvalue1, result: {}", response1.get()); // PS3
LOGGER.info("cmd: SET mykey2 myvalue2, result: {}", response2.get());
LOGGER.info("cmd: SET mykey3 myvalue3, result: {}", response3.get());
}
}
- PS1: jedis.pipelined(): 获取一个Pipeline用以批量执行指令。
- PS2: pipelined.sync(): 同步执行, 通过读取全部Response来同步管道, 这个操作会关闭管道。
- PS3: response1.get(): 获取执行结果。注意: 在执行pipelined.sync()之前, get是无法获取到结果的。
Lettuce
private final Logger LOGGER = LoggerFactory.getLogger(getClass());
/** redis client */
private static final RedisClient CLIENT = RedisClient.create("redis://@test-redis-server:6379/0");
/**
* test pipelining with Lettuce
*/
@Test
public void testPipelining() throws ExecutionException, InterruptedException {
try (StatefulRedisConnection<String, String> connection = CLIENT.connect()) {
RedisAsyncCommands<String, String> async = connection.async(); // PS1
async.setAutoFlushCommands(false);
RedisFuture<String> future1 = async.set("mykey1", "myvalue1");
RedisFuture<String> future2 = async.set("mykey2", "myvalue2");
RedisFuture<String> future3 = async.set("mykey3", "myvalue3");
async.flushCommands(); // PS2
LOGGER.info("cmd: SET mykey1 myvalue1, result: {}", future1.get()); // PS3
LOGGER.info("cmd: SET mykey2 myvalue2, result: {}", future1.get());
LOGGER.info("cmd: SET mykey3 myvalue3, result: {}", future1.get());
}
}
- 简单对比测试
/**
* pipeline vs direct
*/
@Test
public void compared() {
// PS1
try (Jedis jedis = POOL.getResource()) {
long start = System.nanoTime();
Pipeline pipelined = jedis.pipelined();
for (int index = 0; index < 500; index++) {
pipelined.set("mykey" + index, "myvalue" + index);
}
pipelined.sync();
long end = System.nanoTime();
LOGGER.info("pipeline cost: {} ns", end - start);
}
// PS2
try (Jedis jedis = POOL.getResource()) {
long start = System.nanoTime();
for (int index = 0; index < 500; index++) {
jedis.set("mykey" + index, "myvalue" + index);
}
long end = System.nanoTime();
LOGGER.info("direct cost: {} ns", end - start);
}
}
输出：
01:16:00.523 [main] INFO - pipeline cost: 73681257 ns // PS1 - 使用管道
01:16:03.040 [main] INFO - direct cost : 2511915103 ns // PS2 - 未使用管道
500次set执行时间总和已经和管道执行一次的所消耗的时间不在一个量级上了
————————————————————
坐标帝都，白天上班族，晚上是知识的分享者
如果读完觉得有收获的话，欢迎点赞加关注
