C#实现请求唯一性校验支持高并发
- 2019 年 10 月 3 日
- 筆記
使用场景描述：
网络请求中经常会遇到发送的请求，服务端响应是成功的，但是返回的时候出现网络故障，导致客户端无法接收到请求结果，那么客户端程序可能判断为网络故障，而重复发送同一个请求。当然如果接口中定义了请求结果查询接口，那么这种重复会相对少一些。特别是交易类的数据，这种操作更是需要避免重复发送请求。另外一种情况是用户过于快速的点击界面按钮，产生连续的相同内容请求，那么后端也需要进行过滤，这种一般出现在系统对接上，无法去控制第三方系统的业务逻辑，需要从自身业务逻辑里面去限定。
其他需求描述：
这类请求一般存在时间范围和高并发的特点，就是短时间内会出现重复的请求，因此对模块需要支持高并发性。
技术实现：
对请求的业务内容进行MD5摘要，并且将MD5摘要存储到缓存中，每个请求数据都通过这个一个公共的调用的方法进行判断。
代码实现：
公共调用代码 UniqueCheck 采用单例模式创建唯一对象，便于在多线程调用的时候，只访问一个统一的缓存库
/* * volatile就像大家更熟悉的const一样，volatile是一个类型修饰符（type specifier）。 * 它是被设计用来修饰被不同线程访问和修改的变量。 * 如果没有volatile，基本上会导致这样的结果：要么无法编写多线程程序，要么编译器失去大量优化的机会。 */ private static readonly object lockHelper = new object(); private volatile static UniqueCheck _instance; /// <summary> /// 获取单一实例 /// </summary> /// <returns></returns> public static UniqueCheck GetInstance() { if (_instance == null) { lock (lockHelper) { if (_instance == null) _instance = new UniqueCheck(); } } return _instance; }
这里需要注意volatile的修饰符，在实际测试过程中，如果没有此修饰符，在高并发的情况下会出现报错。
自定义一个可以进行并发处理队列，代码如下：ConcurrentLinkedQueue
1 using System; 2 using System.Collections.Generic; 3 using System.Text; 4 using System.Threading; 5 6 namespace PackgeUniqueCheck 7 { 8 /// <summary> 9 /// 非加锁并发队列，处理100个并发数以内 10 /// </summary> 11 /// <typeparam name="T"></typeparam> 12 public class ConcurrentLinkedQueue<T> 13 { 14 private class Node<K> 15 { 16 internal K Item; 17 internal Node<K> Next; 18 19 public Node(K item, Node<K> next) 20 { 21 this.Item = item; 22 this.Next = next; 23 } 24 } 25 26 private Node<T> _head; 27 private Node<T> _tail; 28 29 public ConcurrentLinkedQueue() 30 { 31 _head = new Node<T>(default(T), null); 32 _tail = _head; 33 } 34 35 public bool IsEmpty 36 { 37 get { return (_head.Next == null); } 38 } 39 /// <summary> 40 /// 进入队列 41 /// </summary> 42 /// <param name="item"></param> 43 public void Enqueue(T item) 44 { 45 Node<T> newNode = new Node<T>(item, null); 46 while (true) 47 { 48 Node<T> curTail = _tail; 49 Node<T> residue = curTail.Next; 50 51 //判断_tail是否被其他process改变 52 if (curTail == _tail) 53 { 54 //A 有其他process执行C成功，_tail应该指向新的节点 55 if (residue == null) 56 { 57 //C 其他process改变了tail节点，需要重新取tail节点 58 if (Interlocked.CompareExchange<Node<T>>( 59 ref curTail.Next, newNode, residue) == residue) 60 { 61 //D 尝试修改tail 62 Interlocked.CompareExchange<Node<T>>(ref _tail, newNode, curTail); 63 return; 64 } 65 } 66 else 67 { 68 //B 帮助其他线程完成D操作 69 Interlocked.CompareExchange<Node<T>>(ref _tail, residue, curTail); 70 } 71 } 72 } 73 } 74 /// <summary> 75 /// 队列取数据 76 /// </summary> 77 /// <param name="result"></param> 78 /// <returns></returns> 79 public bool TryDequeue(out T result) 80 { 81 Node<T> curHead; 82 Node<T> curTail; 83 Node<T> next; 84 while (true) 85 { 86 curHead = _head; 87 curTail = _tail; 88 next = curHead.Next; 89 if (curHead == _head) 90 { 91 if (next == null) //Queue为空 92 { 93 result = default(T); 94 return false; 95 } 96 if (curHead == curTail) //Queue处于Enqueue第一个node的过程中 97 { 98 //尝试帮助其他Process完成操作 99 Interlocked.CompareExchange<Node<T>>(ref _tail, next, curTail); 100 } 101 else 102 { 103 //取next.Item必须放到CAS之前 104 result = next.Item; 105 //如果_head没有发生改变，则将_head指向next并退出 106 if (Interlocked.CompareExchange<Node<T>>(ref _head, 107 next, curHead) == curHead) 108 break; 109 } 110 } 111 } 112 return true; 113 } 114 /// <summary> 115 /// 尝试获取最后一个对象 116 /// </summary> 117 /// <param name="result"></param> 118 /// <returns></returns> 119 public bool TryGetTail(out T result) 120 { 121 result = default(T); 122 if (_tail == null) 123 { 124 return false; 125 } 126 result = _tail.Item; 127 return true; 128 } 129 } 130 }
虽然是一个非常简单的唯一性校验逻辑，但是要做到高效率，高并发支持，高可靠性，以及低内存占用，需要实现这样的需求，需要做细致的模拟测试。
1 using System; 2 using System.Collections.Generic; 3 using System.Text; 4 using System.Threading; 5 using System.Collections; 6 7 namespace PackgeUniqueCheck 8 { 9 public class UniqueCheck 10 { 11 /* 12 * volatile就像大家更熟悉的const一样，volatile是一个类型修饰符（type specifier）。 13 * 它是被设计用来修饰被不同线程访问和修改的变量。 14 * 如果没有volatile，基本上会导致这样的结果：要么无法编写多线程程序，要么编译器失去大量优化的机会。 15 */ 16 private static readonly object lockHelper = new object(); 17 18 private volatile static UniqueCheck _instance; 19 20 /// <summary> 21 /// 获取单一实例 22 /// </summary> 23 /// <returns></returns> 24 public static UniqueCheck GetInstance() 25 { 26 if (_instance == null) 27 { 28 lock (lockHelper) 29 { 30 if (_instance == null) 31 _instance = new UniqueCheck(); 32 } 33 } 34 return _instance; 35 } 36 37 private UniqueCheck() 38 { 39 //创建一个线程安全的哈希表，作为字典缓存 40 _DataKey = Hashtable.Synchronized(new Hashtable()); 41 Queue myqueue = new Queue(); 42 _DataQueue = Queue.Synchronized(myqueue); 43 _Myqueue = new ConcurrentLinkedQueue<string>(); 44 _Timer = new Thread(DoTicket); 45 _Timer.Start(); 46 } 47 48 #region 公共属性设置 49 /// <summary> 50 /// 设定定时线程的休眠时间长度：默认为1分钟 51 /// 时间范围：1-7200000,值为1毫秒到2小时 52 /// </summary> 53 /// <param name="value"></param> 54 public void SetTimeSpan(int value) 55 { 56 if (value > 0&& value <=7200000) 57 { 58 _TimeSpan = value; 59 } 60 } 61 /// <summary> 62 /// 设定缓存Cache中的最大记录条数 63 /// 值范围：1-5000000,1到500万 64 /// </summary> 65 /// <param name="value"></param> 66 public void SetCacheMaxNum(int value) 67 { 68 if (value > 0 && value <= 5000000) 69 { 70 _CacheMaxNum = value; 71 } 72 } 73 /// <summary> 74 /// 设置是否在控制台中显示日志 75 /// </summary> 76 /// <param name="value"></param> 77 public void SetIsShowMsg(bool value) 78 { 79 Helper.IsShowMsg = value; 80 } 81 /// <summary> 82 /// 线程请求阻塞增量 83 /// 值范围：1-CacheMaxNum，建议设置为缓存最大值的10%-20% 84 /// </summary> 85 /// <param name="value"></param> 86 public void SetBlockNumExt(int value) 87 { 88 if (value > 0 && value <= _CacheMaxNum) 89 { 90 _BlockNumExt = value; 91 } 92 } 93 /// <summary> 94 /// 请求阻塞时间 95 /// 值范围：1-max，根据阻塞增量设置请求阻塞时间 96 /// 阻塞时间越长，阻塞增量可以设置越大，但是请求实时响应就越差 97 /// </summary> 98 /// <param name="value"></param> 99 public void SetBlockSpanTime(int value) 100 { 101 if (value > 0) 102 { 103 _BlockSpanTime = value; 104 } 105 } 106 #endregion 107 108 #region 私有变量 109 /// <summary> 110 /// 内部运行线程 111 /// </summary> 112 private Thread _runner = null; 113 /// <summary> 114 /// 可处理高并发的队列 115 /// </summary> 116 private ConcurrentLinkedQueue<string> _Myqueue = null; 117 /// <summary> 118 /// 唯一内容的时间健值对 119 /// </summary> 120 private Hashtable _DataKey = null; 121 /// <summary> 122 /// 内容时间队列 123 /// </summary> 124 private Queue _DataQueue = null; 125 /// <summary> 126 /// 定时线程的休眠时间长度：默认为1分钟 127 /// </summary> 128 private int _TimeSpan = 3000; 129 /// <summary> 130 /// 定时计时器线程 131 /// </summary> 132 private Thread _Timer = null; 133 /// <summary> 134 /// 缓存Cache中的最大记录条数 135 /// </summary> 136 private int _CacheMaxNum = 500000; 137 /// <summary> 138 /// 线程请求阻塞增量 139 /// </summary> 140 private int _BlockNumExt = 10000; 141 /// <summary> 142 /// 请求阻塞时间 143 /// </summary> 144 private int _BlockSpanTime = 100; 145 #endregion 146 147 #region 私有方法 148 private void StartRun() 149 { 150 _runner = new Thread(DoAction); 151 _runner.Start(); 152 Helper.ShowMsg("内部线程启动成功！"); 153 } 154 155 private string GetItem() 156 { 157 string tp = string.Empty; 158 bool result = _Myqueue.TryDequeue(out tp); 159 return tp; 160 } 161 /// <summary> 162 /// 执行循环操作 163 /// </summary> 164 private void DoAction() 165 { 166 while (true) 167 { 168 while (!_Myqueue.IsEmpty) 169 { 170 string item = GetItem(); 171 _DataQueue.Enqueue(item); 172 if (!_DataKey.ContainsKey(item)) 173 { 174 _DataKey.Add(item, DateTime.Now); 175 } 176 } 177 //Helper.ShowMsg("当前数组已经为空，处理线程进入休眠状态..."); 178 Thread.Sleep(2); 179 } 180 } 181 /// <summary> 182 /// 执行定时器的动作 183 /// </summary> 184 private void DoTicket() 185 { 186 while (true) 187 { 188 Helper.ShowMsg("当前数据队列个数：" + _DataQueue.Count.ToString()); 189 if (_DataQueue.Count > _CacheMaxNum) 190 { 191 while (true) 192 { 193 Helper.ShowMsg(string.Format("当前队列数：{0}，已经超出最大长度：{1},开始进行清理操作...", _DataQueue.Count, _CacheMaxNum.ToString())); 194 string item = _DataQueue.Dequeue().ToString(); 195 if (!string.IsNullOrEmpty(item)) 196 { 197 if (_DataKey.ContainsKey(item)) 198 { 199 _DataKey.Remove(item); 200 } 201 if (_DataQueue.Count <= _CacheMaxNum) 202 { 203 Helper.ShowMsg("清理完成，开始休眠清理线程..."); 204 break; 205 } 206 } 207 } 208 } 209 Thread.Sleep(_TimeSpan); 210 } 211 } 212 213 /// <summary> 214 /// 线程进行睡眠等待 215 /// 如果当前负载压力大大超出了线程的处理能力 216 /// 那么需要进行延时调用 217 /// </summary> 218 private void BlockThread() 219 { 220 if (_DataQueue.Count > _CacheMaxNum + _BlockNumExt) 221 { 222 Thread.Sleep(_BlockSpanTime); 223 } 224 } 225 #endregion 226 227 #region 公共方法 228 /// <summary> 229 /// 开启服务线程 230 /// </summary> 231 public void Start() 232 { 233 if (_runner == null) 234 { 235 StartRun(); 236 } 237 else 238 { 239 if (_runner.IsAlive == false) 240 { 241 StartRun(); 242 } 243 } 244 245 } 246 /// <summary> 247 /// 关闭服务线程 248 /// </summary> 249 public void Stop() 250 { 251 if (_runner != null) 252 { 253 _runner.Abort(); 254 _runner = null; 255 } 256 } 257 258 /// <summary> 259 /// 添加内容信息 260 /// </summary> 261 /// <param name="item">内容信息</param> 262 /// <returns>true:缓存中不包含此值，队列添加成功，false:缓存中包含此值，队列添加失败</returns> 263 public bool AddItem(string item) 264 { 265 BlockThread(); 266 item = Helper.MakeMd5(item); 267 if (_DataKey.ContainsKey(item)) 268 { 269 return false; 270 } 271 else 272 { 273 _Myqueue.Enqueue(item); 274 return true; 275 } 276 } 277 /// <summary> 278 /// 判断内容信息是否已经存在 279 /// </summary> 280 /// <param name="item">内容信息</param> 281 /// <returns>true：信息已经存在于缓存中，false：信息不存在于缓存中</returns> 282 public bool CheckItem(string item) 283 { 284 item = Helper.MakeMd5(item); 285 return _DataKey.ContainsKey(item); 286 } 287 #endregion 288 289 } 290 }
模拟测试代码：
private static string _example = Guid.NewGuid().ToString(); private static UniqueCheck _uck = null; static void Main(string[] args) { _uck = UniqueCheck.GetInstance(); _uck.Start(); _uck.SetIsShowMsg(false); _uck.SetCacheMaxNum(20000000); _uck.SetBlockNumExt(1000000); _uck.SetTimeSpan(6000); _uck.AddItem(_example); Thread[] threads = new Thread[20]; for (int i = 0; i < 20; i++) { threads[i] = new Thread(AddInfo); threads[i].Start(); } Thread checkthread = new Thread(CheckInfo); checkthread.Start(); string value = Console.ReadLine(); checkthread.Abort(); for (int i = 0; i < 50; i++) { threads[i].Abort(); } _uck.Stop(); } static void AddInfo() { while (true) { _uck.AddItem(Guid.NewGuid().ToString()); } } static void CheckInfo() { while (true) { Console.WriteLine("开始时间：{0}...", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.ffff")); Console.WriteLine("插入结果：{0}", _uck.AddItem(_example)); Console.WriteLine("结束时间：{0}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.ffff"));
//调整进程休眠时间，可以测试高并发的情况 //Thread.Sleep(1000); } }
测试截图：
