Redis管道

Redis之管道的使用

原文地址: https://blog.piaoruiqing.com/blog/2019/06/24/redis管道

关键词

Redis Pipelining: 客户端可以向服务器发送多个请求而无需等待回复, 最后只需一步即可读取回复.

RTT(Round Trip Time): 往返时间.

为什么要用管道

Redis是使用client-server模型和Request/Response协议的TCP服务器. 这意味着通常通过以下步骤完成请求：

• 客户端向服务器发送查询, 并通常以阻塞方式从套接字读取服务器响应.
• 服务器处理该命令并将响应发送回客户端.

应用程序与Redis通过网络进行连接, 可能非常快(本地回环), 也可能很慢. 但无论网络延迟是多少, 数据包都需要时间从客户端传输到服务器, 然后从服务器返回到客户端以进行回复(此时间称为RTT). 当客户端需要连续执行许多请求时(例如, 将多个元素添加到同一列表或使用多个键填充数据库), 很容易发现这种频繁操作很影响性能. 使用管道将多次操作通过一次IO发送给Redis服务器, 然后一次性获取每一条指令的结果, 以减少网络上的开销.

频繁操作但未使用管道的情形如下图:

sequenceDiagram
    participant client
    participant server

    client ->> +server:  
        Note over client,server: set mykey1 myvalue1
    server -->> -client: OK
    client ->> +server:  
        Note over client,server: set mykey1 myvalue1
    server -->> -client: OK
    client ->> +server:  
        Note over client,server: ...
    server -->> -client: OK

使用管道后如下图:

sequenceDiagram
    participant client
    participant server

    client ->> +server:  
        Note over client,server: set mykey1 myvalue1 
 set mykey2 myvalue2 
...
    server -->> -client: OK, OK, OK ...

如何使用

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(); // (一)
        Response response1 = pipelined.set("mykey1", "myvalue1");
        Response response2 = pipelined.set("mykey2", "myvalue2");
        Response response3 = pipelined.set("mykey3", "myvalue3");

        pipelined.sync();   // (二)

        LOGGER.info("cmd: SET mykey1 myvalue1, result: {}", response1.get());   // (三)
        LOGGER.info("cmd: SET mykey2 myvalue2, result: {}", response2.get());
        LOGGER.info("cmd: SET mykey3 myvalue3, result: {}", response3.get());
    }
}

• (一): jedis.pipelined(): 获取一个Pipeline用以批量执行指令.
• (二): pipelined.sync(): 同步执行, 通过读取全部Response来同步管道, 这个操作会关闭管道.
• (三): 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 connection = CLIENT.connect()) {

        RedisAsyncCommands async = connection.async();
        async.setAutoFlushCommands(false);
        RedisFuture future1 = async.set("mykey1", "myvalue1");
        RedisFuture future2 = async.set("mykey2", "myvalue2");
        RedisFuture future3 = async.set("mykey3", "myvalue3");

        async.flushCommands();

        LOGGER.info("cmd: SET mykey1 myvalue1, result: {}", future1.get());
        LOGGER.info("cmd: SET mykey2 myvalue2, result: {}", future1.get());
        LOGGER.info("cmd: SET mykey3 myvalue3, result: {}", future1.get());
    }
}

RedisTemplate

private final Logger LOGGER = LoggerFactory.getLogger(getClass());

@Resource
private StringRedisTemplate stringRedisTemplate;

/**
 * test pipelining with RedisTemplate
 */
@Test
public void testPipelining() {

    List