C# redis集群批量操作之slot計算出16384個字符串
引入一個大家都用的到的需求來說吧。
需求：要在三主三從的redis集群，存入數據，會對數據進行批量刪除操作，數據要求要在redis集群負載均衡。
思路：
1.存入數據好辦
1 var connect = ConnectionMultiplexer.Connect(redisConn); 2 var redisDb = connect.GetDatabase(); 3 var res1 = redisDb.StringSet("1", DateTime.Now.ToString(), TimeSpan.FromSeconds(600)); 4 var res2 = redisDb.StringSet("1111", DateTime.Now.ToString(), TimeSpan.FromSeconds(600));
2.批量刪除直接異常
1 redisDb.KeyDelete(new RedisKey[] { "1", "1111" });
Ex:“Multi-key operations must involve a single slot; keys can use ‘hash tags’ to help this, i.e. ‘{/users/12345}/account’ and ‘{/users/12345}/contacts’ will always be in the same slot”
3.查到異常是因為redis hash slot 機制導致的，什麼是 hash slot？
hash slot 介紹：//redis.io/topics/cluster-tutorial
4.加上hash slot 字符串，讓key進入同一個slot
var res1 = redisDb.StringSet(“{myslot}key1”, DateTime.Now.ToString(), TimeSpan.FromSeconds(600));
var res2 = redisDb.StringSet(“{myslot}key2”, DateTime.Now.ToString(), TimeSpan.FromSeconds(600));
redisDb.KeyDelete(new RedisKey[] { “{myslot}key1”, “{myslot}key2” });
能進行批量操作，但是都被分配到了同一台服務器上的同一個槽點，不負載均衡。
5.如何負載均衡， 讓Key分佈到各個服務器，並且可以批量操作？
如果知道每個槽點對應的字符串，key可以按照算法計算出自己對應的字符串，加上後，就可以進行分組批量增刪改操作。
6.hash slot 計算方法
HASH_SLOT = CRC16(key) mod 16384 （crc16-XMODEM）
介紹 ：//redis.io/topics/cluster-spec
7.net core 計算出16384個slot 字符串 算法例子和 結果模板
1 static void Main(string[] args) 2 { 3 var data = new Dictionary<int, string>(); 4 var i = 0; 5 while (data.Keys.Count < 16384) 6 { 7 var temp = i.ToString("X"); 8 var value = Crc16(Encoding.UTF8.GetBytes(temp)) % 16384; 9 data[int.Parse(value.ToString())] = temp; 10 i++; 11 12 } 13 var sb = new StringBuilder(); 14 foreach (var item in data.OrderBy(s => s.Key)) 15 { 16 var temp = $"slot num:{item.Key} str:{item.Value} \r\n"; 17 Console.WriteLine(temp); 18 sb.Append(temp); 19 } 20 File.WriteAllText("data.txt", sb.ToString()); 21 Console.ReadLine(); 22 } 23 private static ushort Crc16(byte[] bytes) 24 { 25 ushort poly = 0x1021; 26 ushort[] table = new ushort[256]; 27 ushort initialValue = 0x0; 28 ushort temp, a; 29 ushort crc = initialValue; 30 for (int i = 0; i < table.Length; ++i) 31 { 32 temp = 0; 33 a = (ushort)(i << 8); 34 for (int j = 0; j < 8; ++j) 35 { 36 if (((temp ^ a) & 0x8000) != 0) 37 temp = (ushort)((temp << 1) ^ poly); 38 else 39 temp <<= 1; 40 a <<= 1; 41 } 42 table[i] = temp; 43 } 44 for (int i = 0; i < bytes.Length; ++i) 45 { 46 crc = (ushort)((crc << 8) ^ table[((crc >> 8) ^ (0xff & bytes[i]))]); 47 } 48 return crc; 49 }
8.校驗算出來的字符串 對應 的slot位置 是否正確
9.批量設置和批量刪除方法
假定三主三從，那麼三台服務器，取九個slot字符串，這九個是均分的位置(均分利於集群擴展)。即16384/10=1638 1638是第一位，1638*2是第二位，以此類推取字符串
共九個[“1A73F”,”18B13″,”1AAD3″,”184FF”,”143BF”,”18413″,”17B8D”,”18BFF”,”1B1C4″]
先分組，再批量插入，再批量刪除
1 try 2 { 3 var redisConn = "{集群地址}"; 4 var connect = ConnectionMultiplexer.Connect(redisConn); 5 var redisDb = connect.GetDatabase(); 6 var res1 = redisDb.StringSet("1", DateTime.Now.ToString(), TimeSpan.FromSeconds(600)); 7 var res2 = redisDb.StringSet("1111", DateTime.Now.ToString(), TimeSpan.FromSeconds(600)); 8 redisDb.KeyDelete(new RedisKey[] { "1", "1111" }); 9 10 11 var redisSlotKeyList = new string[] { "1A73F", "18B13", "1AAD3", "184FF", "143BF", "18413", "17B8D", "18BFF", "1B1C4" }; 12 var userIdArray = new int[] { 1, 2, 3, 4, 5, 6, 7, 8, 9, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20 }; 13 //group 14 var dic = new Dictionary<int, Dictionary<RedisKey,RedisValue>>(); 15 foreach (var userId in userIdArray) 16 { 17 var index = userId % redisSlotKeyList.Length; 18 var slotKey = redisSlotKeyList[index]; 19 var redisKey = $"{{{slotKey}}}test_{userId}"; 20 Console.WriteLine($"{ redisKey} {userId}"); 21 if (dic.ContainsKey(index)) 22 { 23 dic[index].Add(redisKey, DateTime.Now.ToLongTimeString()); 24 } 25 else 26 { 27 dic[index] = new Dictionary<RedisKey, RedisValue> { { new RedisKey(redisKey), new RedisValue("values") } }; 28 } 29 } 30 31 //set 32 foreach (var item in dic) 33 { 34 var addRes = redisDb.StringSet(item.Value.ToArray()); 35 Console.WriteLine(addRes); 36 } 37 38 39 //delete 40 foreach (var item in dic) 41 { 42 var deleteRes = redisDb.KeyDelete(item.Value.Keys.ToArray()); 43 Console.WriteLine(deleteRes); 44 } 45 46 47 48 } 49 catch (Exception ex) 50 { 51 throw ex; 52 }
View Code
10.注意點
集群的分割slot配置不一定的均分的，提前先查看，命令：cluster nodes。 查看之後再根據實際情況取slot string
