工作中經常使用redis作為佇列，但redis佇列彈出值時，只能逐個彈出，無法批量獲取資料，在資料量很大時，在連線的獲取和釋放佔用了較多的時間，效率上不是很好，且逐個獲取，對後面的入庫MySQL的操作也有影響，只能逐個入庫。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().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

測試上述方法是否好使，新增入佇列方法

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> getQueueCountSingle(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萬，效果如下

2020-08-16 18:53:43.731  INFO 1748 --- [           main] c.redispro.pipleline.test.RedisUtilTest  : pipe line use time: 24179

https://redis.io/topics/pipelining

https://docs.spring.io/spring-data/redis/docs/current/reference/html/#pipeline