快捷導(dǎo)航

Redis大量數(shù)據(jù)插入過(guò)程

更新時(shí)間：2025年08月05日 08:40:33 作者：懶蟲(chóng)蟲(chóng)~

本文介紹Redis處理大數(shù)據(jù)插入的三種高效方法：redis-cli管道模式、Jedis pipelined()方法及RedisTemplate批量保存,旨在減少通信開(kāi)銷(xiāo)并提升吞吐量

方式一：使用Luke協(xié)議，通過(guò)redis-cli –pipe發(fā)送數(shù)據(jù)到服務(wù)器

使用正常模式的Redis 客戶(hù)端執(zhí)行大量數(shù)據(jù)插入不是一個(gè)好主意：因?yàn)橐粋€(gè)個(gè)的插入會(huì)有大量的時(shí)間浪費(fèi)在每一個(gè)命令往返時(shí)間上。

使用管道（pipelining）是一種可行的辦法，但是在大量插入數(shù)據(jù)的同時(shí)又需要執(zhí)行其他新命令時(shí)，這時(shí)讀取數(shù)據(jù)的同時(shí)需要確保請(qǐng)可能快的的寫(xiě)入數(shù)據(jù)。

只有一小部分的客戶(hù)端支持非阻塞輸入/輸出(non-blocking I/O),并且并不是所有客戶(hù)端能以最大限度的提高吞吐量的高效的方式來(lái)分析答復(fù)。

例如，如果我們需要生成一個(gè)10億的`keyN -> ValueN’的大數(shù)據(jù)集，我們會(huì)創(chuàng)建一個(gè)如下的redis命令集的文件：

SET Key0 Value0
SET Key1 Value1
...
SET KeyN ValueN

從Redis 2.6開(kāi)始redis-cli支持一種新的被稱(chēng)之為pipe mode的新模式用于執(zhí)行大量數(shù)據(jù)插入工作。使用pipe mode模式的執(zhí)行命令如下：

cat data.txt | redis-cli --pipe

這將產(chǎn)生類(lèi)似如下的輸出：

All data transferred. Waiting for the last reply...
Last reply received from server.
errors: 0, replies: 1000000

使用redis-cli將有效的確保錯(cuò)誤輸出到Redis實(shí)例的標(biāo)準(zhǔn)輸出里面。

1.1 生成Redis協(xié)議

它會(huì)非常簡(jiǎn)單的生成和解析Redis協(xié)議，Redis協(xié)議文檔請(qǐng)參考Redis協(xié)議說(shuō)明。但是為了生成大量數(shù)據(jù)插入的目標(biāo)，你需要了解每一個(gè)細(xì)節(jié)協(xié)議，每個(gè)命令會(huì)用如下方式表示：

*<args><cr><lf>
$<len><cr><lf>
<arg0><cr><lf>
<arg1><cr><lf>
...
<argN><cr><lf>

這里的是”\r”（或者是ASCII的13）、是”\n”（或者是ASCII的10）。

例如：命令SET key value協(xié)議格式如下：

*3<cr><lf>
$3<cr><lf>
SET<cr><lf>
$3<cr><lf>
key<cr><lf>
$5<cr><lf>
value<cr><lf>

或表示為引用字符串：

"*3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$5\r\nvalue\r\n"

你需要將大量插入數(shù)據(jù)的命令按照上面的方式一個(gè)接一個(gè)的生成到文件。

1.2 pipe mode的工作原理是什么？

dis-難點(diǎn)是保證recli在pipe mode模式下執(zhí)行和netcat一樣快的同時(shí)，如何能理解服務(wù)器發(fā)送的最后一個(gè)回復(fù)。

這是通過(guò)以下方式獲得：

redis-cli –pipe試著盡可能快的發(fā)送數(shù)據(jù)到服務(wù)器。

讀取數(shù)據(jù)的同時(shí)，解析它。

一旦沒(méi)有更多的數(shù)據(jù)輸入，它就會(huì)發(fā)送一個(gè)特殊的ECHO命令，后面跟著20個(gè)隨機(jī)的字符。我們相信可以通過(guò)匹配回復(fù)相同的20個(gè)字符是同一個(gè)命令的行為。
一旦這個(gè)特殊命令發(fā)出，收到的答復(fù)就開(kāi)始匹配這20個(gè)字符，當(dāng)匹配時(shí)，就可以成功退出了。

同時(shí)，在分析回復(fù)的時(shí)候，我們會(huì)采用計(jì)數(shù)器的方法計(jì)數(shù)，以便在最后能夠告訴我們大量插入數(shù)據(jù)的數(shù)據(jù)量。

1.3 示例代碼操作

1.3.1 準(zhǔn)備數(shù)據(jù)文件，格式是文本文件，名稱(chēng)是：redis_commands.txt。

我在Windows環(huán)境下生成了一個(gè)txt文件，一條數(shù)據(jù)一行，代碼如下：

                 SET Key0 Value0
                 SET Key1 Value1
                 SET Key2 Value2
                 SET Key3 Value3
                 SET Key4 Value4
                 SET Key5 Value5
                 SET Key6 Value6
                 SET Key7 Value7
                 SET Key8 Value8
                 SET Key9 Value9
                 SET Key10 Value10
                 ...
                 SET KeyN ValueN

public class getStringTest {
    /**
     * 格式化成輸入字符串
     */
    private String getString(String... args) {
        StringBuilder sb = new StringBuilder();
        sb.append("*").append(args.length).append("\r\n");
        for (String arg : args) {
            sb.append("$").append(arg.length()).append("\r\n");
            sb.append(arg).append("\r\n");
        }
        return sb.toString();
    }
 
    @Test
    public void initFile2() {
        Long startTime = System.currentTimeMillis();
        String file = "d:\\d.txt";
        BufferedWriter w = null;
        StringBuilder sb = new StringBuilder();
        try {
            w = new BufferedWriter(new OutputStreamWriter(new FileOutputStream(file), "utf-8"));
                for(int i=0 ;i < 10000000;i++){
                //for (int i = 1; i <= 100; i++) {
                if (i / 3 == 0) {
                    w.flush();
                }
                sb.setLength(0);
                sb.append(this.getString("set", "u" + i, "name" + i));
//sb.append(this.getString("hmset", "usr" + i, "userid", "usr" + i, "username", "usrname" + i));
                w.append(sb.toString());
            }
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                w.flush();
                w.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        long endTime = System.currentTimeMillis();
        System.out.println("耗時(shí): "+(endTime - startTime)/1000+" s。");
    }
}

我生成了1000萬(wàn)的數(shù)據(jù)，因?yàn)檫@個(gè)文本文件我是在Windows環(huán)境下生成的，所以需要格式轉(zhuǎn)換。

1.3.2 如果使用Windows環(huán)境下生成的文件，需要進(jìn)行格式轉(zhuǎn)換，如果是在Linux環(huán)境下生成的文件就不需要格式轉(zhuǎn)換，如果文本文件比較大，執(zhí)行轉(zhuǎn)換時(shí)間會(huì)有幾秒，等待即可。

執(zhí)行格式轉(zhuǎn)換

[root@linux ~]# unix2dos redis_commands.txt
unix2dos:converting file redis_commands.txt to DOS format ...

以上代碼進(jìn)行格式轉(zhuǎn)換完畢

需要說(shuō)明一點(diǎn)，unix2dos這個(gè)命令需要先安裝，如果沒(méi)有安裝，會(huì)提示：command not found。

執(zhí)行以下命令安裝：

[root@linux ~]# yum install unix2dos

1.3.3 進(jìn)行數(shù)據(jù)批量插入

[root@linux ~]# cat d.txt | redis-cli -h 134.177.11.22 -p 6379 [-a "password"] -n 0 --pipe
All data transferred.Waiting for the last reply...
Last reply received from server.
errors:0，replies：10000000

方案二：采用Jedis的父類(lèi)中的pipelined()方法獲取管道

我們可以采用Jedis的父類(lèi)中的pipelined()方法獲取管道,它可以實(shí)現(xiàn)一次性發(fā)送多條命令并一次性返回結(jié)果,這樣就大量的減少了客戶(hù)端與Redis的通信次數(shù),可以有效的提高程序效率(但是,因?yàn)镽edis要一次性返回所有結(jié)果,它會(huì)把這些結(jié)果都緩存起來(lái),因此命令越多,緩存消耗的內(nèi)存也會(huì)越大,具體還要視情況而定).此外Pipeline的原理是隊(duì)列(先進(jìn)先出),這樣也保證了數(shù)據(jù)的順序性。

public static void main(String[] args) throws Exception {
 
        Jedis jedis = new Jedis("127.0.0.1", 6474);
        Pipeline p = jedis.pipelined();
        p.setex("key_a", 120, "11111");
        p.setex("key_b", 120, "2222");
        p.sync();
        if (jedis != null && jedis.isConnected()) {
            jedis.close();
        }
    }

方案三：使用RedisTemplate批量保存數(shù)據(jù)

public void saveDataToRedis(Map<String, String> map) {
        redisTemplate.executePipelined(new RedisCallback<String>() {
            @Override
            public String doInRedis(RedisConnection connection) throws DataAccessException {
                map.forEach((key, value) -> connection.set(redisTemplate.getKeySerializer().serialize(key), redisTemplate.getValueSerializer().serialize(value)));
                return null;
            }
        });
    }