【Azure Developer】一个复制Redis Key到另一个Redis服务的工具(redis_copy_net8)

技术分享 2年前 (2024-07-13) 0 999+

关注

介绍一个简单的工具，用于将Redis数据从一个redis端点复制到另一个redis端点，基于原始存储库转换为.NET 8：https://github.com/LuBu0505/redis-copy-net8

Redis Copy .NET8

Redis Copy 控制台工具允许将 Redis 数据从一个 Redis 服务端复制到另一个。

Note: 不支持redis集群

软件要求

运行 Redis Copy 工具需要以下软件。它可能会在其他版本上运行.

.NET 8
VS Code / Visual Studio 2022

下载源代码

clone https://github.com/LuBu0505/redis-copy-net8.git

使用方式

选项 1 -- 使用 AppSetting.json

将“< ... >”替换为真实的redis端点

{   "SourceRedisConnectionString": "<source redis name>:6380,password=<your password>,ssl=True,abortConnect=False", //Source Redis ConnectionString   "DestRedisConnectionString": "<Destination redis name>:6380,password=<your password>,ssl=True,abortConnect=False" //Destination Redis ConnectionString }

选项 2 -- 使用命令参数

redis-copy-net8.exe Parameter Description:   --se           Required. SourceEndpoint *.redis.cache.windows.net   --sa           Required. Source password   --sp           (Default: 6380) Source port   --sssl         (Default: true) Connect Source over ssl    --de           Required. DestinationEndpoint *.redis.cache.windows.net   --da           Required. Destination Password   --dp           (Default: 6380) Destination port   --dssl         (Default: true) Destination Source over ssl   --help         Display this help screen.   --version      Display version information.  eg:  redis-copy-net8.exe --se <xxxxxx.redis.cache.chinacloudapi.cn> --sa <******************> --de <xxxxxx.redis.cache.chinacloudapi.cn> --da <******************>

Redis Copy 工具的工作流程

第 1 阶段：准备Redis源和目标信息

使用 StackExchange.Redis ConnectionMultiplexer 类，默认创建20个连接。
检查源redis的Used Memory、Keyspace信息
根据Keys数量拆分成更多子任务

            var infoGroup = sourcecon.BasicRetryInfo((conn) => conn.GetServer(conn.GetEndPoints()[0]).Info());              foreach (var info in infoGroup)             {                 if (info.Key.Equals("Memory"))                 {                     Console.WriteLine($"==t# {info.Key}");                     var lists = info.ToList().Where(i => i.Key.Equals("used_memory_human") || i.Key.Equals("maxmemory_human")).ToList();                     foreach (var list in lists)                         Console.WriteLine($"==t  {list.ToString()}");                 }                  if (info.Key.Equals("Keyspace"))                 {                     Console.WriteLine($"==t# {info.Key}");                     foreach (var list in info.ToList())                     {                         long dbindex, dbkeys = 0;                          long.TryParse(Regex.Match(list.Key, @"d+.*d*").Value, out dbindex);                         long.TryParse(list.Value.Split(new char[] { ',' })[0].Split(new char[] { '=' })[1], out dbkeys);                          dictdbIdxKeysNum[dbindex] = dbkeys;                          totalKeysSource += dbkeys;                          Console.WriteLine($"==t  {list.ToString()}");                     }                 }             }

第二阶段：复制

循环执行复制Redis Keys的子任务，SCAN列出所有Keys。
创建更多子任务以使用 StackExchange.Redis bacth 操作进行 TTL，验证Key是否过期，DUMP出Key的byte[]信息
使用批量操作将Key恢复到目标Redis
如果遇到异常，则将Key信息添加到失败队列中。
检查移动的keys的进度，同时检查失败的队列，如果不为空，将重新运行移动任务

 var allkeys = sourcecon.BasicRetryInfo((conn) => conn.GetServer(conn.GetEndPoints()[0]).Keys(dbindex).Skip(skipKeys).Take(takeKeys)).ToArray(); var sourcedb = sourcecon.GetConection().GetDatabase(dbindex);  var destdb = destcon.GetConection().GetDatabase(dbindex);   foreach (var keys in SplitKeys(allkeys))  {      var rbatch = sourcedb.CreateBatch();      var ttltask = new List<Task<TimeSpan?>>();      var dumptask = new List<Task<byte[]?>>();      foreach (var key in keys)      {          ttltask.Add(rbatch.KeyTimeToLiveAsync(key));           dumptask.Add(rbatch.KeyDumpAsync(key));      }      rbatch.Execute();       var ttlResults = Task.WhenAll(ttltask).Result;      var dumpkResults = Task.WhenAll(dumptask).Result;       //Restore the key to destation DB.      var destBatch = destdb.CreateBatch();       var i = 0;      foreach (var key in keys)      {          destBatch.KeyRestoreAsync(key, dumpkResults[i], ttlResults[i]);          i++;      }      destBatch.Execute();       //Random select one key to verify in Phase 3.       if (keys.Count() > 0)      {          int index = RandomNumberGenerator.GetInt32(keys.Count());          verifiedKeys.Add((dbindex, keys.ElementAt<RedisKey>(index).ToString()));      }        lock (lockObject)      {          totalKeysCopied += keys.Count();      }  }

第三阶段：验证

随机选取某个key, 一个一个的检查他们的值在两个Redis服务器之间是否相同

            foreach (var key in verifiedKeys)             {                 try                 {                     var sourdump = await sourcecon.BasicRetryInfo(async (sc) => sc.GetDatabase(key.Item1).KeyDumpAsync(key.Item2));                     var destdump = await destcon.BasicRetryInfo(async (sc) => sc.GetDatabase(key.Item1).KeyDumpAsync(key.Item2));                      if (!sourdump.Result.SequenceEqual(destdump.Result))                     {                         Console.Write($"n");                         Console.WriteLine($"== {key} Verify Failed");                     }                     else                     {                         Console.Write($"{key}, ");                     }                 }                 catch (Exception ex)                 {                     Console.BackgroundColor = ConsoleColor.Red;                     Console.WriteLine($"=={DateTime.Now.ToLocalTime()} Verify {key} failed ({ex.Message})");                     Console.BackgroundColor = ConsoleColor.Black;                 }             }