Springboot下使用Redis管道(pipeline)進行批量操作的介紹

今天小編給大家分享的是Springboot下使用Redis管道(pipeline)進行批量操作的介紹，相信很多人都不太了解，為了讓大家更加了解，所以給大家總結了以下內容，一起往下看吧。一定會有所收獲的哦。

前言

之前有業務場景需要批量插入數據到Redis中，做的過程中也有一些感悟，因此記錄下來，以防忘記。下面的內容會涉及到

• 分別使用for、管道處理批量操作，比較其所花費時間。

• 分別使用RedisCallback、SessionCallback進行Redis pipeline 操作

• 解釋RedisCallback、SessionCallback這兩種用法的區別

管道（pipeline）的優勢

以下內容結合了redis官方文檔，總結出自己的想法。

Redis Pipeline官網地址：https://redis.io/docs/manual/pipelining/

1.網絡傳輸(RTT)開銷少

Redis的傳輸層是基于TCP協議,一次操作請求的完成，存在網絡傳輸來回的開銷，即使Redis每秒能接受10萬的請求，但也會因為網絡傳輸而浪費很多時間，導致降低整體的性能。所以面對大量的批量處理，可以使用Redis的管道（pipeline），優勢在于多次指令操作只會使用一次的網絡傳輸的開銷。

PS:像批量插入、批量獲取，RedisTemplate提供了multiSet、multiGet的方法可以進行操作，不過像multiSet并不支持批量設置key的過期時間，可以考慮業務場景進行使用

2.提高redis每秒可以執行操作的數量

在進行批量操作的前提下

不使用管道的時候，每一次Redis執行命令，都要涉及到讀(read)和寫(write)的系統調用，系統會將用戶端切換到內核端。上下文切換會有一定的消耗使用管道的話，多條命令只需要一個讀(read)，多條響應只需要一個寫(write)，可想而知，這其中省下了很多的消耗。

環境配置

• JDK8

• Spring boot 2.6.13

• spring-boot-starter-data-redis

分別使用for、管道處理批量操作，比較其所花費時間

public void testForOrPipeline(){
     //使用for
    StopWatch stopWatch2=new StopWatch();
    stopWatch2.start();
    for(int i=0;i<10000;i++){
        String value = String.valueOf(i);
        String key = "test:" + value;
        redisTemplate.opsForValue().set(key, value, 10, TimeUnit.SECONDS);
    }
    stopWatch2.stop();
    System.out.println("for所需時間："+stopWatch2.getTotalTimeSeconds()+"s");
    //使用管道
    StopWatch stopWatch3=new StopWatch();
    stopWatch3.start();
    List<Boolean> list = redisTemplate.executePipelined(new SessionCallback<Object>() {
        @Override
        public Object execute(RedisOperations operations) throws DataAccessException {
            for (int i = 0; i < 10000; i++) {
                String value = String.valueOf(i);
                String key = "test:" + value;
                operations.opsForValue().set(key, value, 10, TimeUnit.SECONDS);
            }
            return null;
        }
    });
    stopWatch3.stop();
    System.out.println("管道所需時間："+stopWatch3.getTotalTimeSeconds()+"s");
}

PS: 本地，且只有一個客戶端的情況下測試（做不到嚴謹性，見諒）

目前只是本地跑（網絡傳輸所帶來的開銷本身會很小），如果redis服務端是在其他地區的服務器上，這兩種方式所需的時間相差還會越來越大。

分別使用RedisCallback、SessionCallback進行Redis pipeline 操作

RedisCallback

private void RedisCallBackHandler() {
    //這里獲取String類型的序列化器
    RedisSerializer stringSerializer = redisTemplate.getStringSerializer();
    //第二個參數是指定結果反序列化器，用于反序列化管道中讀到的數據，不是必傳，
    //如果不傳，則使用自定義RedisTemplate的配置，
    //如果沒有自定義，則使用RedisTemplate默認的配置（JDK反序列化）
    List list = redisTemplate.executePipelined(new RedisCallback<Object>() {
        @Override
        public Object doInRedis(RedisConnection connection) throws DataAccessException {
            for (int i = 0; i < 10; i++) {
                String value = String.valueOf(i);
                String key="test:"+value;
                connection.setEx(stringSerializer.serialize(key),10,stringSerializer.serialize(value));
            }
            //這里bytes只會獲取到null，因為這里get操作只是放在管道里面，并沒有
            //真正執行，所以獲取不到值
            //byte[] bytes = connection.get("test:1".getBytes());
            connection.get("test:1".getBytes());
            //executePipelined 這個方法需要返回值為null，不然會拋異常，
            //這一點可以查看executePipelined源碼
            return null;
        }
    }, stringSerializer);
    list.stream().forEach(result->{
        System.out.println(result);
    });
}

執行結果：

SessionCallback

private void SessionCallBackHandler() {
    //這里獲取String類型的序列化
    RedisSerializer stringSerializer = redisTemplate.getStringSerializer();
    //第二個參數是指定結果反序列化器，用于反序列化管道中讀到的數據，不是必傳，
    //如果不傳，則使用自定義RedisTemplate的配置，
    //如果沒有自定義，則使用RedisTemplate默認的配置（JDK反序列化）
    List list = redisTemplate.executePipelined(new SessionCallback<Object>() {
        @Override
        public Object execute(RedisOperations operations) throws DataAccessException {
            for (int i = 0; i < 10; i++) {
                String value = String.valueOf(i);
                String key = "test:" + value;
                operations.opsForValue().set(key, value, 10, TimeUnit.SECONDS);
            }
            //這里o只會獲取到null，因為這里get操作只是放在管道里面，并沒有真正執行，所以獲取不到值
            //Object o = operations.opsForValue().get("test:1");
            operations.opsForValue().get("test:1");
            //executePipelined 這個方法需要返回值為null，不然會拋異常，
            //這一點可以查看executePipelined源碼
            return null;
        }
    }, stringSerializer);
    list.stream().forEach(result->{
        System.out.println(result);
    });
}

執行結果：

解釋RedisCallback、SessionCallback這兩種用法的區別

上面代碼顯示了RedisCallback、SessionCallback這兩種都能實現相同的效果，那么這兩個又有什么區別呢？

SessionCallback 的使用比RedisCallback要友好一些

SessionCallback的execute方法提供給使用者使用的是RedisOperations接口類，RedisTemplate實現類

RedisCallback的doInRedis方法提供給使用者使用的是RedisConnection接口類，也就是LettuceConnection是實現類

RedisConnection提供了字節數組類型的get和set方法，有關序列化部分的細節還需要我們去關心。（和使用原生jdbc感受差不多），而RedisTemplate負責序列化和連接管理，不需要讓使用者關系這一塊的部分。總結： 個人感覺從日常使用上應該都傾向于SessionCallback，而個別特殊有關底層的業務，可能就需要RedisCallback。

關于Springboot下使用Redis管道(pipeline)進行批量操作的介紹就分享到這里了，希望以上內容可以對大家有一定的參考價值，可以學以致用。如果喜歡本篇文章，不妨把它分享出去讓更多的人看到。