Redis如何使用Pipeline實現(xiàn)批處理操作

更新時間：2025年02月17日 10:03:55 作者：記得開心一點嘛

Redis?Pipeline?是一種優(yōu)化?Redis?操作的機制,通過將多個命令打包發(fā)送到?Redis?服務器,減少客戶端與服務器之間的網絡往返時間,本文主要來聊聊Redis如何使用Pipeline實現(xiàn)批處理操作,需要的可以了解下

在正常情況下，我們每次發(fā)送 Redis 命令時，客戶端會等待 Redis 服務器的響應，直到接收到結果后，才會發(fā)送下一個命令。這種方式雖然保證了操作的順序性，但在執(zhí)行大量命令時會產生很大的網絡延遲。

通過 Pipeline 技術，我們的客戶端可以將多個命令同時發(fā)送給 Redis 服務器，并且不需要等待每個命令的返回結果，直到所有命令都被執(zhí)行完畢，客戶端再一起獲取返回值。這樣能減少每個命令的等待時間，大幅提高執(zhí)行效率。

Redis Pipeline 是一種優(yōu)化 Redis 操作的機制，通過將多個命令打包發(fā)送到 Redis 服務器，減少客戶端與服務器之間的網絡往返時間（RTT），從而顯著提升性能。

在默認情況下，Redis 客戶端與服務器之間的通信是請求-響應模式，即：

1客戶端發(fā)送一個命令到服務器。

2.服務器執(zhí)行命令并返回結果。

3.客戶端等待響應后再發(fā)送下一個命令。

這種模式在命令數(shù)量較少時沒有問題，但在需要執(zhí)行大量命令時，網絡往返時間（RTT）會成為性能瓶頸。所以我們需要實現(xiàn)下面目的：

1.將多個命令打包發(fā)送到服務器。

2.服務器依次執(zhí)行這些命令，并將結果一次性返回給客戶端。

3.減少網絡開銷，提升性能。

以下是一個簡單的 Java 示例，展示了如何使用 Jedis（Redis 的一個 Java 客戶端）執(zhí)行 Pipeline：

注意：批處理時不建議一次攜帶太多命令，并且Pipeline的多個命令之間不具備原子性。

// 創(chuàng)建 Jedis 實例
Jedis jedis = new Jedis("localhost", 6379);
 
// 使用 pipelining 方式批量執(zhí)行命令
Pipeline pipeline = jedis.pipelined();
 
// 批量操作：使用 pipeline 來緩存命令
for (int i = 0; i < 1000; i++) {
    pipeline.set("key" + i, "value" + i);
}
 
// 同步執(zhí)行所有命令
pipeline.sync();

pipelined() 方法：創(chuàng)建一個 Pipeline 對象，它緩存所有要執(zhí)行的命令。

批量設置命令：通過 pipeline.set() 將多個 SET 命令放入管道中，但命令并不會立即執(zhí)行。

sync() 方法：通過調用 sync() 方法，客戶端將會把所有緩存的命令一次性發(fā)送給 Redis，并等待它們完成執(zhí)行。

但是這些都是在單機模式下的批處理，那對于集群來說該如何使用呢？

向MSet或Pipeline這樣的批處理需要在一次請求中攜帶多條命令，而此時如何Redis是一個集群，那批處理命令的多個key必須落在同一個插槽中，否則就會導致執(zhí)行失敗。

一般推薦使用并行插槽來解決，如果使用hash_tag，可能會出現(xiàn)大量的key分配同一插槽導致數(shù)據(jù)傾斜，而并行插槽不會。

那么這里我們模擬一下并行插槽實現(xiàn)：

將多個鍵值對按照Redis集群的槽位進行分組，然后分別使用jedisCluster.mset()方法按組設置鍵值對。

public class JedisClusterTest {
 
    // 聲明一個JedisCluster對象，用于與Redis集群進行交互
    private JedisCluster jedisCluster;
 
    // 在每個測試方法執(zhí)行之前，初始化JedisCluster連接
    @BeforeEach
    void setUp() {
        // 配置Jedis連接池
        JedisPoolConfig poolConfig = new JedisPoolConfig();
        poolConfig.setMaxTotal(8);
        poolConfig.setMaxIdle(8);
        poolConfig.setMinIdle(0);
        poolConfig.setMaxWaitMillis(1000);
 
        // 創(chuàng)建一個HashSet，用于存儲Redis集群的節(jié)點信息
        HashSet<HostAndPort> nodes = new HashSet<>();
        // 添加Redis集群的節(jié)點信息（IP和端口）
        nodes.add(new HostAndPort("192.168.150.101", 7001));
        nodes.add(new HostAndPort("192.168.150.101", 7002));
        nodes.add(new HostAndPort("192.168.150.101", 7003));
        nodes.add(new HostAndPort("192.168.150.101", 8001));
        nodes.add(new HostAndPort("192.168.150.101", 8002));
        nodes.add(new HostAndPort("192.168.150.101", 8003));
 
        // 使用配置的連接池和節(jié)點信息初始化JedisCluster對象
        jedisCluster = new JedisCluster(nodes, poolConfig);
    }
 
    // 測試方法：使用mset命令一次性設置多個鍵值對
    @Test
    void testMSet() {
        // 使用JedisCluster的mset方法，一次性設置多個鍵值對
        // 但是jedisCluster默認是無法解決批處理問題的，需要我們手動解決
        jedisCluster.mset("name", "Jack", "age", "21", "sex", "male");
    }
 
    // 測試方法：使用mset命令按槽位分組設置多個鍵值對
    @Test
    void testMSet2() {
        // 創(chuàng)建一個HashMap，用于存儲多個鍵值對
        Map<String, String> map = new HashMap<>(3);
        map.put("name", "Jack");
        map.put("age", "21");
        map.put("sex", "Male");
 
        // 將map中的鍵值對按照Redis集群的槽位進行分組
        Map<Integer, List<Map.Entry<String, String>>> result = map.entrySet()
                .stream()
                .collect(Collectors.groupingBy(
                        // 使用ClusterSlotHashUtil計算每個鍵對應的槽位
                        entry -> ClusterSlotHashUtil.calculateSlot(entry.getKey()))
                );
 
        // 遍歷按哈希槽分組后的結果
        for (List<Map.Entry<String, String>> list : result.values()) {
            // 創(chuàng)建一個數(shù)組用于批量設置Redis的鍵值對
            String[] arr = new String[list.size() * 2];  // 每個鍵值對包含兩個元素
            int j = 0;  // 索引變量，用于在數(shù)組中定位位置
            for (int i = 0; i < list.size(); i++) {
                j = i << 1;  // 通過位移計算數(shù)組中的位置
                Map.Entry<String, String> e = list.get(i);  // 獲取當前的鍵值對
                arr[j] = e.getKey();  // 將鍵放入數(shù)組中
                arr[j + 1] = e.getValue();  // 將值放入數(shù)組中
            }
            // 批量設置Redis集群中的鍵值對
            jedisCluster.mset(arr);
        }
    }
 
    // 在每個測試方法執(zhí)行之后，關閉JedisCluster連接
    @AfterEach
    void tearDown() {
        // 如果JedisCluster對象不為空，則關閉連接
        if (jedisCluster != null) {
            jedisCluster.close();
        }
    }
}

而在Redis集群環(huán)境下，如果需要批量獲取多個鍵的值，可以使用multiGet方法。multiGet是RedisTemplate提供的一個方法，用于一次性獲取多個鍵的值。然而，需要注意的是，multiGet在集群環(huán)境下要求所有鍵必須位于同一個槽位（slot），否則會拋出異常。

@Service
public class RedisService {
 
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
 
    /**
     * 跨槽位批量獲取多個鍵的值
     */
    public Map<String, Object> batchGetCrossSlot(List<String> keys) {
        // 按槽位分組
        Map<Integer, List<String>> slotKeyMap = keys.stream()
                .collect(Collectors.groupingBy(ClusterSlotHashUtil::calculateSlot));
 
        // 存儲最終結果
        Map<String, Object> result = new HashMap<>();
 
        // 對每個槽位的鍵分別調用multiGet
        for (Map.Entry<Integer, List<String>> entry : slotKeyMap.entrySet()) {
            List<String> slotKeys = entry.getValue();
            List<Object> slotValues = redisTemplate.opsForValue().multiGet(slotKeys);
 
            // 將結果存入Map
            for (int i = 0; i < slotKeys.size(); i++) {
                result.put(slotKeys.get(i), slotValues.get(i));
            }
        }
 
        return result;
    }
 
    /**
     * 測試跨槽位批量獲取方法
     */
    public void testBatchGetCrossSlot() {
        List<String> keys = Arrays.asList("name", "age", "sex");
        Map<String, Object> values = batchGetCrossSlot(keys);
 
        // 打印結果
        values.forEach((key, value) -> {
            System.out.println("Key: " + key + ", Value: " + value);
        });
    }
}

到此這篇關于Redis如何使用Pipeline實現(xiàn)批處理操作的文章就介紹到這了,更多相關Redis Pipeline批處理操作內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: