一起來看CORE源碼(一) ConcurrentDictionary
先貼源碼地址
https://github.com/dotnet/corefx/blob/master/src/System.Collections.Concurrent/src/System/Collections/Concurrent/ConcurrentDictionary.cs
.NET CORE很大一個好處就是代碼的開源,你可以詳細的查看你使用類的源代碼,並學習微軟的寫法和實現思路。
這裏我對.net core中ConcurrentDictionary源碼進行了分析,裏面采用了Volatile.Read和write(volatile作用:確保本條指令不會因編譯器的優化而省略,且要求每次直接從內存地址讀值,而不走寄存器),然後也使用了lock這種混合鎖,而且還定義了更細顆粒度的鎖。所以多線程使用ConcurrentDictionary集合還是比較好的選擇。
本來想把本篇放到我的《C#異步編程系列》,不過後來感覺那個系列寫的已經算是收尾了,而且以後還會有寫更多core源碼分析的文字,所以就單獨新增一個系列把。
ConcurrentDictionary內部私有類
先上源碼,再仔細聊
/// <summary> /// Tables that hold the internal state of the ConcurrentDictionary /// /// Wrapping the three tables in a single object allows us to atomically /// replace all tables at once./// </summary> private sealed class Tables { // A singly-linked list for each bucket. // 單鏈表數據結構的桶,裏面的節點就是對應字典值 internal readonly Node[] _buckets; // A set of locks, each guarding a section of the table. //鎖的數組 internal readonly object[] _locks; // The number of elements guarded by each lock.internal volatile int[] _countPerLock; internal Tables(Node[] buckets, object[] locks, int[] countPerLock) { _buckets = buckets; _locks = locks; _countPerLock = countPerLock; } } /// <summary> /// A node in a singly-linked list representing a particular hash table bucket. /// 由Dictionary裏的Entry改成Node,並且把next放到Node裏 /// </summary> private sealed class Node { internal readonly TKey _key; internal TValue _value; internal volatile Node _next; internal readonly int _hashcode; internal Node(TKey key, TValue value, int hashcode, Node next) { _key = key; _value = value; _next = next; _hashcode = hashcode; } } private volatile Tables _tables; // Internal tables of the dictionary private IEqualityComparer<TKey> _comparer; // Key equality comparer // The maximum number of elements per lock before a resize operation is triggered // 每個鎖對應的元素最大個數,如果超過,要重新進行resize tables private int _budget;
首先,內部類定義為私有且密封,這樣就保證了無法從外部進行篡改,而且註意volatile關鍵字的使用,這確保了我們多線程操作的時候,最終都是去內存中讀取對應地址的值和操作對應地址的值。
internal readonly object[] _locks; internal volatile int[] _countPerLock;
以上兩個類是為了高性能及並發鎖所建立的對象,實際方法上鎖時,使用如下語句
lock (tables._locks[lockNo]) Monitor.Enter(tables._locks[lockNo], ref lockTaken);
以上兩種調用方式是等價的,都會阻塞執行,直到獲取到鎖(對於Monitor我很多時候會盡可能使用TryEnter,畢竟不阻塞,不過這個類的實現一定要使用阻塞式的,這樣程序邏輯才能繼續往下走。更多關於Monitor我在 《C#異步編程(四)混合模式線程同步》裏面有詳細介紹)
這樣,實現了顆粒化到每個單獨的鍵值的鎖,最大限度的保證了並發。
這裏lockNo參數是通過GetBucketAndLockNo方法獲取的,方法通過out變量返回值。
/// <summary> /// Computes the bucket and lock number for a particular key. ///這裏獲取桶的索引和鎖的索引,註意,鎖的索引和桶未必是同一個值。 /// </summary> private static void GetBucketAndLockNo(int hashcode, out int bucketNo, out int lockNo, int bucketCount, int lockCount) { bucketNo = (hashcode & 0x7fffffff) % bucketCount; lockNo = bucketNo % lockCount; }
上面方法中
hashcode 是通過private IEqualityComparer<TKey> _comparer對象的GetHashCode方法通過key獲取到的。
bucketCount是整個table的長度。
lockCount是現有的鎖的數組
TryAdd方法
我們從最簡單的TryAdd方法開始介紹,這裏ConcurrentDictionary類的封裝非常合理,暴露出來的方法,很多是通過統一的內部方法進行執行,比如更新刪除等操作等,都有類內部唯一的私有方法進行執行,然後通過向外暴漏各種參數不同的方法,來實現不同行為。
public bool TryAdd(TKey key, TValue value) { if (key == null) ThrowKeyNullException(); TValue dummy; return TryAddInternal(key, _comparer.GetHashCode(key), value, false, true, out dummy); }
上面TryAddInternal的參數對應如下
/// <summary> /// Shared internal implementation for inserts and updates. /// If key exists, we always return false; and if updateIfExists == true we force update with value; /// If key doesn‘t exist, we always add value and return true; /// </summary> private bool TryAddInternal(TKey key, int hashcode, TValue value, bool updateIfExists, bool acquireLock, out TValue resultingValue)
也就說說,updateIfExists為false,存在值的情況下,TryAdd不會更新原有值,而是直接返回false。我的多線程並發寫庫就是利用了這個特性,這個案例我會在本文最後介紹。現在我們來看TryAddInternal內部,廢話不多說,上源碼(大部分都註釋過了,所以直接閱讀即可)
//while包在外面,為了continue,如果發生了_tables私有變量在操作過程被其他線程修改的情況 while (true) { int bucketNo, lockNo; //變量復制到方法本地變量 判斷tables是否在操作過程中被其他線程修改。 Tables tables = _tables; //提到過的獲取桶的索引和鎖的索引 GetBucketAndLockNo(hashcode, out bucketNo, out lockNo, tables._buckets.Length, tables._locks.Length); //是否要擴大tables bool resizeDesired = false; //是否成功獲取鎖,成功的話會在final塊中進行退出 bool lockTaken = false; try { if (acquireLock) Monitor.Enter(tables._locks[lockNo], ref lockTaken); // If the table just got resized, we may not be holding the right lock, and must retry. // This should be a rare occurrence. if (tables != _tables) { continue; } // Try to find this key in the bucket Node prev = null; //這裏如果找到對應地址為空,會直接跳出循環,說明對應的key沒有添加鍋 //不為空的時候,會進行返回false(具體是否更新根據updateIfExists)(當然也存在會有相同_hashcode值的情況,所以還要對key進行判定,key不同,繼續往後找,直到找到相同key) for (Node node = tables._buckets[bucketNo]; node != null; node = node._next) { Debug.Assert((prev == null && node == tables._buckets[bucketNo]) || prev._next == node); //對hashcode和key進行判定,確保找到的就是要更新的 if (hashcode == node._hashcode && _comparer.Equals(node._key, key)) { // The key was found in the dictionary. If updates are allowed, update the value for that key. // We need to create a new node for the update, in order to support TValue types that cannot // be written atomically, since lock-free reads may be happening concurrently. if (updateIfExists) { if (s_isValueWriteAtomic) { node._value = value; } else { Node newNode = new Node(node._key, value, hashcode, node._next); if (prev == null) { Volatile.Write(ref tables._buckets[bucketNo], newNode); } else { prev._next = newNode; } } resultingValue = value; } else { resultingValue = node._value; } return false; } prev = node; } // The key was not found in the bucket. Insert the key-value pair. Volatile.Write<Node>(ref tables._buckets[bucketNo], new Node(key, value, hashcode, tables._buckets[bucketNo])); checked { tables._countPerLock[lockNo]++; } // // If the number of elements guarded by this lock has exceeded the budget, resize the bucket table. // It is also possible that GrowTable will increase the budget but won‘t resize the bucket table. // That happens if the bucket table is found to be poorly utilized due to a bad hash function. // if (tables._countPerLock[lockNo] > _budget) { resizeDesired = true; } } finally { if (lockTaken) Monitor.Exit(tables._locks[lockNo]); } // // The fact that we got here means that we just performed an insertion. If necessary, we will grow the table. // // Concurrency notes: // - Notice that we are not holding any locks at when calling GrowTable. This is necessary to prevent deadlocks. // - As a result, it is possible that GrowTable will be called unnecessarily. But, GrowTable will obtain lock 0 // and then verify that the table we passed to it as the argument is still the current table. // if (resizeDesired) { GrowTable(tables); } resultingValue = value; return true; }
ContainsKey和TryGetValue
ContainsKey和TryGetValue其實內部最後調用的都是私有TryGetValueInternal,這裏ContainsKey調用TryGetValue。
ContainsKey方法
/// <summary> /// Determines whether the ConcurrentDictionary{TKey, TValue} contains the specified key. /// </summary> /// <param name="key">The key to locate in the</param> /// <returns>true if the ConcurrentDictionary{TKey, TValue} contains an element withthe specified key; otherwise, false.</returns> public bool ContainsKey(TKey key) { if (key == null) ThrowKeyNullException(); TValue throwAwayValue; return TryGetValue(key, out throwAwayValue); }
TryGetValue方法
/// <summary> /// Attempts to get the value associated with the specified key from the ConcurrentDictionary{TKey,TValue}. /// </summary> /// <param name="key">The key of the value to get.</param> /// <param name="value">When this method returns, <paramref name="value"/> contains the object from /// the ConcurrentDictionary{TKey,TValue} with the specified key or the default value of /// <returns>true if the key was found in the <see cref="ConcurrentDictionary{TKey,TValue}"/>; /// otherwise, false.</returns> public bool TryGetValue(TKey key, out TValue value) { if (key == null) ThrowKeyNullException(); return TryGetValueInternal(key, _comparer.GetHashCode(key), out value); }
TryGetValueInternal方法
private bool TryGetValueInternal(TKey key, int hashcode, out TValue value) { //用本地變量保存這個table的快照。 // We must capture the _buckets field in a local variable. It is set to a new table on each table resize. Tables tables = _tables; //獲取key對應的桶位置 int bucketNo = GetBucket(hashcode, tables._buckets.Length); // We can get away w/out a lock here. // The Volatile.Read ensures that we have a copy of the reference to tables._buckets[bucketNo]. // This protects us from reading fields (‘_hashcode‘, ‘_key‘, ‘_value‘ and ‘_next‘) of different instances. Node n = Volatile.Read<Node>(ref tables._buckets[bucketNo]); //如果key相符 ,賦值,不然繼續尋找下一個。 while (n != null) { if (hashcode == n._hashcode && _comparer.Equals(n._key, key)) { value = n._value; return true; } n = n._next; } value = default(TValue);//沒找到就賦默認值 return false; }
TryRemove
TryRemove方法
public bool TryRemove(TKey key, out TValue value) { if (key == null) ThrowKeyNullException(); return TryRemoveInternal(key, out value, false, default(TValue)); }
這個方法會調用內部私用的TryRemoveInternal
/// <summary> /// Removes the specified key from the dictionary if it exists and returns its associated value. /// If matchValue flag is set, the key will be removed only if is associated with a particular /// value. /// </summary> /// <param name="key">The key to search for and remove if it exists.</param> /// <param name="value">The variable into which the removed value, if found, is stored.</param> /// <param name="matchValue">Whether removal of the key is conditional on its value.</param> /// <param name="oldValue">The conditional value to compare against if <paramref name="matchValue"/> is true</param> /// <returns></returns> private bool TryRemoveInternal(TKey key, out TValue value, bool matchValue, TValue oldValue) { int hashcode = _comparer.GetHashCode(key); while (true) { Tables tables = _tables; int bucketNo, lockNo; //這裏獲取桶的索引和鎖的索引,註意,鎖的索引和桶未必是同一個值,具體算法看源碼。 GetBucketAndLockNo(hashcode, out bucketNo, out lockNo, tables._buckets.Length, tables._locks.Length); //這裏鎖住的只是對應這個index指向的鎖,而不是所有鎖。 lock (tables._locks[lockNo]) { //這裏table可能被重新分配,所以這裏再次獲取,看得到的是不是同一個table // If the table just got resized, we may not be holding the right lock, and must retry. // This should be a rare occurrence. if (tables != _tables) { continue; } Node prev = null; //這裏同一個桶,可能因為連地址,有很多值,所以要對比key for (Node curr = tables._buckets[bucketNo]; curr != null; curr = curr._next) { Debug.Assert((prev == null && curr == tables._buckets[bucketNo]) || prev._next == curr); //對比是不是要刪除的的那個元素 if (hashcode == curr._hashcode && _comparer.Equals(curr._key, key)) { if (matchValue) { bool valuesMatch = EqualityComparer<TValue>.Default.Equals(oldValue, curr._value); if (!valuesMatch) { value = default(TValue); return false; } } //執行刪除,判斷有沒有上一個節點。然後修改節點指針或地址。 if (prev == null) { Volatile.Write<Node>(ref tables._buckets[bucketNo], curr._next); } else { prev._next = curr._next; } value = curr._value; tables._countPerLock[lockNo]--; return true; } prev = curr; } } value = default(TValue); return false; } }
我的使用實例
之前做項目時候,有個奇怪的場景,就是打電話的時候回調接口保存通話記錄,這裏通過CallId來唯一識別每次通話,但是前端程序是通過websocket跟通話服務建立連接(通話服務是另外一個公司做的)。客戶是呼叫中心,一般在網頁端都是多個頁面操作,所以會有多個websocket連接,這時候每次通話,每個頁面都會回調接口端,保存相同的通話記錄,並發是同一時間的。
我們最早考慮使用消息隊列來過濾重復的請求,但是我仔細考慮了下,發現使用ConcurrentDictionary方式的實現更簡單,具體實現如下(我精簡了下代碼):
private static ConcurrentDictionary<string,string> _strDic=new ConcurrentDictionary<string, string>(); public async Task<BaseResponse> AddUserByAccount(string callId) { if ( _strDic.ContainsKey(callId)) { return BaseResponse.GetBaseResponse(BusinessStatusType.Failed,"鍵值已存在"); } //成功寫入 if (_strDic.TryAdd(callId,callId)) { var recordExist =await _userRepository.FirstOrDefaultAsync(c => c.CallId == callId); if (recordExist ==null) { Record record=new Record { CallId = callId, ………… ………… IsVerify=1 }; _userRepository.Insert(record); _userRepository.SaveChanges(); } return BaseResponse.GetBaseResponse(BusinessStatusType.OK); } //嘗試競爭線程,寫入失敗 return BaseResponse.GetBaseResponse(BusinessStatusType.Failed,"寫入失敗"); }
這裏如果進行同時的並發請求,最後請求都可以通過if ( _strDic.ContainsKey(callId))的判定,因為所有線程同時讀取,都是未寫入狀態。但是多個線程會在TryAdd時有競爭,而且ConcurrentDictionary的實現保證了只有一個線程可以成功更新,其他的都返回失敗。
一起來看CORE源碼(一) ConcurrentDictionary