日期：2024年 11月 16日

標籤： Redis Transaction Lua StackExchangeRedis

摘要：C# 學習筆記

第一部分：定義目標與範圍

Step 1：目標

找出最適合大量資料寫入的 Redis 方法，了解不同方法在不同情境下的表現差異。
Redis 官網文件中有提到幾中與 Redis 交互資料的方式。

Step 2：測試專案與項目

範例代碼有以下專案架構，會有 6 種測試方式

Step 3：執行測試與數據收集

基本測試方針：

每次執行結果都會顯示於 Demo 頁面 (※其中 1 次結果畫面)

第二部分：執行測試

Step 1：取得共用數據 - 32 Kb Json

先透過隨機產生 Json 網站，製作 32 KB Json 資料
並且作為資料來源

Step 2：產生共用數據 - 代碼

每個方法都是共用相同的資料，產生 30 組 RedisKey，1000 筆 HashSet 的 32 KB 資料。
每次寫入共 975 MB 接近 1G 的批次寫入。

/// <summary>
/// 產生資料數
/// </summary>       
private List<RedisHashSetModel> GenerateImmenseData(int countSize = 1000, int accountSize = 30)
{
    var testCollection = new List<RedisHashSetModel>();

    var accountIds = GetAccountIds(accountSize: accountSize);
    for (int index = 0; index < countSize; index++)
    {
        foreach (var accountId in accountIds)
        {
            var hashSetKey = index;
            var redisKey = $@"Account:{accountId}";
            testCollection.Add(new RedisHashSetModel()
            {
                HashKey = $@"{index}",
                RedisKey = redisKey,
                Data = CommonUtil.GetTestJSon
            });
        }
    }

    // 人數 
    List<int> GetAccountIds(int accountSize)
    {
        var result = new List<int>();
        for (int accountId = 0; accountId < accountSize; accountId++)
        {
            result.Add(accountId);
        }
        return result;
    }

    return testCollection;
}

Step 3：執行測試 - 交易模式

切出每 100 筆資料開啟 Redis 的交易模式，這種模式保持了原子性，但一次量大時會占用大量 Redis 記憶體空間。

private void MultiExecTrancationExample(List<RedisHashSetModel> insertData)
{
    var batchSize = 100;
    for (int i = 0; i < insertData.Count(); i += batchSize)
    {
        var transaction = _db.CreateTransaction();
        var batch = insertData.Skip(i).Take(batchSize);
        foreach (var item in batch)
        {
            transaction.HashSetAsync(item.RedisKey, item.HashKey, item.Data);
        }
        // 提交事務
        bool committed = transaction.Execute();
    }
}

5 次執行結果如下

Step 4：執行測試 - 批次 & 管道模式

使用 Redis Pipe 技術批量發送命令，減少網路往返延遲。捨棄原子性。
開啟 30000 個 Task 等待每個都完成。

private void MultiExecPipelineAndBatchExample(List<RedisHashSetModel> insertData)
{            
    var batch = _db.CreateBatch();
    List<Task> tasks = new List<Task>();

    foreach (var item in insertData)
    {
       tasks.Add(batch.HashSetAsync(item.RedisKey, item.HashKey, item.Data));
    }
    batch.Execute();
    Task.WhenAll(tasks).Wait();
}

5 次執行結果如下

Step 5：執行測試 - Gzip 壓縮 & 批次 & 管道模式

增加 Redis Value 支援的 Gzip，進行資料壓縮，嘗試效果。

private void MultiPipelineCompressExample(List<RedisHashSetModel> insertData)
{            
    var batch = _db.CreateBatch();
    List<Task> tasks = new List<Task>();

    foreach (var item in insertData)
    {
        byte[] gzipCompressed = CommonUtil.Compress(item.Data);
        tasks.Add(batch.HashSetAsync(item.RedisKey, item.HashKey, gzipCompressed));
    }
    batch.Execute();
    Task.WhenAll(tasks).Wait();           
}

5 次執行結果如下

用 Gzip 壓縮可到 10 Kb左右，壓縮率 66%

Step 6：執行測試 - MessagePack 二進制資料 & 批次 & 管道模式

使用 MessagePack 將資料序列化為二進制格式，提升序列化/反序列化效率，嘗試效果。

private void MultiPipelineMessagePackExample(List<RedisHashSetModel> insertData)
{
    var batch = _db.CreateBatch();
    var tasks = new List<Task>();

    foreach (var item in insertData)
    {
        byte[] gzipCompressed = MessagePackSerializer.Serialize(item.Data);
        tasks.Add(batch.HashSetAsync(item.RedisKey, item.HashKey, gzipCompressed));
    }
    batch.Execute();
    Task.WhenAll(tasks).Wait();  
}

5 次執行結果如下

Step 7：執行測試 - Gzip + MessagePack 二進制資料 & 批次 & 管道模式

結合 Gzip 壓縮和 MessagePack 序列化，進一步優化數據傳輸。

private void MultiPipelineGzipAndMessagePackExample(List<RedisHashSetModel> insertData)
{
    var batch = _db.CreateBatch();
    var tasks = new List<Task>();

    foreach (var item in insertData)
    {
        byte[] gzipCompressed = MessagePackSerializer.Serialize(item.Data);
        tasks.Add(batch.HashSetAsync(item.RedisKey, item.HashKey, Compress(gzipCompressed)));
    }
    batch.Execute();
    Task.WhenAll(tasks).Wait();

    byte[] Compress(byte[] data)
    {
        using (var mso = new MemoryStream())
        {
            using (var gzip = new GZipStream(mso, CompressionLevel.Optimal))
            {
                gzip.Write(data, 0, data.Length);
            }
            return mso.ToArray();
        }
    }
}

5 次執行結果如下

Step 8：執行測試 - 使用 Lua 腳本，每 3000 筆為一組寫入

使用 Lua 腳本將多個操作封裝在一個原子操作中，減少網路往返次數。
※Lua 預設最大值為 8000

private async Task UseLuaScript(List<RedisHashSetModel> insertData)
{
    var accountGroups = insertData.GroupBy(x => x.RedisKey);
    int totalHashes = insertData.Count();            
    int luaSize = 3000;//Server 如果沒有限制預設值是 8000

    var keys = new List<RedisKey>();
    var argv = new List<RedisValue>();

    int count = 0;
    foreach (var item in accountGroups)
    {
        keys.Add(item.Key);
        foreach (var hashData in item)
        {
            argv.Add(hashData.HashKey);
            argv.Add(hashData.Data);
            count++;
            // 當達到批次大小或最後一個 hash key，執行 Lua 腳本
            if ((count + 1) % luaSize == 0 || count == totalHashes - 1)
            {
                var currentKeys = keys.ToArray();
                var currentArgv = argv.ToArray();

                var result = await _db.ScriptEvaluateAsync(GetluaScript(), currentKeys, currentArgv);                        
                argv.Clear();
            }
        }
        // 當達到批次大小或最後一個 hash key，執行 Lua 腳本
        if ((count + 1) % luaSize == 0 || count == totalHs - 1)
        {
            var currentKeys = keys.ToArray();
            var currentArgv = argv.ToArray();

            var result = await _db.ScriptEvaluateAsync(GetluaScript(), currentKeys, currentArgv);                    
            keys.Clear();
            argv.Clear();
        }
    }

    // Lua 腳本，簡單的 HashKey Value
    string GetluaScript()
    {
        return @"
        for i, key in ipairs(KEYS) do
            local args = {}
            for j = 1, #ARGV, 2 do
                table.insert(args, ARGV[j])      -- 字段
                table.insert(args, ARGV[j + 1])  -- 對應的值
            end
            -- 使用 HSET 一次設置多個字段
            redis.call('HSET', key, unpack(args))
        end
        return 'OK'
";
    }
}

5 次執行結果如下

第三部分：分析結果 & 結論

Step 1：重複 5 次執行測試 - 總計

總耗時間與補充備註：

Step 2：重複 5 次執行測試 - 範圍區間

每種方法的平均、最大、最小耗時時間：

Step 3：結論 & 適合場景與分析

交易模式

管道批次

Gzip 壓縮 & 批次 & 管道模式

MessagePack 二進制資料 & 批次 & 管道模式

Gzip + MessagePack 二進制資料 & 批次 & 管道模式

使用 Lua 腳本，每 3000 筆一組批次寫入