Redis 之 Pipeline 详解

1、简介

Redis 是一种基于客户端-服务端模型以及请求/响应的 TCP 服务。一次 Redis 客户端发起的请求，经过服务端的响应后，大致会经历如下的步骤：

1. 客户端发起一个（查询/插入）请求，并监听 socket 返回，通常情况都是阻塞模式等待 Redis 服务器的响应

2. 服务端处理命令，并且返回处理结果给客户端

3. 客户端接收到服务的返回结果，程序从阻塞代码处返回

Redis 客户端和服务端之间通过网络连接进行数据传输，这个连接可以很快（loopback 接口）或很慢（建立了一个多次跳转的网络连接）。无论网络延如何延时，数据包总是能从客户端到达服务器，并从服务器返回数据回复客户端，这个时间被称之为 RTT(Round Trip Time - 往返时间)。我们可以很容易就意识到，Redis 在连续请求服务端时，即使 Redis 每秒能处理 100k 请求，但也会因为网络传输花费大量时间，导致整体性能的下降。​

因此如果遇到大量的批处理，我们可以考虑使用 Redis 的 pipeline（管道）。值得注意的是，管道技术并不是 Redis 特有的技术，管道技术往往需要客户端-服务器的共同配合，大部分工作任务其实是在客户端完成，很显然 Redis 支持管道技术，按照官网的意思，Redis 的最低版本就考虑了管道技术的支持性设计。​

如下图，多个连续的 incr 指令，使用 pipeline（管道）后，多个连续的 incr 指令只会花费一次网络来回开销，这个开销会随着 n 数值的增大，大幅减少网络 io 开销，从而提升整体服务的性能。

2、深究 pipeline

在上述简介中，提到了管道技术优化的是网络传输的耗时时间，这里通过 Redis 客户端-服务端的一次完整的网络请求来回，深入探索 pipeline 的本质。

1. 客户端调用 write 将数据写入操作系统内核(kernel)为 socket 连接分配的发送缓冲区(send buffer)

2. 客户端操作系统内核将发送缓冲区(send buffer)的数据发送到网卡(NIC)

3. 网卡(NIC)将数据通过路由(route)将数据送到 Redis 服务器机器网卡(NIC)

4. 服务器操作系统内核(kernel)将网卡(NIC)接收的数据，写入内核为 socket 分配的接收缓冲区(recv buffer)

5. 服务器进程从接收缓冲区调用 read 读取数据，并进行数据逻辑处理

6. 数据处理完成之后，服务器进程调用 write 将响应数据写入操作系统内核为 socket 分配的发送缓冲区

7. 操作系统内核将发送缓冲区的数据发送到服务器网卡

8. 服务器网卡将响应数据通过路由发送到客户端网卡

9. 客户端网卡接收响应数据

10. 客户端操作系统内核读取网卡接收到的服务器响应数据，并写入操作系统为 socket 连接分配的介绍缓冲区

11. 客户端进程调用 read 从接收缓冲区中读取服务器响应数据

12. 一次完整网络请求来回过程结束

对于 pipeline 技术而言，就是将 n * 12 个步骤，合并成 1 * 12，这样服务请求响应的总体时间将会大大的减少。

有个值得注意的点：在上述网络请求来回中，可能出现我们经常说到的 io 阻塞：

1. 当 write 操作发生，并且发送缓冲区（send buffer）满时，就会导致 write 操作阻塞

2. 当 read 操作发生，并且接收缓冲区（recv buffer）满时，就会导致 read 操作阻塞

上述的这两个阻塞如果出现，将会导致整个请求时间变长，因此我们操作大批量指令的时候，比如 10k 个指令，我们可以合理的对指令分多次批量发送，这样可以减少出现阻塞的情况，也可以避免服务器响应一个过大的答复包，导致客户端内存负载过重。​

3、benchmark 压测 pipeline

使用 Redis 提供的 benchmark 对 Redis 进行性能测试，如过你是 Windows 下的 Redis，在安装目录下有个 redis-benchmark.exe，进入 cmd 命令模式测试即可

如果你是在 Linux 下的 redis，在安装目录的 src 目录下有个 redis-benchmark

redis-benchmark 的全部指令参数如下所示，我们这里测试 pipeline，需要使用-P

通过普通方式测试 set 指令和 pipeline 方式测试 set 指令，可以看到 Redis 服务不同的 QPS：

• 普通 set 方式，Redis QPS 大概在 5.3 万左右

• 当使用 pipeline set 时，随着管道内并行请求数量的增加，Redis QPS 可以达到 100 万以上

4、Jedis 使用 pipeline

测试代码

package com.liziba.redis;
import redis.clients.jedis.Jedis;import redis.clients.jedis.Pipeline;
import java.io.IOException;
/** * <p> * 测试pipeline * </p> * * @Author: Liziba * @Date: 2021/9/14 22:43 */public class PipelineTest {
    public static void main(String[] args) throws IOException {        Jedis client = new Jedis("127.0.0.1", 6379);
        long startPipe = System.currentTimeMillis();        Pipeline pipe = client.pipelined();        pipe.multi();        for (int i = 0; i < 100000; i++) {            pipe.set("pipe" + i, i + "" );        }        pipe.exec();        pipe.close();        long endPipe = System.currentTimeMillis();        System.out.println("pipeline set cost time : " + (endPipe - startPipe) + "ms");

        for (int i = 0; i < 100000; i++) {            client.set("normal" + i, i + "");        }        System.out.println("normal set cost time : " + (System.currentTimeMillis() - endPipe)+ "ms");    }
}

测试结果