快捷導(dǎo)航

Springboot下使用Redis管道(pipeline)進(jìn)行批量操作

更新時(shí)間：2023年05月14日 14:58:49 作者：我贏了算我輸

本文主要介紹了Spring?boot?下使用Redis管道（pipeline）進(jìn)行批量操作，文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

前言

之前有業(yè)務(wù)場(chǎng)景需要批量插入數(shù)據(jù)到Redis中，做的過程中也有一些感悟，因此記錄下來，以防忘記。下面的內(nèi)容會(huì)涉及到

分別使用for、管道處理批量操作，比較其所花費(fèi)時(shí)間。
分別使用RedisCallback、SessionCallback進(jìn)行Redis pipeline 操作
解釋RedisCallback、SessionCallback這兩種用法的區(qū)別

管道（pipeline）的優(yōu)勢(shì)

以下內(nèi)容結(jié)合了redis官方文檔，總結(jié)出自己的想法。

Redis Pipeline官網(wǎng)地址：https://redis.io/docs/manual/pipelining/

1.網(wǎng)絡(luò)傳輸(RTT)開銷少

Redis的傳輸層是基于TCP協(xié)議,一次操作請(qǐng)求的完成，存在網(wǎng)絡(luò)傳輸來回的開銷，即使Redis每秒能接受10萬的請(qǐng)求，但也會(huì)因?yàn)榫W(wǎng)絡(luò)傳輸而浪費(fèi)很多時(shí)間，導(dǎo)致降低整體的性能。所以面對(duì)大量的批量處理，可以使用Redis的管道（pipeline），優(yōu)勢(shì)在于多次指令操作只會(huì)使用一次的網(wǎng)絡(luò)傳輸?shù)拈_銷。

PS:像批量插入、批量獲取，RedisTemplate提供了multiSet、multiGet的方法可以進(jìn)行操作，不過像multiSet并不支持批量設(shè)置key的過期時(shí)間，可以考慮業(yè)務(wù)場(chǎng)景進(jìn)行使用

2.提高redis每秒可以執(zhí)行操作的數(shù)量

在進(jìn)行批量操作的前提下

不使用管道的時(shí)候，每一次Redis執(zhí)行命令，都要涉及到讀(read)和寫(write)的系統(tǒng)調(diào)用，系統(tǒng)會(huì)將用戶端切換到內(nèi)核端。上下文切換會(huì)有一定的消耗使用管道的話，多條命令只需要一個(gè)讀(read)，多條響應(yīng)只需要一個(gè)寫(write)，可想而知，這其中省下了很多的消耗。

環(huán)境配置

JDK8
Spring boot 2.6.13
spring-boot-starter-data-redis

分別使用for、管道處理批量操作，比較其所花費(fèi)時(shí)間

public void testForOrPipeline(){
     //使用for
    StopWatch stopWatch1=new StopWatch();
    stopWatch1.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);
    }
    stopWatch1.stop();
    System.out.println("for所需時(shí)間："+stopWatch1.getTotalTimeSeconds()+"s");
    //使用管道
    StopWatch stopWatch2=new StopWatch();
    stopWatch2.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;
        }
    });
    stopWatch2.stop();
    System.out.println("管道所需時(shí)間："+stopWatch2.getTotalTimeSeconds()+"s");
}

方法	第一次	第二次	第三次	第四次
for	3.07s	3.07s	3.13s	3.12s
pipeline	0.55s	0.63s	0.60s	0.68s

PS: 本地，且只有一個(gè)客戶端的情況下測(cè)試（做不到嚴(yán)謹(jǐn)性，見諒）

目前只是本地跑（網(wǎng)絡(luò)傳輸所帶來的開銷本身會(huì)很小），如果redis服務(wù)端是在其他地區(qū)的服務(wù)器上，這兩種方式所需的時(shí)間相差還會(huì)越來越大。

分別使用RedisCallback、SessionCallback進(jìn)行Redis pipeline 操作

RedisCallback

private void RedisCallBackHandler() {
    //這里獲取String類型的序列化器
    RedisSerializer stringSerializer = redisTemplate.getStringSerializer();
    //第二個(gè)參數(shù)是指定結(jié)果反序列化器，用于反序列化管道中讀到的數(shù)據(jù)，不是必傳，
    //如果不傳，則使用自定義RedisTemplate的配置，
    //如果沒有自定義，則使用RedisTemplate默認(rèn)的配置（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只會(huì)獲取到null，因?yàn)檫@里get操作只是放在管道里面，并沒有
            //真正執(zhí)行，所以獲取不到值
            //byte[] bytes = connection.get("test:1".getBytes());
            connection.get("test:1".getBytes());
            //executePipelined 這個(gè)方法需要返回值為null，不然會(huì)拋異常，
            //這一點(diǎn)可以查看executePipelined源碼
            return null;
        }
    }, stringSerializer);
    list.stream().forEach(result->{
        System.out.println(result);
    });
}

執(zhí)行結(jié)果：

SessionCallback

private void SessionCallBackHandler() {
    //這里獲取String類型的序列化
    RedisSerializer stringSerializer = redisTemplate.getStringSerializer();
    //第二個(gè)參數(shù)是指定結(jié)果反序列化器，用于反序列化管道中讀到的數(shù)據(jù)，不是必傳，
    //如果不傳，則使用自定義RedisTemplate的配置，
    //如果沒有自定義，則使用RedisTemplate默認(rèn)的配置（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只會(huì)獲取到null，因?yàn)檫@里get操作只是放在管道里面，并沒有真正執(zhí)行，所以獲取不到值
            //Object o = operations.opsForValue().get("test:1");
            operations.opsForValue().get("test:1");
            //executePipelined 這個(gè)方法需要返回值為null，不然會(huì)拋異常，
            //這一點(diǎn)可以查看executePipelined源碼
            return null;
        }
    }, stringSerializer);
    list.stream().forEach(result->{
        System.out.println(result);
    });
}

執(zhí)行結(jié)果：

解釋RedisCallback、SessionCallback這兩種用法的區(qū)別

上面代碼顯示了RedisCallback、SessionCallback這兩種都能實(shí)現(xiàn)相同的效果，那么這兩個(gè)又有什么區(qū)別呢？

SessionCallback 的使用比RedisCallback要友好一些

SessionCallback的execute方法提供給使用者使用的是RedisOperations接口類，RedisTemplate實(shí)現(xiàn)類

RedisCallback的doInRedis方法提供給使用者使用的是RedisConnection接口類，也就是LettuceConnection是實(shí)現(xiàn)類

RedisConnection提供了字節(jié)數(shù)組類型的get和set方法，有關(guān)序列化部分的細(xì)節(jié)還需要我們?nèi)リP(guān)心。（和使用原生jdbc感受差不多），而RedisTemplate負(fù)責(zé)序列化和連接管理，不需要讓使用者關(guān)系這一塊的部分?？偨Y(jié)：個(gè)人感覺從日常使用上應(yīng)該都傾向于SessionCallback，而個(gè)別特殊有關(guān)底層的業(yè)務(wù)，可能就需要RedisCallback。

到此這篇關(guān)于Springboot下使用Redis管道(pipeline)進(jìn)行批量操作的文章就介紹到這了,更多相關(guān)Springboot Redis管道批量操作內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: