一、pipeline出现的背景：

redis客户端执行一条命令分4个过程：

发送命令－〉命令排队－〉命令执行－〉返回结果

这个过程称为Round trip time(简称RTT, 往返时间)，mget mset有效节约了RTT，但大部分命令（如hgetall，并没有mhgetall）不支持批量操作，需要消耗N次RTT ，这个时候需要pipeline来解决这个问题

二、pepeline的性能

1、未使用pipeline执行N条命令

image.png

2、使用了pipeline执行N条命令

image.png

3、两者性能对比

image.png
小结：这是一组统计数据出来的数据，使用Pipeline执行速度比逐条执行要快，特别是客户端与服务端的网络延迟越大，性能体能越明显。
下面贴出测试代码分析两者的性能差异：

  1. @Test
  2. public void pipeCompare() {
  3.     Jedis redis = new Jedis("192.168.1.111", 6379);
  4.     redis.auth("12345678");//授权密码 对应redis.conf的requirepass密码
  5.     Map<String, String> data = new HashMap<String, String>();
  6.     redis.select(8);//使用第8个库
  7.     redis.flushDB();//清空第8个库所有数据
  8.     // hmset
  9.     long start = System.currentTimeMillis();
  10.     // 直接hmset
  11.     for (int i = 0; i < 10000; i++) {
  12.         data.clear();  //清空map
  13.         data.put("k_" + i, "v_" + i);
  14.         redis.hmset("key_" + i, data); //循环执行10000条数据插入redis
  15.     }
  16.     long end = System.currentTimeMillis();
  17.     System.out.println("    共插入:[" + redis.dbSize() + "]条 .. ");
  18.     System.out.println("1,未使用PIPE批量设值耗时" + (end - start) / 1000 + "秒..");
  19.     redis.select(8);
  20.     redis.flushDB();
  21.     // 使用pipeline hmset
  22.     Pipeline pipe = redis.pipelined();
  23.     start = System.currentTimeMillis();
  24.     //
  25.     for (int i = 0; i < 10000; i++) {
  26.         data.clear();
  27.         data.put("k_" + i, "v_" + i);
  28.         pipe.hmset("key_" + i, data); //将值封装到PIPE对象，此时并未执行，还停留在客户端
  29.     }
  30.     pipe.sync(); //将封装后的PIPE一次性发给redis
  31.     end = System.currentTimeMillis();
  32.     System.out.println("    PIPE共插入:[" + redis.dbSize() + "]条 .. ");
  33.     System.out.println("2,使用PIPE批量设值耗时" + (end - start) / 1000 + "秒 ..");
  34.     //--------------------------------------------------------------------------------------------------
  35.     // hmget
  36.     Set<String> keys = redis.keys("key_*"); //将上面设值所有结果键查询出来
  37.     // 直接使用Jedis hgetall
  38.     start = System.currentTimeMillis();
  39.     Map<String, Map<String, String>> result = new HashMap<String, Map<String, String>>();
  40.     for (String key : keys) {
  41.         //此处keys根据以上的设值结果，共有10000个，循环10000次
  42.         result.put(key, redis.hgetAll(key)); //使用redis对象根据键值去取值，将结果放入result对象
  43.     }
  44.     end = System.currentTimeMillis();
  45.     System.out.println("    共取值:[" + redis.dbSize() + "]条 .. ");
  46.     System.out.println("3,未使用PIPE批量取值耗时 " + (end - start) / 1000 + "秒 ..");
  48.     // 使用pipeline hgetall
  49.     result.clear();
  50.     start = System.currentTimeMillis();
  51.     for (String key : keys) {
  52.         pipe.hgetAll(key); //使用PIPE封装需要取值的key,此时还停留在客户端，并未真正执行查询请求
  53.     }
  54.     pipe.sync();  //提交到redis进行查询
  56.     end = System.currentTimeMillis();
  57.     System.out.println("    PIPE共取值:[" + redis.dbSize() + "]条 .. ");
  58.     System.out.println("4,使用PIPE批量取值耗时" + (end - start) / 1000 + "秒 ..");
  60.     redis.disconnect();
  61.     }
  1. ![image.png](https://cdn.nlark.com/yuque/0/2022/png/10371552/1649768452391-211dd6e5-1728-404c-8bb0-9e8aea7b42d2.png#clientId=uf60febed-e407-4&crop=0&crop=0&crop=1&crop=1&from=paste&id=ub2489b33&margin=%5Bobject%20Object%5D&name=image.png&originHeight=358&originWidth=583&originalType=url&ratio=1&rotation=0&showTitle=false&size=173021&status=done&style=none&taskId=u6e9e8321-6610-4c01-94d9-9942a760ee8&title=)

三、原生批命令(mset, mget)与Pipeline对比

1、原生批命令是原子性，pipeline是非原子性

(原子性概念:一个事务是一个不可分割的最小工作单位,要么都成功要么都失败。原子操作是指你的一个业务逻辑必须是不可拆分的. 处理一件事情要么都成功，要么都失败，原子不可拆分)

2、原生批命令一命令多个key, 但pipeline支持多命令（存在事务），非原子性

3、原生批命令是服务端实现，而pipeline需要服务端与客户端共同完成

四、Pipeline正确使用方式

使用pipeline组装的命令个数不能太多，不然数据量过大，增加客户端的等待时间，还可能造成网络阻塞，可以将大量命令的拆分多个小的pipeline命令完成。

1、Jedis中的pipeline使用方式

大家知道redis提供了mset、mget方法，但没有提供mdel方法，如果想实现，可以借助pipeline实现。

2、Jedis中的pipeline使用步骤：

获取jedis对象（一般从连接池中获取）
获取jedis对象的pipeline对象
添加指令
执行指令
测试类方法：

  1. @Test
  2.     public void testCommond() {
  3.         // 工具类初始化
  4.         JedisUtils jedis = new JedisUtils("192.168.1.111", 6379, "12345678");
  6.         for (int i = 0; i < 100; i++) {
  7.             // 设值
  8.             jedis.set("n" + i, String.valueOf(i));
  9.         }
  10.         System.out.println("keys from redis return =======" + jedis.keys("*"));
  12.     }
  14.     // 使用pipeline批量删除
  15.      @Test
  16.     public void testPipelineMdel() {
  17.         // 工具类初始化
  18.         JedisUtils jedis = new JedisUtils("192.168.1.111", 6379, "12345678");
  19.         List<String> keys = new ArrayList<String>();
  20.         for (int i = 0; i < 100; i++) {
  21.             keys.add("n" + i);
  22.         }
  23.         jedis.mdel(keys);
  24.         System.out.println("after mdel the redis return ---------" + jedis.keys("*"));
  25.     }

JedisUtils下的mdel方法：

  1. /**
  2.      * 删除多个字符串key 并释放连接
  3.      * 
  4.      * @param keys*
  5.      * @return 成功返回value 失败返回null
  6.      */
  7.     public boolean mdel(List<String> keys) {
  8.         Jedis jedis = null;
  9.         boolean flag = false;
  10.         try {
  11.             jedis = pool.getResource();//从连接借用Jedis对象
  12.             Pipeline pipe = jedis.pipelined();//获取jedis对象的pipeline对象
  13.             for(String key:keys){
  14.                 pipe.del(key); //将多个key放入pipe删除指令中
  15.             }
  16.             pipe.sync(); //执行命令，完全此时pipeline对象的远程调用 
  17.             flag = true;
  18.         } catch (Exception e) {
  19.             pool.returnBrokenResource(jedis);
  20.             e.printStackTrace();
  21.         } finally {
  22.             returnResource(pool, jedis);
  23.         }
  24.         return flag;
  25.     }

使用pipeline提交所有操作并返回执行结果：

  1. @Test
  2.     public void testPipelineSyncAll() {
  3.         // 工具类初始化
  4.         Jedis jedis = new Jedis("192.168.1.111", 6379);
  5.         jedis.auth("12345678");
  6.         // 获取pipeline对象
  7.         Pipeline pipe = jedis.pipelined();
  8.         pipe.multi();
  9.         pipe.set("name", "james"); // 调值
  10.         pipe.incr("age");// 自增
  11.         pipe.get("name");
  12.         pipe.discard();
  13.         // 将不同类型的操作命令合并提交，并将操作操作以list返回
  14.         List<Object> list = pipe.syncAndReturnAll();
  16.         for (Object obj : list) {
  17.             // 将操作结果打印出来
  18.             System.out.println(obj);
  19.         }
  20.         // 断开连接，释放资源
  21.         jedis.disconnect();
  22.     }

五、redis事务

pipeline是多条命令的组合，为了保证它的原子性，redis提供了简单的事务。

1、redis的简单事务，

一组需要一起执行的命令放到multi和exec两个命令之间，其中multi代表事务开始，exec代表事务结束。
image.png

2、停止事务discard

image.png

3、命令错误，语法不正确，导致事务不能正常结束

image.png

4、运行错误，语法正确，但类型错误，事务可以正常结束

image.png

5、watch命令：

使用watch后， multi失效，事务失效
image.png

WATCH的机制是：在事务EXEC命令执行时，Redis会检查被WATCH的key，只有被WATCH的key从WATCH起始时至今没有发生过变更，EXEC才会被执行。如果WATCH的key在WATCH命令到EXEC命令之间发生过变化，则EXEC命令会返回失败。

小结：redis提供了简单的事务，不支持事务回滚
————————————————
版权声明：本文为CSDN博主「lingengy」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/w1lgy/article/details/84455579