概念

Redis客户端执行一条命令分为如下四个过程:

发送命令
命令排队
命令执行
返回结果
其中第一个和第四个称为执行一条命令的往返时间,Round Trip Time(RTT).
Pipeline指管道技术,客户端允许将多个请求一次发送给服务器,过程中不需要等待请求的回复,服务端将结果都查回后,一次返回给客户端,通过这种方式能够提高吞吐量

对比分析

Redis提供了批量操作命令,如mget,mset,但有些命令不支持批量操作,如果有一个场景需要批量操作某些key,加入有1000个

(我们将使用批量和不使用批量操作时的排队时间忽略)在我们没有使用批量操作时所需要的时间是
1000次RTT时间+1000次命令执行时间

2.使用pipeline操作一次传递命令时,所需要的时间是:
1次RTT时间+1000次命令执行时间

举例分析

测试不使用pipeline和使用pipeline分别向redis中存放一千条

public class PipeLineDemo {
    public static void main(String[] args) {
        testUnUsed();
        testUsed();
    }
    //没使用pipeline
    public static void testUnUsed(){
        long currentTimeMillis = System.currentTimeMillis();
        Jedis jedis = new Jedis("192.168.20.188", 6379);
        for (int i = 0; i < 1000; i++) {
            jedis.set("aaaa" + i, "aaaa" + i);
        }
        long endTimeMillis = System.currentTimeMillis();
        System.out.println("unused "+(endTimeMillis - currentTimeMillis));
    }
    //使用pipeline
    public static void testUsed(){
        long currentTimeMillis = System.currentTimeMillis();
        Jedis jedis = new Jedis("192.168.20.188", 6379);
        Pipeline pipelined = jedis.pipelined();
        for (int i = 0; i < 1000; i++) {
            pipelined.set("bbbb" + i, i + "bbbb");
        }
        pipelined.sync();
        long endTimeMillis = System.currentTimeMillis();
        System.out.println("used "+ (endTimeMillis - currentTimeMillis));
    }
}

执行结果如下:

unused 879
used 51

结果显示,使用了pipeline后提升了将近17呗,效果非常明显。

保证事务

pipeline是一种批量操作,如何保证其事务?在这里需要声明redis的事务是不支持回滚的。它能保证的是这次的命令都不执行。pipeline期间是独占链接的,在pipeline开启,并未关闭的情况下,执行其他非pipeline命令将会报异常,这一点也许要注意。事务的使用如下代码:

//使用pipeline
    public static void testUsed(){
        long currentTimeMillis = System.currentTimeMillis();
        Jedis jedis = new Jedis("192.168.20.188", 6379);
        Pipeline pipelined = jedis.pipelined();
        //开启事务
        pipelined.multi();
        for (int i = 0; i < 1000; i++) {
            pipelined.set("bbbb" + i, i + "bbbb");
        }
        //提交事务
        pipelined.exec();
        //关闭pipeline
        pipelined.sync();
        long endTimeMillis = System.currentTimeMillis();
        System.out.println("used"+ (endTimeMillis - currentTimeMillis));
    }