Redis中管道操作pipeline的實(shí)現(xiàn)

更新時間：2025年03月24日 09:57:55 作者：煙火纏過客

RedisPipeline是一種優(yōu)化客戶端與服務(wù)器通信的技術(shù),通過批量發(fā)送和接收命令減少網(wǎng)絡(luò)往返次數(shù),提高命令執(zhí)行效率,本文就來介紹一下Redis中管道操作pipeline的實(shí)現(xiàn),具有一定的參考價(jià)值,感興趣的可以了解一下

什么是pipeline

在 Redis 中，Pipeline（管道）是一種客戶端與服務(wù)器間通信的優(yōu)化機(jī)制，旨在減少網(wǎng)絡(luò)往返時間和提高命令執(zhí)行效率。以下是 Redis Pipeline 的具體定義和特點(diǎn)：

1.批量發(fā)送與接收：

使用 Pipeline 時，客戶端不再逐條發(fā)送命令，而是將多個命令一次性打包成一個請求包發(fā)送給 Redis 服務(wù)器。相應(yīng)地，服務(wù)器在接收到這個請求包后，不是立即返回每條命令的執(zhí)行結(jié)果，而是先將所有命令依次執(zhí)行完畢，然后將所有結(jié)果打包成一個響應(yīng)包返回給客戶端。
這種做法顯著減少了客戶端與服務(wù)器之間網(wǎng)絡(luò)通信的次數(shù)，尤其是對于需要執(zhí)行大量命令的場景，能夠極大地降低網(wǎng)絡(luò)延遲帶來的影響。

2.異步執(zhí)行

盡管 Redis Pipeline 中的所有命令是在服務(wù)器端按順序執(zhí)行的，但由于客戶端與服務(wù)器之間的通信是批量進(jìn)行的，客戶端可以在發(fā)送完一批命令后立刻開始處理其他任務(wù)，而無需等待每個命令的單獨(dú)響應(yīng)。這種異步處理方式可以更好地利用客戶端的計(jì)算資源，提高整體應(yīng)用程序的并發(fā)性能。

3.命令隔離

在 Pipeline 中，每個命令的執(zhí)行互不影響，即一個命令的執(zhí)行結(jié)果不會影響后續(xù)命令的執(zhí)行。這意味著即使某條命令執(zhí)行失敗，也不會阻止后續(xù)命令的執(zhí)行?？蛻舳嗽诮馕鲰憫?yīng)包時，可以根據(jù)響應(yīng)內(nèi)容判斷每條命令的執(zhí)行結(jié)果。

4.使用場景

Pipeline 主要適用于需要對 Redis 執(zhí)行大量命令的操作，如數(shù)據(jù)批量導(dǎo)入、大規(guī)模數(shù)據(jù)更新、復(fù)雜查詢等。這些操作若不使用 Pipeline，可能會因?yàn)榫W(wǎng)絡(luò)延遲導(dǎo)致整體執(zhí)行時間顯著增加。
對于涉及事務(wù)（transaction）的操作，雖然也可以使用 Pipeline 來打包命令，但需要注意的是，Pipeline 不提供事務(wù)的原子性和一致性保證。如果需要確保一組命令作為一個原子單位執(zhí)行，應(yīng)使用 Redis 的 MULTI/EXEC 命令來開啟事務(wù)。

5.注意事項(xiàng)

雖然 Pipeline 能夠顯著提高命令執(zhí)行效率，但一次性發(fā)送的命令數(shù)量不宜過大，否則可能導(dǎo)致數(shù)據(jù)包過大，增加網(wǎng)絡(luò)傳輸壓力，甚至超過 Redis 服務(wù)器或客戶端的緩沖區(qū)限制，引發(fā)錯誤。合理的命令打包大小需要根據(jù)實(shí)際環(huán)境和網(wǎng)絡(luò)狀況進(jìn)行調(diào)整
在使用 Pipeline 時，由于命令的響應(yīng)是延遲返回的，客戶端需要做好錯誤處理和重試策略，尤其是在網(wǎng)絡(luò)不穩(wěn)定或服務(wù)器負(fù)載較高的情況下。

總結(jié)來說，Redis Pipeline 是一種客戶端與服務(wù)器間高效通信的技術(shù)，通過批量發(fā)送和接收命令，減少網(wǎng)絡(luò)往返次數(shù)，提高命令執(zhí)行效率，尤其適用于大量命令操作的場景。在使用時需注意命令打包大小的控制以及錯誤處理。

場景一:我要向redis新增大批量的數(shù)據(jù)

Redis Pipeline允許一次性發(fā)送多個命令到Redis服務(wù)器，而無需等待每個命令的響應(yīng)，顯著減少了網(wǎng)絡(luò)往返時間和潛在的延遲。在Spring Boot應(yīng)用中，可以使用RedisTemplate的executePipelined()方法實(shí)現(xiàn)：

@Autowired
private StringRedisTemplate redisTemplate

public void batchInsertUsersWithPipeline(List<User> users, String keyPrefix, long ttlSeconds) {
    redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
        for (User user : users) {
            String key = generateKey(keyPrefix, user.getId());
            String value = objectMapper.writeValueAsString(user);

            connection.setEx(key.getBytes(), (int) ttlSeconds, value.getBytes());
        }
        return null;
    });
}

分批處理

盡管Pipeline提高了效率，但對于千萬級數(shù)據(jù)，一次性發(fā)送所有命令可能導(dǎo)致內(nèi)存溢出或網(wǎng)絡(luò)阻塞。因此，建議將數(shù)據(jù)分批處理，每批包含適量的記錄（如1000條），逐批發(fā)送至Redis：

public void insertUsersInBatches(List<User> users, String keyPrefix, long ttlSeconds, int batchSize) {
    int start = 0;
    while (start < users.size()) {
        int end = Math.min(start + batchSize, users.size());
        List<User> batch = users.subList(start, end);
        batchInsertUsersWithPipeline(batch, keyPrefix, ttlSeconds);
        start = end;
    }
}

batchInsertUsersWithPipeline方法利用Redis Pipeline機(jī)制發(fā)送批量命令，可以在一定程度上提高插入操作的并發(fā)性，減少網(wǎng)絡(luò)往返時間和整體耗時。然而，Pipeline本身并不能嚴(yán)格保證所有命令同時成功或失敗，其主要特性如下：

1.原子性：

Redis命令在Pipeline內(nèi)部是原子性的，即單個命令的執(zhí)行不會被其他命令中斷。
注意：這并不意味著整個Pipeline的所有命令作為一個整體具有原子性。Pipeline中的命令仍然是依次執(zhí)行的，只是客戶端與服務(wù)器之間的通信過程被優(yōu)化了。

2.響應(yīng)順序

Redis服務(wù)器會按接收到命令的順序返回結(jié)果。即使在Pipeline中并發(fā)發(fā)送多個命令，客戶端接收到的響應(yīng)也將按照命令發(fā)送的順序排列。

3.故障處理

如果Pipeline中的某個命令執(zhí)行失?。ㄈ缯Z法錯誤、key不存在等），后續(xù)命令通常仍會繼續(xù)執(zhí)行。
錯誤信息會包含在相應(yīng)命令的響應(yīng)中，客戶端可以根據(jù)這些信息判斷哪些命令執(zhí)行成功，哪些失敗。

綜上所述，batchInsertUsersWithPipeline方法不能嚴(yán)格保證所有命令同時成功或失敗。在實(shí)際使用中，如果需要確保一批數(shù)據(jù)要么全部成功插入，要么全部失敗回滾，可以采取以下策略：

事務(wù)（ MULTI/EXEC/DISCARD ）：

redis提供了事務(wù)（Transaction）功能，通過MULTI、EXEC和DISCARD等命令，可以將一組命令打包在一起執(zhí)行，只有當(dāng)所有命令都能成功執(zhí)行時，整個事務(wù)才會提交；否則，任何命令失敗都將導(dǎo)致整個事務(wù)回滾。
盡管Redis事務(wù)不支持回滾到某一特定狀態(tài)（即不保證隔離性），但在批量插入場景下，它可以滿足“全有或全無”的要求。

Lua腳本：

使用Lua腳本編寫批量插入邏輯，腳本在Redis服務(wù)器端執(zhí)行，具備原子性。即使在網(wǎng)絡(luò)中斷或服務(wù)器重啟等異常情況下，腳本要么完全執(zhí)行，要么完全不執(zhí)行，不會出現(xiàn)部分成功部分失敗的情況。

batchInsertUsersWithPipeline方法中的connection中各個方法的區(qū)別是什么？

1.connection.setEx(key.getBytes(), (int) ttlSeconds, value.getBytes());

這一行調(diào)用了RedisConnection的setEx方法，用于設(shè)置一個帶有過期時間（Time To Live，TTL）的鍵值對。參數(shù)說明如下：

key.getBytes(): 將給定的鍵（字符串）轉(zhuǎn)換為字節(jié)數(shù)組，這是Redis底層通信協(xié)議所要求的格式。
(int) ttlSeconds: 將過期時間（以秒為單位）轉(zhuǎn)換為整數(shù)類型，表示鍵值對在指定秒數(shù)后自動過期并被刪除。
value.getBytes(): 同樣將給定的值（用戶對象序列化后的JSON字符串）轉(zhuǎn)換為字節(jié)數(shù)組

setEx方法確保在設(shè)置鍵值對的同時為其設(shè)定一個過期時間。如果鍵已經(jīng)存在，該方法會更新鍵的值和過期時間。這個操作在Pipeline模式下是原子的，即在同一時刻只有一個setEx命令被執(zhí)行。

2.connection.multi(); 和 connection.exec();

這兩個方法涉及Redis的事務(wù)（Transaction）功能。在Pipeline模式下，由于我們希望保持較高的性能，一般不會使用這兩個方法。但如果確實(shí)需要保證一批命令的原子性，可以使用如下方式：

connection.multi(): 開啟一個事務(wù)塊，后續(xù)的所有命令都會被放入這個事務(wù)中，直到調(diào)用exec方法。在Pipeline模式下，調(diào)用multi方法可能會破壞原有的性能優(yōu)化效果。
connection.exec(): 提交并執(zhí)行事務(wù)中的所有命令。如果事務(wù)中有任何一個命令執(zhí)行失敗，其他命令也會被取消執(zhí)行，整個事務(wù)被視為失敗。

在您的batchInsertUsersWithPipeline方法中并沒有使用multi和exec，因?yàn)镻ipeline已經(jīng)提供了高效的批量執(zhí)行機(jī)制，而且這里的目的是提高插入性能，而不是實(shí)現(xiàn)嚴(yán)格的事務(wù)行為。

綜上所述，batchInsertUsersWithPipeline方法中直接使用了setEx方法，利用Pipeline來高效地批量插入帶有過期時間的鍵值對。如果需要實(shí)現(xiàn)更嚴(yán)格的事務(wù)控制，應(yīng)考慮使用Redis的事務(wù)（MULTI/EXEC）或Lua腳本，但這通常會犧牲一定的性能，并且與Pipeline機(jī)制不完全兼容。在實(shí)際應(yīng)用場景中，應(yīng)根據(jù)業(yè)務(wù)需求權(quán)衡選擇合適的操作方式。

3.connection.set()和connection.setNx有什么區(qū)別

connection.set() 和 connection.setNx() 都是Redis的鍵值對設(shè)置方法，它們的主要區(qū)別在于是否存在條件以及對已有鍵的處理方式：

1.connection.set(key, value)

這是最基礎(chǔ)的設(shè)置鍵值對的方法，無論鍵是否存在，都會直接覆蓋（或創(chuàng)建）對應(yīng)的鍵值對。參數(shù)說明如下：

key: 要設(shè)置的鍵。
value: 要關(guān)聯(lián)的值。

行為特點(diǎn)：

無條件設(shè)置：不論鍵是否存在，都會執(zhí)行設(shè)置操作。
覆蓋已有鍵：如果鍵已存在，其原有值會被新的值覆蓋。
創(chuàng)建新鍵：如果鍵不存在，會創(chuàng)建一個新的鍵值對。

2.connection.setNx(key, value)

這是帶有條件的設(shè)置鍵值對方法，僅當(dāng)鍵不存在時才會設(shè)置鍵值對。參數(shù)與set()相同：

key: 要設(shè)置的鍵
value: 要關(guān)聯(lián)的值。

行為特點(diǎn)

有條件設(shè)置：僅在鍵不存在的情況下執(zhí)行設(shè)置操作。
不覆蓋已有鍵：如果鍵已存在，該方法不會有任何動作，既不會改變鍵的值，也不會拋出錯誤。
創(chuàng)建新鍵：如果鍵不存在，會創(chuàng)建一個新的鍵值對。

總結(jié)來說，connection.set()無條件地設(shè)置或更新鍵值對，而connection.setNx()則是在鍵不存在時才設(shè)置鍵值對，如果鍵已存在，則不會執(zhí)行任何操作。前者適用于常規(guī)的鍵值更新或插入，后者常用于實(shí)現(xiàn)鎖機(jī)制、唯一性檢查等場景，確保某個鍵的值只在首次設(shè)置時有效。在您的batchInsertUsersWithPipeline方法中，由于目標(biāo)是批量插入新數(shù)據(jù)，所以使用了setEx方法（帶有過期時間的set），確保每個用戶數(shù)據(jù)作為一個新的鍵值對被添加到Redis中。如果您需要在插入前檢查鍵的唯一性，可以考慮使用setNx方法。不過，對于批量插入場景，通常假設(shè)數(shù)據(jù)是新的且鍵不存在，因此直接使用setEx更為常見。

場景二:大批量刪除redis中的數(shù)據(jù)

public void batchDeleteKeysWithPipeline(List<String> keys) {
    redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
        for (String key : keys) {
            connection.del(key.getBytes());
        }
        return null;
    });
}

redisTemplate.executePipelined() 方法創(chuàng)建了一個Pipeline上下文，允許您在回調(diào)函數(shù)內(nèi)發(fā)送多個命令而不等待響應(yīng)。
回調(diào)函數(shù)遍歷要刪除的鍵列表，對每個鍵調(diào)用 connection.del(key.getBytes())。del 方法用于刪除指定鍵，將鍵名轉(zhuǎn)換為字節(jié)數(shù)組后傳遞給Redis。
所有del命令在Pipeline中被連續(xù)發(fā)送至Redis服務(wù)器，期間客戶端不會等待任何響應(yīng)。
當(dāng)回調(diào)函數(shù)執(zhí)行完畢并返回時，Pipeline中的命令會被一次性發(fā)送至Redis，并接收所有命令的響應(yīng)。由于命令是在一次網(wǎng)絡(luò)往返中批量發(fā)送的，因此比單獨(dú)執(zhí)行每個刪除命令效率更高。

場景三:刪除redis中千萬級別的數(shù)據(jù)

1.批量刪除策略

使用 SCAN 命令結(jié)合 DEL 命令實(shí)現(xiàn)批量刪除。
SCAN 命令用于增量式地迭代數(shù)據(jù)集，避免一次性獲取所有鍵導(dǎo)致內(nèi)存溢出。

DEL 命令用于刪除單個或多個鍵。

2.并行處理

利用多線程或異步任務(wù)將批量刪除操作分散到多個工作線程中，提高刪除效率。

3.Redis 客戶端優(yōu)化：

選擇高性能、支持批量操作和管道（Pipeline）功能的 Redis 客戶端庫，如 Jedis 或 Lettuce。

4.監(jiān)控與故障恢復(fù)：

在執(zhí)行大規(guī)模刪除操作時，密切關(guān)注 Redis 的性能指標(biāo)（如 CPU、內(nèi)存、網(wǎng)絡(luò)帶寬等）以及客戶端程序的狀態(tài)。
準(zhǔn)備應(yīng)對可能的異常情況，如斷連重試、數(shù)據(jù)一致性檢查等。

基于Jedis客戶端實(shí)現(xiàn)

import redis.clients.jedis.Jedis;
import redis.clients.jedis.ScanParams;
import redis.clients.jedis.ScanResult;

public class RedisDataDeleter {

    private static final int SCAN_BATCH_SIZE = 1000; // 可根據(jù)實(shí)際情況調(diào)整
    private static final String MATCH_PATTERN = "*"; // 匹配所有鍵

    public void deleteAllKeys(Jedis jedis) {
        ScanParams scanParams = new ScanParams().count(SCAN_BATCH_SIZE).match(MATCH_PATTERN);

        String cursor = "0";
        while (true) {
            ScanResult<String> scanResult = jedis.scan(cursor, scanParams);
            cursor = scanResult.getCursor();

            List<String> keysToDelete = scanResult.getResult();
            if (!keysToDelete.isEmpty()) {
                // 使用 Pipeline 批量刪除鍵
                Pipeline pipeline = jedis.pipelined();
                for (String key : keysToDelete) {
                    pipeline.del(key);
                }
                pipeline.sync(); // 執(zhí)行批量命令
            }

            if ("0".equals(cursor)) {
                break; // 掃描完成
            }
        }
    }
}

注意

請確保在生產(chǎn)環(huán)境中適當(dāng)調(diào)整 SCAN_BATCH_SIZE 參數(shù)，使其既能充分利用系統(tǒng)資源，又不會對 Redis 服務(wù)器造成過大壓力。
在執(zhí)行大規(guī)模刪除操作前，最好先備份重要數(shù)據(jù)，并在非高峰期進(jìn)行操作，以減少對業(yè)務(wù)的影響。

如果條件允許，建議升級到 Redis 6.x 版本，并啟用 activedefrag 配置項(xiàng)，有助于在刪除大量數(shù)據(jù)后及時進(jìn)行碎片整理，保持 Redis 內(nèi)存的高效利用。同時，監(jiān)控 Redis 的內(nèi)存使用情況和碎片率，必要時手動觸發(fā) BGREWRITEAOF 或 BGSAVE 操作。

maven

<dependencies>
    <!-- ... 其他依賴 ... -->
    <dependency>
        <groupId>redis.clients</groupId>
        <artifactId>jedis</artifactId>
        <version>3.7.0</version> <!-- 根據(jù)實(shí)際版本號調(diào)整 -->
    </dependency>
</dependencies>

jedis連接池配置

spring.redis.host=192.168.1.100
spring.redis.port=6379
spring.redis.password=mysecretpassword  # 如果有密碼，請?zhí)顚?

# Jedis 連接池配置
spring.redis.jedis.pool.max-active=10
spring.redis.jedis.pool.max-idle=6
spring.redis.jedis.pool.min-idle=2
spring.redis.jedis.pool.max-wait=2000ms

jedisConfig

@Configuration
public class JedisConfig {

    @Value("${spring.redis.host}")
    private String host;

    @Value("${spring.redis.port}")
    private int port;

    @Value("${spring.redis.password}")
    private String password;

    @Bean
    public JedisPool jedisPool() {
        JedisPoolConfig poolConfig = new JedisPoolConfig();
        poolConfig.setMaxTotal(Integer.parseInt(env.getProperty("spring.redis.jedis.pool.max-active")));
        poolConfig.setMaxIdle(Integer.parseInt(env.getProperty("spring.redis.jedis.pool.max-idle")));
        poolConfig.setMinIdle(Integer.parseInt(env.getProperty("spring.redis.jedis.pool.min-idle")));
        poolConfig.setMaxWaitMillis(Long.parseLong(env.getProperty("spring.redis.jedis.pool.max-wait")));

        return new JedisPool(poolConfig, host, port, Protocol.DEFAULT_TIMEOUT, password);
    }
}

實(shí)現(xiàn) Redis 數(shù)據(jù)刪除服務(wù)

@Service
public class RedisDataDeleterService {

    @Autowired
    private JedisPool jedisPool;

    public void deleteAllKeys() {
        try (Jedis jedis = jedisPool.getResource()) {
            ScanParams scanParams = new ScanParams().match("*").count(1000);

            String cursor = "0";
            while (true) {
                ScanResult<String> scanResult = jedis.scan(cursor, scanParams);
                cursor = scanResult.getCursor();

                List<String> keysToDelete = scanResult.getResult();
                if (!keysToDelete.isEmpty()) {
                    Pipeline pipeline = jedis.pipelined();
                    for (String key : keysToDelete) {
                        pipeline.del(key);
                    }
                    pipeline.sync();
                }

                if ("0".equals(cursor)) {
                    break;
                }
            }
        }
    }
}

調(diào)用刪除服務(wù)

@RestController
@RequestMapping("/redis")
public class RedisController {

    @Autowired
    private RedisDataDeleterService redisDataDeleterService;

    @GetMapping("/delete-all-keys")
    public ResponseEntity<?> deleteAllKeys() {
        redisDataDeleterService.deleteAllKeys();
        return ResponseEntity.ok().build();
    }
}

基于Lettuce

maven

<dependencies>
    <!-- ... 其他依賴 ... -->
    <dependency>
        <groupId>io.lettuce</groupId>
        <artifactId>lettuce-core</artifactId>
        <version>6.2.?</version> <!-- 根據(jù)實(shí)際版本號調(diào)整 -->
    </dependency>
</dependencies>

配置Lettuce

Spring Boot 自動配置會為 Lettuce 提供連接池支持。在 application.properties 或 application.yml 中配置 Redis 連接信息：

spring.redis.host=192.168.1.100
spring.redis.port=6379
spring.redis.password=mysecretpassword  # 如果有密碼，請?zhí)顚?

使用 Lettuce 客戶端執(zhí)行批量刪除操作：

@Service
public class RedisDataDeleterService {

    @Autowired
    private RedisConnectionFactory connectionFactory;

    public void deleteAllKeys() {
        RedisAsyncCommands<String, String> asyncCommands = connectionFactory.getConnection().async();

        ScanArgs scanArgs = ScanArgs.Builder.matches("*").count(1000);
        RedisFuture<ScanResult<String>> scanFuture = asyncCommands.scan(ScanCursor.INITIAL, scanArgs);

        AtomicBoolean isRunning = new AtomicBoolean(true);
        AtomicReference<ScanCursor> lastCursor = new AtomicReference<>(ScanCursor.INITIAL);

        // 異步處理掃描結(jié)果
        scanFuture.thenAccept(scanResult -> {
            lastCursor.set(scanResult.getCursor());
            List<String> keysToDelete = scanResult.getKeys();
            if (!keysToDelete.isEmpty()) {
                RedisFuture<Long> delFuture = asyncCommands.del(keysToDelete.toArray(new String[0]));
                delFuture.thenAccept(count -> {
                    if (isRunning.get()) {
                        // 如果仍在運(yùn)行，繼續(xù)掃描
                        deleteAllKeysRecursive(asyncCommands, scanArgs, lastCursor, isRunning);
                    }
                });
            } else {
                isRunning.set(false);
            }
        });

        // 設(shè)置超時時間（可根據(jù)實(shí)際情況調(diào)整）
        CompletableFuture.runAsync(() -> {
            try {
                Thread.sleep(120000); // 2分鐘超時
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            isRunning.set(false);
        });
    }

    private void deleteAllKeysRecursive(RedisAsyncCommands<String, String> asyncCommands,
                                       ScanArgs scanArgs,
                                       AtomicReference<ScanCursor> lastCursor,
                                       AtomicBoolean isRunning) {
        if (isRunning.get()) {
            asyncCommands.scan(lastCursor.get(), scanArgs).thenAccept(scanResult -> {
                lastCursor.set(scanResult.getCursor());
                List<String> keysToDelete = scanResult.getKeys();
                if (!keysToDelete.isEmpty()) {
                    asyncCommands.del(keysToDelete.toArray(new String[0])).thenAccept(count -> {
                        if (isRunning.get()) {
                            deleteAllKeysRecursive(asyncCommands, scanArgs, lastCursor, isRunning);
                        }
                    });
                } else {
                    isRunning.set(false);
                }
            });
        }
    }
}

調(diào)用

@RestController
@RequestMapping("/redis")
public class RedisController {

    @Autowired
    private RedisDataDeleterService redisDataDeleterService;

    @GetMapping("/delete-all-keys")
    public ResponseEntity<?> deleteAllKeys() {
        redisDataDeleterService.deleteAllKeys();
        return ResponseEntity.ok().build();
    }
}

到此這篇關(guān)于Redis中管道操作pipeline的實(shí)現(xiàn)的文章就介紹到這了,更多相關(guān)Redis管道操作pipeline內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: