Redisson批量操作類RBuckets和管道利器RBatch

2022 年 5 月 14 日
筆記
mq, MQ|MongoDB

《Spring Boot 整合Redisson配置篇》
《Spring Boot 整合Redisson操作Redis基礎篇》
《Redisson批量操作類RBuckets和管道利器RBatch》

摘要：使用Redisson的類RBuckets和RBatch批量操作Redis，減少網路請求次數。

綜述

Redis的部署方式有單節點部署、哨兵方式部署、集群方式部署3種方式，這3中方式都使用的是原生的redis。本文基於單節點部署，使用的Spring Boot版本為2.5.x。

RBuckets批量操作

在Spring Boot項目中，通過RBuckets介面實現批量操作多個RBucket對象，官方示例如下：

RBuckets buckets = redisson.getBuckets();
Map<String, V> loadedBuckets = buckets.get("myBucket1", "myBucket2", "myBucket3");
Map<String, Object> map = new HashMap<>();
map.put("myBucket1", new MyObject());
map.put("myBucket2", new MyObject());
// 利用Redis的事務特性，同時保存所有的通用對象桶，如果任意一個通用對象桶已經存在則放棄保存其他所有數據。
buckets.trySet(map);
// 同時保存全部通用對象桶。
buckets.set(map);

方法介紹：

Map<String,V> get(String… keys)：返回桶的key-value對。
boolean trySet(Map<String,?> buckets)：利用Redis的事務特性，同時保存所有的通用對象桶，如果任意一個通用對象桶已經存在則放棄保存其他所有數據。
void set(Map<String,?> buckets)：同時保存全部通用對象桶。

RBatch 批量操作

多個連續命令可以通過RBatch對象在一次網路會話請求里合併發送，這樣省去了產生多個請求消耗的時間和資源。這在Redis中叫做管道。

RBatch管道功能就是REDIS的批量發送，實際上是客戶端的功能，與服務端無關。相當於把多個請求的命令放在一個數據包通過TCP發送到服務端，然後客戶端再一次性讀取所有的命令回應。管道技術最顯著的優勢是提高了 redis 服務的性能。

   /**
     * 批量操作
     */
    private void batchDemo() throws ExecutionException, InterruptedException {
        Map<String, String> map = new HashMap<>();
        map.put("abc", "testStr");
        map.put("abcDemo", "redis");
        redisUtils.setMassStrings(map);

        log.info("String 測試數據：{}", redisUtils.getStr("abc") + " "
                + redisUtils.getStr("abcDemo"));

        RBatch batch = redisUtils.createBatch();
        // 模擬購物車場景，真實場景中請替換店鋪ID shopId 和商品ID commodityId
        String field = "shopId:commodityId";
        // 把即將執⾏的命令放進 RBatch
        RMapAsync testMap = batch.getMap("customerId:"+ 32L);
        // 更新value，並返回上一次的值
        String commodityNum = "mapValue" + String.valueOf((int)(Math.random()*9 + 100));
        log.info("當前商品數量commodityNum是：{}", commodityNum);
        testMap.putAsync(field, commodityNum);
        testMap.putAsync("test2", "mapValue3");
        testMap.putAsync("test2", "mapValue5");
        testMap.putAsync("test："+ String.valueOf((int)(Math.random()*900 + 100)), String.valueOf((int)(Math.random()*900 + 100)));

        RAtomicLongAsync counter = batch.getAtomicLong("counter");

        RFuture<Long> num = counter.incrementAndGetAsync();

        // 執行RBatch中的全部命令，並返回執行結果
        BatchResult result = batch.execute();
        List list = result.getResponses();
        log.info("Map Batch 執行結果：{}", list);
        log.info("計數器當前值：{}", num.get());
    }

執行batchDemo()後，控制台列印結果如下：

StudyRedissonController - String 測試數據：testStr redis
StudyRedissonController - 當前商品數量commodityNum是：mapValue106
StudyRedissonController - Map Batch 執行結果：[mapValue101, mapValue5, mapValue3, null, 8]
StudyRedissonController - 計數器當前值：8

測試用例主要介紹了Hash，順便介紹一下它的使用場景：

存儲結構化的數據，比如 Java 中的對象。其實 Java 中的對象也可以用 string 進行存儲，只需要將對象序列化成 json 字元串就可以，但是如果這個對象的某個屬性更新比較頻繁的話，那麼每次就需要重新將整個對象序列化存儲，這樣消耗開銷比較大。可如果用 hash 來存儲對象的每個屬性，那麼每次只需要更新要更新的屬性就可以。
購物車場景。以業務線+用戶id作為key，以店鋪編號+商品的id作為存儲的field，以選購商品數量作為鍵值對的value，這樣就構成了購物車的三個要素。

在集群模式下，所有的命令會按各個槽所在的節點，篩選分配到各個節點並同時發送。每個節點返回的結果將會匯總到最終的結果列表裡。上述demo中用到的工具類如下：

@Component
public class RedisUtils {

    private RedisUtils() {
    }

    /**
     * 默認快取時間
     */
    private static final Long DEFAULT_EXPIRED = 32000L;

    /**
     * 自動裝配redisson client對象
     */
    @Resource
    private RedissonClient redissonClient;
    /**
     * 獲取getBuckets 對象
     *
     * @return RBuckets 對象
     */
    public RBuckets getBuckets() {
        return redissonClient.getBuckets();
    }
    /**
     * 讀取快取中的字元串，永久有效
     *
     * @param key 快取key
     * @return 字元串
     */
    public String getStr(String key) {
        RBucket<String> bucket = redissonClient.getBucket(key);
        return bucket.get();
    }

   // ---------------- 批量操作 ------------------------
    /**
     * 獲取RBatch
     *
     * @return RBatch
     */
    public RBatch createBatch() {
        return redissonClient.createBatch();
    }

    /**
     * 批量移除快取
     *
     * @param keys key 對象
     */
    public void deleteBatch(String... keys) {
        if (null == keys) {
            return;
        }
        this.getKeys().delete(keys);
    }

    /**
     * 批量快取字元串，缺點：不可以設置過期時間
     *
     * @param map 快取key-value
     */
    public void setMassStrings(Map<String, String> map) {
        if (MapUtils.isEmpty(map)) {
            return;
        }
        RBuckets buckets = redissonClient.getBuckets();
        // 同時保存全部通用對象桶。
        buckets.set(map);
    }

    /**
     * 批量快取字元串，支援過期
     *
     * @param map 快取key-value
     * @param leaseTime 快取有效期，必傳
     */
    public void setMassStrings(Map<String, String> map, long leaseTime) {
        if (MapUtils.isEmpty(map)) {
            return;
        }
        final long expireTime = leaseTime <= 0L ? DEFAULT_EXPIRED : leaseTime;
        RBatch batch = redissonClient.createBatch();
        map.forEach(new BiConsumer<String, String>() {
            public void accept(String key, String value) {
                batch.getBucket(key).setAsync(value, expireTime, TimeUnit.SECONDS);
            }
        });
        batch.execute();
    }

}

結束語

關於redisson中如何使用批量操作類RBuckets和管道利器RBatch就分享到這裡，希望本文對大家的學習或者工作具有一定的參考和學習價值；如果有疑問，大家可以在評論區留言交流，也希望大家多多點贊關注。謝謝大家對樓蘭胡楊的支援！

Reference

//www.bookstack.cn/read/redisson-wiki-zh/spilt.4.2.-配置方法.md

Tags: mq MQ|MongoDB