?2016-2026 Matools All rights reserved,
粵ICP備17059708號
使用場景描述：
網(wǎng)絡請求中經(jīng)常會遇到發(fā)送的請求，服務端響應是成功的，但是返回的時候出現(xiàn)網(wǎng)絡故障，導致客戶端無法接收到請求結果，那么客戶端程序可能判斷為網(wǎng)絡故障，而重復發(fā)送同一個請求。當然如果接口中定義了請求結果查詢接口，那么這種重復會相對少一些。特別是交易類的數(shù)據(jù)，這種操作更是需要避免重復發(fā)送請求。另外一種情況是用戶過于快速的點擊界面按鈕，產(chǎn)生連續(xù)的相同內(nèi)容請求，那么后端也需要進行過濾，這種一般出現(xiàn)在系統(tǒng)對接上，無法去控制第三方系統(tǒng)的業(yè)務邏輯，需要從自身業(yè)務邏輯里面去限定。
其他需求描述：
這類請求一般存在時間范圍和高并發(fā)的特點，就是短時間內(nèi)會出現(xiàn)重復的請求，因此對模塊需要支持高并發(fā)性。
技術實現(xiàn)：
對請求的業(yè)務內(nèi)容進行MD5摘要，并且將MD5摘要存儲到緩存中，每個請求數(shù)據(jù)都通過這個一個公共的調用的方法進行判斷。
代碼實現(xiàn)：
公共調用代碼 UniqueCheck 采用單例模式創(chuàng)建唯一對象，便于在多線程調用的時候，只訪問一個統(tǒng)一的緩存庫
/* * volatile就像大家更熟悉的const一樣，volatile是一個類型修飾符（type specifier）。 *
它是被設計用來修飾被不同線程訪問和修改的變量。 * 如果沒有volatile，基本上會導致這樣的結果：要么無法編寫多線程程序，要么編譯器失去大量優(yōu)化的機會。
*/ 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的修飾符，在實際測試過程中，如果沒有此修飾符，在高并發(fā)的情況下會出現(xiàn)報錯。
自定義一個可以進行并發(fā)處理隊列，代碼如下：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 /// 非加鎖并發(fā)隊列，處理100個并發(fā)數(shù)以內(nèi) 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執(zhí)行C成功，_tail應該指向新的節(jié)點 55 if (residue == null) 56 { 57 //C
其他process改變了tail節(jié)點，需要重新取tail節(jié)點 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 /// 隊列取數(shù)據(jù) 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沒有發(fā)生改變，則將_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 }
雖然是一個非常簡單的唯一性校驗邏輯，但是要做到高效率，高并發(fā)支持，高可靠性，以及低內(nèi)存占用，需要實現(xiàn)這樣的需求，需要做細致的模擬測試。
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，基本上會導致這樣的結果：要么無法編寫多線程程序，要么編譯器失去大量優(yōu)化的機會。 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 //創(chuàng)建一個線程安全的哈希表，作為字典緩存 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中的最大記錄條數(shù) 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，根據(jù)阻塞增量設置請求阻塞時間 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 /// 內(nèi)部運行線程 111 ///
</summary> 112 private Thread _runner = null; 113 /// <summary> 114 ///
可處理高并發(fā)的隊列115 /// </summary> 116 private ConcurrentLinkedQueue<string> _Myqueue =
null; 117 /// <summary> 118 /// 唯一內(nèi)容的時間健值對 119 /// </summary> 120 private
Hashtable _DataKey =null; 121 /// <summary> 122 /// 內(nèi)容時間隊列 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中的最大記錄條數(shù) 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("內(nèi)部線程啟動成功！"); 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 ///
執(zhí)行循環(huán)操作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("當前數(shù)組已經(jīng)為空，處理線程進入休眠狀態(tài)..."); 178 Thread.Sleep(2); 179 } 180 } 181
/// <summary> 182 /// 執(zhí)行定時器的動作 183 /// </summary> 184 private void DoTicket()
185 { 186 while (true) 187 { 188 Helper.ShowMsg("當前數(shù)據(jù)隊列個數(shù)：" +
_DataQueue.Count.ToString());189 if (_DataQueue.Count > _CacheMaxNum) 190 { 191
while (true) 192 { 193 Helper.ShowMsg(string.Format("
當前隊列數(shù)：{0}，已經(jīng)超出最大長度：{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 ///
添加內(nèi)容信息260 /// </summary> 261 /// <param name="item">內(nèi)容信息</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 /// 判斷內(nèi)容信息是否已經(jīng)存在 279 /// </summary> 280 /// <param name="item">內(nèi)容信息</param>
281 /// <returns>true：信息已經(jīng)存在于緩存中，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"
));
//調整進程休眠時間，可以測試高并發(fā)的情況 //Thread.Sleep(1000); } }
測試截圖：
?
熱門工具 換一換