工作中经常使用redis作为队列，但redis队列弹出值时，只能逐个弹出，无法批量获取数据，在数据量很大时，在连接的获取和释放占用了较多的时间，效率上不是很好，只能逐个入库。Redis pipeline可以解决该问题，允许发送多个请求，批量获取数据
Springboot pipeline
springboot pipeline使用比较简单，直接调用方法即可，如下
public List<String> getQueueValues(final String queueKey, final int getCount) { List<Object> result = stringRedisTemplate.executePipelined(new RedisCallback<List<String>>() { @Override public List<String> doInRedis(RedisConnection connection) throws DataAccessException { for (int i = 0; i < getCount; i++) { connection.lPop(queueKey.getBytes()); } return null; } }); List<String> collect = null; if (result.size() > 0) { collect = result.stream().filter(item -> item != null).map(item -> item.toString()) .collect(Collectors.toList()); } return collect; }
此处使用的为RedisConnection，但其真实类型为StringRedisConnection，如果有需要，可以将其转换为StringRedisConnection，在进行后面操作。这里的返回值必须为null，否则将会出现如下错误
Caused by: org.springframework.dao.InvalidDataAccessApiUsageException: Callback cannot return a non-null value as it gets overwritten by the pipeline at org.springframework.data.redis.core.RedisTemplate.lambda$executePipelined$1(RedisTemplate.java:330) ~[spring-data-redis-2.3.3.RELEASE.jar:2.3.3.RELEASE] at org.springframework.data.redis.core.RedisTemplate.execute(RedisTemplate.java:228) ~[spring-data-redis-2.3.3.RELEASE.jar:2.3.3.RELEASE] at org.springframework.data.redis.core.RedisTemplate.execute(RedisTemplate.java:188) ~[spring-data-redis-2.3.3.RELEASE.jar:2.3.3.RELEASE] at org.springframework.data.redis.core.RedisTemplate.execute(RedisTemplate.java:175) ~[spring-data-redis-2.3.3.RELEASE.jar:2.3.3.RELEASE] at org.springframework.data.redis.core.RedisTemplate.executePipelined(RedisTemplate.java:324) ~[spring-data-redis-2.3.3.RELEASE.jar:2.3.3.RELEASE] at org.springframework.data.redis.core.RedisTemplate.executePipelined(RedisTemplate.java:314) ~[spring-data-redis-2.3.3.RELEASE.jar:2.3.3.RELEASE] at com.redispro.pipleline.RedisUtil.getQueueValues(RedisUtil.java:30) ~[classes/:na] at com.redispro.pipleline.test.RedisUtilTest.run(RedisUtilTest.java:34) ~[classes/:na] at org.springframework.boot.SpringApplication.callRunner(SpringApplication.java:795) [spring-boot-2.3.3.RELEASE.jar:2.3.3.RELEASE] ... 5 common frames omitted
但上述代码在使用时需要进行一些处理，比如现在队列中只有1个值，但是每次都批量获取500条数据，则也会组织500个pop命令，并且在返回的集合中也只有1个值，其他的都是null。可以在使用之前先判断队列大小，队列大于500，则按照500，小于等于500，则按照实际大小来获取，这样可以避免每次都组织很多多余的命令。
添加入队列方法
public void pushQueue(String queueKey, List<String> queueValues) { stringRedisTemplate.opsForList().rightPushAll(queueKey, queueValues); }
测试
List<String> list = new ArrayList<String>(); int count = 5; for (int i = 0; i < count; i++) { list.add(StringUtils.join("queue_value", i)); } redisUtil.pushQueue("testQueue", list); logger.info("Pipeline result: {}", redisUtil.getQueueValues("testQueue", count));
结果，先进先出，正常。
2020-08-16 18:34:26.459 INFO 8880 --- [ main] c.redispro.pipleline.test.RedisUtilTest : Pipeline result: [queue_value0, queue_value1, queue_value2, queue_value3, queue_value4]
Stringboot pipeline和逐个获取队列值用时对比
redis性能受到很多因素的影响，本次只是简单的做下测试，可能不是很准确，不过可以简单说明pipeline比单个获取效率更高。
逐个获取队列值
public List<String> getQueueValuesSingle(String queueKey, int getCount) { List<String> result = new ArrayList<String>(getCount); String leftPop = null; for (int i = 0; i < getCount; i++) { leftPop = stringRedisTemplate.opsForList().leftPop(queueKey); result.add(leftPop); } return result; }
分别测试pipeline和单次逐个获取
1 List<String> list = new ArrayList<String>(); 2 3 int count = 100000; 4 for (int i = 0; i < count; i++) { 5 list.add(StringUtils.join("queue_value", i)); 6 } 7 8 redisUtil.pushQueue("testQueue", list); 9 long start = System.currentTimeMillis() 10 11 redisUtil.getQueueValues("testQueue", count); 12 logger.info("pipe line use time: {}", System.currentTimeMillis() - start); 13 14 // redisUtil.getQueueCountSingle("testQueue", count); 15 // logger.info("Single get batch use time: {}", System.currentTimeMillis() - start);
以上，先执行11和12行pipeline批量获取，在注释掉这两行，放开14,15行注册，测试单次逐个获取用时
最终结果如下,不用关注具体的用时，因为不同的服务器，用时差别会很大，本次测试只是在一个vmware虚拟机上进行测试，pipeline比单词逐个获取快了18倍
2020-08-16 18:48:19.119 INFO 11292 --- [ main] c.redispro.pipleline.test.RedisUtilTest : pipe line use time: 2552
2020-08-16 18:49:44.255 INFO 2748 --- [ main] c.redispro.pipleline.test.RedisUtilTest : Single get batch use time: 47468
如果将上面条数改为100万，效果如下，pipeline快了20倍左右
2020-08-16 18:53:43.731 INFO 1748 --- [ main] c.redispro.pipleline.test.RedisUtilTest : pipe line use time: 24179
2020-08-16 19:02:45.684 INFO 19632 --- [ main] c.redispro.pipleline.test.RedisUtilTest : Single get batch use time: 490412
但一般一次获取不了这么多的数据，使用200
2020-08-16 19:06:23.569 INFO 17852 --- [ main] c.redispro.pipleline.test.RedisUtilTest : pipe line use time: 79 2020-08-16 19:06:52.363 INFO 16404 --- [ main] c.redispro.pipleline.test.RedisUtilTest : Single get batch use time: 113
100个
2020-08-16 19:12:10.427 INFO 11036 --- [ main] c.redispro.pipleline.test.RedisUtilTest : pipe line use time: 108 2020-08-16 19:12:37.291 INFO 20136 --- [ main] c.redispro.pipleline.test.RedisUtilTest : Single get batch use time: 65
10个
2020-08-16 19:07:25.762 INFO 16252 --- [ main] c.redispro.pipleline.test.RedisUtilTest : pipe line use time: 50 2020-08-16 19:08:18.523 INFO 19368 --- [ main] c.redispro.pipleline.test.RedisUtilTest : Single get batch use time: 8
1个
2020-08-16 19:10:16.268 INFO 7600 --- [ main] c.redispro.pipleline.test.RedisUtilTest : pipe line use time: 46 2020-08-16 19:11:03.790 INFO 19380 --- [ main] c.redispro.pipleline.test.RedisUtilTest : Single get batch use time: 2
通过以上的测试，可以简单看出，单从时间上来看，也不是在任何时候都适合使用pipeline的，只有在数据量大到一定程度，使用pipeline才会达到想要的目的，但是也不能根据上面的测试就断定要超过多少个值，使用pipelne，具体还需要根据自己的服务器性能来确定。
备注
理论上，多个命令节省了线程创建和销毁的开销，只要是大于1个，就应该是pipeline执行的快，可能是pipeline在初始化时需要一些处理，后面有时间在测试一下，先跑一下pipeline，然后再执行批量。pipeline其实并不适合于不确定大小的队列，因为pipeline是批量发送的脚本，所有的脚本也是批量返回，假设现在队列中只有1个数据，每次都批量获取500个，则有499个命令是多余执行的，且返回的集合中，有499个null值，如果在获取之前先判断下队列长度，大于500，则按照500，小于500，按照真实长度获取，则多线程情况下不安全，故pipeline在使用时，建议在获取固定长度的集合中获取数据或者没有现成安全的情况下使用，如果是多线程从队列获取数据，还是需要使用lua脚本来控制，将获取长度和获取队列值放到一个原子性的操作中比较合适
以下为参考资料
https://redis.io/topics/pipelining
https://docs.spring.io/spring-data/redis/docs/current/reference/html/#pipeline