Redis Pipelining可以一次发送多个命令，并按顺序执行、返回结果，节省RTT(Round Trip Time)。

使用Pipelining

Jedis客户端支持Redis的Pipelining，使用方式如下：

public static void main(String[] args) {
    Jedis jedis = null;
    Pipeline pipeline = null;
    try{
        jedis = new Jedis("localhost", 6379);
        //使用pipeline
        pipeline = jedis.pipelined();
        //开始时间
        long start = System.currentTimeMillis();
        //删除lists
        pipeline.del("lists");
        //循环添加10000个元素
        for(int i = 0; i < 10000; i++){
            pipeline.rpush("lists", i + "");
        }
        //执行
        pipeline.sync();
        //结束时间
        long end = System.currentTimeMillis();
        System.out.println(end - start);
    }catch (Exception e){
        e.printStackTrace();
    }finally {
        if(pipeline != null){
            try {
                pipeline.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        if(jedis != null){
            jedis.close();
        }
    }
}

=> "237"

同样的命令，不使用Pipelining消耗时间是800ms-900ms。测试时Redis Server和Client都运行在同一台机器，由于本地环回接口(loopback interface)的原因RTT会非常短，真实环境下的差距会更大。

从协议层面看Pipelining

从Redis的RESP协议上看，Pipelining并没有什么特殊的地方，只是把多个命令连续的发送给Redis Server，然后一一解析返回结果：

public static void main(String[] args) throws Exception{

    Socket socket = new Socket();
    //TIME_WAIT状态下可以复用端口
    socket.setReuseAddress(true);
    //空闲时发送数据包，确认服务端状态
    socket.setKeepAlive(true);
    //关闭Nagle算法，尽快发送
    socket.setTcpNoDelay(true);
    //调用close方法立即关闭socket，丢弃所有未发送的数据包
    socket.setSoLinger(true, 0);
    //连接server
    socket.connect(new InetSocketAddress("localhost", 6379), 3000);
    //设置读取时超时时间
    socket.setSoTimeout(3000);

    OutputStream os = socket.getOutputStream();
    InputStream is = socket.getInputStream();
    /**
     * SET 命令  写第一个命令
     * 协议: array 3个元素 SET simpleKey simpleValue
     */
    os.write(getBytes("*3\r\n$3\r\nSET\r\n$9\r\nsimpleKey\r\n$11\r\nsimpleValue\r\n"));

    /**
     * GET 命令  写第二个命令
     * 协议: array 2个元素 GET simpleKey
     */
    os.write(getBytes("*2\r\n$3\r\nGET\r\n$9\r\nsimpleKey\r\n"));

    os.flush();

    /**
     * 解析第一个命令SET的返回结果
     */
    String result = analysisResult(is);

    System.out.println("SET command response : " + result);
    System.out.println();

    /**
     * 解析第二个命令GET返回结果
     */
    String value = analysisResult(is);
    System.out.println("GET command response : " + value);

}

=>
response type is : +
SET command response : OK

response type is : $
$ value len : 11
GET command response : simpleValue

RESP协议和详细代码可以参考《Redis协议：RESP》。

Jedis的Pipelining实现方式

Pipeline pipeline = jedis.pipelined();

通过Jedis对象的pipelined方法可以创建Pipeline对象。pipelined方法内部实际上是把Jedis对象的client赋给了pipeline。在《Redis客户端：Jedis》中介绍过Jedis类的结构，Pipeline类的结构与Jedis类似也实现了多个接口。不同的是方法的返回值，所有Pipeline中方法的返回值都被封装成了Response类。

当通过Pipeline对象执行命令时，同样也会委托给内部的Client对象去执行，但不会立即调用client的getXXX方法获取返回结果，而是创建了一个Response对象：

  public Response<Long> del(String key) {
    getClient(key).del(key);
    return getResponse(BuilderFactory.LONG);
  }

DEL命令的返回值是0或1表示是否删除成功，所以传入了BuilderFactory.LONG用来解析Integer型的返回结果。

  protected <T> Response<T> getResponse(Builder<T> builder) {
    Response<T> lr = new Response<T>(builder);
    pipelinedResponses.add(lr);
    return lr;
  }

在getResponse方法中，创建了Response对象，每个Response对象都有一个解析返回结果的Builder。Response按照命令的执行顺序被添加到pipelinedResponses队列中。

Pipeline对象的sync方法会真正的执行命令：

  public void sync() {
    if (getPipelinedResponseLength() > 0) {
      //这里会真正的调用client执行命令，并获取返回结果
      List<Object> unformatted = client.getAll();
      //将按照协议解析、分隔好的返回结果，按顺序赋给队列中的Response
      for (Object o : unformatted) {
        generateResponse(o);
      }
    }
  }

在getAll方法中会按照RESP协议的结构解析返回结果，将输入流中的内容按照协议格式切分成每个命令的返回结果：

  public List<Object> getAll(int except) {
    List<Object> all = new ArrayList<Object>();
    //执行命令
    flush();
    //pipelinedCommands是一个计数器，记录了执行了多少个命令
    while (pipelinedCommands > except) {
      try {
        //readProtocolWithCheckingBroken中解析了返回结果
        all.add(readProtocolWithCheckingBroken());
      } catch (JedisDataException e) {
        all.add(e);
      }
      //每解析一个命令，计数减1，为0时退出循环
      pipelinedCommands--;
    }
    return all;
  }

readProtocolWithCheckingBroken方法解析的方式与普通Jedis解析方式一致。当从Response获取返回结果时，会用设置好的Builder把unformatted的数据转换成对应的结构。

public class Response<T> {
    protected T response = null;
    ...
    
    //通过get获取命令返回值
    public T get() {
        ...
        if (!built) {
           //通过builder格式化返回结果
            build();
        }
        ...
        return response;
    }
    
    private void build() {
        ...
        response = builder.build(data);
        ...
    }
}