C# ConcurrentBag的實(shí)現(xiàn)原理

 public class ConcurrentBag<T> : IProducerConsumerCollection<T>, IReadOnlyCollection<T>    {        // ThreadLocalList對(duì)象包含每個(gè)線程的數(shù)據(jù)        ThreadLocal<ThreadLocalList> m_locals;         // 這個(gè)頭指針和尾指針指向中的第一個(gè)和最后一個(gè)本地列表，這些本地列表分散在不同線程中        // 允許在線程局部對(duì)象上枚舉        volatile ThreadLocalList m_headList, m_tailList;         // 這個(gè)標(biāo)志是告知操作線程必須同步操作        // 在GlobalListsLock 鎖中 設(shè)置        bool m_needSync;
}

[Serializable]// 構(gòu)造了雙向鏈表的節(jié)點(diǎn)internal class Node{    public Node(T value)    {        m_value = value;    }    public readonly T m_value;    public Node m_next;    public Node m_prev;}/// <summary>/// 集合操作類型/// </summary>internal enum ListOperation{    None,    Add,    Take};/// <summary>/// 線程鎖定的類/// </summary>internal class ThreadLocalList{    // 雙向鏈表的頭結(jié)點(diǎn) 如果為null那么表示鏈表為空    internal volatile Node m_head;    // 雙向鏈表的尾節(jié)點(diǎn)    private volatile Node m_tail;    // 定義當(dāng)前對(duì)List進(jìn)行操作的種類     // 與前面的 ListOperation 相對(duì)應(yīng)    internal volatile int m_currentOp;    // 這個(gè)列表元素的計(jì)數(shù)    private int m_count;    // The stealing count    // 這個(gè)不是特別理解 好像是在本地列表中 刪除某個(gè)Node 以后的計(jì)數(shù)    internal int m_stealCount;    // 下一個(gè)列表 可能會(huì)在其它線程中    internal volatile ThreadLocalList m_nextList;    // 設(shè)定鎖定是否已進(jìn)行    internal bool m_lockTaken;    // The owner thread for this list    internal Thread m_ownerThread;    // 列表的版本，只有當(dāng)列表從空變?yōu)榉强战y(tǒng)計(jì)是底層    internal volatile int m_version;    /// <summary>    /// ThreadLocalList 構(gòu)造器    /// </summary>    /// <param name="ownerThread">擁有這個(gè)集合的線程</param>    internal ThreadLocalList(Thread ownerThread)    {        m_ownerThread = ownerThread;    }    /// <summary>    /// 添加一個(gè)新的item到鏈表首部    /// </summary>    /// <param name="item">The item to add.</param>    /// <param name="updateCount">是否更新計(jì)數(shù).</param>    internal void Add(T item, bool updateCount)    {        checked        {            m_count++;        }        Node node = new Node(item);        if (m_head == null)        {            Debug.Assert(m_tail == null);            m_head = node;            m_tail = node;            m_version++; // 因?yàn)檫M(jìn)行初始化了，所以將空狀態(tài)改為非空狀態(tài)        }        else        {            // 使用頭插法 將新的元素插入鏈表            node.m_next = m_head;            m_head.m_prev = node;            m_head = node;        }        if (updateCount) // 更新計(jì)數(shù)以避免此添加同步時(shí)溢出        {            m_count = m_count - m_stealCount;            m_stealCount = 0;        }    }    /// <summary>    /// 從列表的頭部刪除一個(gè)item    /// </summary>    /// <param name="result">The removed item</param>    internal void Remove(out T result)    {        // 雙向鏈表刪除頭結(jié)點(diǎn)數(shù)據(jù)的流程        Debug.Assert(m_head != null);        Node head = m_head;        m_head = m_head.m_next;        if (m_head != null)        {            m_head.m_prev = null;        }        else        {            m_tail = null;        }        m_count--;        result = head.m_value;
    }    /// <summary>    /// 返回列表頭部的元素    /// </summary>    /// <param name="result">the peeked item</param>    /// <returns>True if succeeded, false otherwise</returns>    internal bool Peek(out T result)    {        Node head = m_head;        if (head != null)        {            result = head.m_value;            return true;        }        result = default(T);        return false;    }    /// <summary>    /// 從列表的尾部獲取一個(gè)item    /// </summary>    /// <param name="result">the removed item</param>    /// <param name="remove">remove or peek flag</param>    internal void Steal(out T result, bool remove)    {        Node tail = m_tail;        Debug.Assert(tail != null);        if (remove) // Take operation        {            m_tail = m_tail.m_prev;            if (m_tail != null)            {                m_tail.m_next = null;            }            else            {                m_head = null;            }            // Increment the steal count            m_stealCount++;        }        result = tail.m_value;    }    /// <summary>    /// 獲取總計(jì)列表計(jì)數(shù), 它不是線程安全的, 如果同時(shí)調(diào)用它, 則可能提供不正確的計(jì)數(shù)    /// </summary>    internal int Count    {        get        {            return m_count - m_stealCount;        }    }}

/// <summary>/// 嘗試獲取無主列表，無主列表是指線程已經(jīng)被暫停或者終止，但是集合中的部分?jǐn)?shù)據(jù)還存儲(chǔ)在那里/// 這是避免內(nèi)存泄漏的方法/// </summary>/// <returns></returns>private ThreadLocalList GetUnownedList(){    //此時(shí)必須持有全局鎖    Contract.Assert(Monitor.IsEntered(GlobalListsLock));    // 從頭線程列表開始枚舉 找到那些已經(jīng)被關(guān)閉的線程    // 將它所在的列表對(duì)象 返回    ThreadLocalList currentList = m_headList;    while (currentList != null)    {        if (currentList.m_ownerThread.ThreadState == System.Threading.ThreadState.Stopped)        {            currentList.m_ownerThread = Thread.CurrentThread; // the caller should acquire a lock to make this line thread safe            return currentList;        }        currentList = currentList.m_nextList;    }    return null;}/// <summary>/// 本地幫助方法，通過線程對(duì)象檢索線程線程本地列表/// </summary>/// <param name="forceCreate">如果列表不存在，那么創(chuàng)建新列表</param>/// <returns>The local list object</returns>private ThreadLocalList GetThreadList(bool forceCreate){    ThreadLocalList list = m_locals.Value;    if (list != null)    {        return list;    }    else if (forceCreate)    {        // 獲取用于更新操作的 m_tailList 鎖        lock (GlobalListsLock)        {            // 如果頭列表等于空，那么說明集合中還沒有元素            // 直接創(chuàng)建一個(gè)新的            if (m_headList == null)            {                list = new ThreadLocalList(Thread.CurrentThread);                m_headList = list;                m_tailList = list;            }            else            {               // ConcurrentBag內(nèi)的數(shù)據(jù)是以雙向鏈表的形式分散存儲(chǔ)在各個(gè)線程的本地區(qū)域中                // 通過下面這個(gè)方法 可以找到那些存儲(chǔ)有數(shù)據(jù) 但是已經(jīng)被停止的線程                // 然后將已停止線程的數(shù)據(jù) 移交到當(dāng)前線程管理                list = GetUnownedList();                // 如果沒有 那么就新建一個(gè)列表 然后更新尾指針的位置                if (list == null)                {                    list = new ThreadLocalList(Thread.CurrentThread);                    m_tailList.m_nextList = list;                    m_tailList = list;                }            }            m_locals.Value = list;        }    }    else    {        return null;    }    Debug.Assert(list != null);    return list;}/// <summary>/// Adds an object to the <see cref="ConcurrentBag{T}"/>./// </summary>/// <param name="item">The object to be added to the/// <see cref="ConcurrentBag{T}"/>. The value can be a null reference/// (Nothing in Visual Basic) for reference types.</param>public void Add(T item){    // 獲取該線程的本地列表, 如果此線程不存在, 則創(chuàng)建一個(gè)新列表 (第一次調(diào)用 add)    ThreadLocalList list = GetThreadList(true);    // 實(shí)際的數(shù)據(jù)添加操作 在AddInternal中執(zhí)行    AddInternal(list, item);}/// <summary>/// </summary>/// <param name="list"></param>/// <param name="item"></param>private void AddInternal(ThreadLocalList list, T item){    bool lockTaken = false;    try    {        #pragma warning disable 0420        Interlocked.Exchange(ref list.m_currentOp, (int)ListOperation.Add);        #pragma warning restore 0420        // 同步案例:        // 如果列表計(jì)數(shù)小于兩個(gè), 因?yàn)槭请p向鏈表的關(guān)系 為了避免與任何竊取線程發(fā)生沖突 必須獲取鎖        // 如果設(shè)置了 m_needSync, 這意味著有一個(gè)線程需要凍結(jié)包 也必須獲取鎖        if (list.Count < 2 || m_needSync)        {            // 將其重置為None 以避免與竊取線程的死鎖            list.m_currentOp = (int)ListOperation.None;            // 鎖定當(dāng)前對(duì)象            Monitor.Enter(list, ref lockTaken);        }        // 調(diào)用 ThreadLocalList.Add方法 將數(shù)據(jù)添加到雙向鏈表中        // 如果已經(jīng)鎖定 那么說明線程安全  可以更新Count 計(jì)數(shù)        list.Add(item, lockTaken);    }    finally    {        list.m_currentOp = (int)ListOperation.None;        if (lockTaken)        {            Monitor.Exit(list);        }    }}

/// <summary>/// 本地幫助器方法釋放所有本地列表鎖/// </summary>private void ReleaseAllLocks(){    // 該方法用于在執(zhí)行線程同步以后 釋放掉所有本地鎖    // 通過遍歷每個(gè)線程中存儲(chǔ)的 ThreadLocalList對(duì)象 釋放所占用的鎖    ThreadLocalList currentList = m_headList;    while (currentList != null)    {        if (currentList.m_lockTaken)        {            currentList.m_lockTaken = false;            Monitor.Exit(currentList);        }        currentList = currentList.m_nextList;    }}/// <summary>/// 從凍結(jié)狀態(tài)解凍包的本地幫助器方法/// </summary>/// <param name="lockTaken">The lock taken result from the Freeze method</param>private void UnfreezeBag(bool lockTaken){    // 首先釋放掉 每個(gè)線程中 本地變量的鎖    // 然后釋放全局鎖    ReleaseAllLocks();    m_needSync = false;    if (lockTaken)    {        Monitor.Exit(GlobalListsLock);    }}/// <summary>/// 本地幫助器函數(shù)等待所有未同步的操作/// </summary>private void WaitAllOperations(){    Contract.Assert(Monitor.IsEntered(GlobalListsLock));
    ThreadLocalList currentList = m_headList;    // 自旋等待 等待其它操作完成    while (currentList != null)    {        if (currentList.m_currentOp != (int)ListOperation.None)        {            SpinWait spinner = new SpinWait();            // 有其它線程進(jìn)行操作時(shí)，會(huì)將cuurentOp 設(shè)置成 正在操作的枚舉            while (currentList.m_currentOp != (int)ListOperation.None)            {                spinner.SpinOnce();            }        }        currentList = currentList.m_nextList;    }}/// <summary>/// 本地幫助器方法獲取所有本地列表鎖/// </summary>private void AcquireAllLocks(){    Contract.Assert(Monitor.IsEntered(GlobalListsLock));    bool lockTaken = false;    ThreadLocalList currentList = m_headList;        // 遍歷每個(gè)線程的ThreadLocalList 然后獲取對(duì)應(yīng)ThreadLocalList的鎖    while (currentList != null)    {        // 嘗試/最后 bllock 以避免在獲取鎖和設(shè)置所采取的標(biāo)志之間的線程港口        try        {            Monitor.Enter(currentList, ref lockTaken);        }        finally        {            if (lockTaken)            {                currentList.m_lockTaken = true;                lockTaken = false;            }        }        currentList = currentList.m_nextList;    }}/// <summary>/// Local helper method to freeze all bag operations, it/// 1- Acquire the global lock to prevent any other thread to freeze the bag, and also new new thread can be added/// to the dictionary/// 2- Then Acquire all local lists locks to prevent steal and synchronized operations/// 3- Wait for all un-synchronized operations to be done/// </summary>/// <param name="lockTaken">Retrieve the lock taken result for the global lock, to be passed to Unfreeze method</param>private void FreezeBag(ref bool lockTaken){    Contract.Assert(!Monitor.IsEntered(GlobalListsLock));    // 全局鎖定可安全地防止多線程調(diào)用計(jì)數(shù)和損壞 m_needSync    Monitor.Enter(GlobalListsLock, ref lockTaken);    // 這將強(qiáng)制同步任何將來的添加/執(zhí)行操作    m_needSync = true;    // 獲取所有列表的鎖    AcquireAllLocks();    // 等待所有操作完成    WaitAllOperations();}/// <summary>/// 本地幫助器函數(shù)返回列表中的包項(xiàng), 這主要由 CopyTo 和 ToArray 使用。/// 這不是線程安全, 應(yīng)該被稱為凍結(jié)/解凍袋塊/// 本方法是私有的 只有使用 Freeze/UnFreeze之后才是安全的 /// </summary>/// <returns>List the contains the bag items</returns>private List<T> ToList(){    Contract.Assert(Monitor.IsEntered(GlobalListsLock));    // 創(chuàng)建一個(gè)新的List    List<T> list = new List<T>();    ThreadLocalList currentList = m_headList;    // 遍歷每個(gè)線程中的ThreadLocalList 將里面的Node的數(shù)據(jù) 添加到list中    while (currentList != null)    {        Node currentNode = currentList.m_head;        while (currentNode != null)        {            list.Add(currentNode.m_value);            currentNode = currentNode.m_next;        }        currentList = currentList.m_nextList;    }    return list;}/// <summary>/// Returns an enumerator that iterates through the <see/// cref="ConcurrentBag{T}"/>./// </summary>/// <returns>An enumerator for the contents of the <see/// cref="ConcurrentBag{T}"/>.</returns>/// <remarks>/// The enumeration represents a moment-in-time snapshot of the contents/// of the bag.  It does not reflect any updates to the collection after /// <see cref="GetEnumerator"/> was called.  The enumerator is safe to use/// concurrently with reads from and writes to the bag./// </remarks>public IEnumerator<T> GetEnumerator(){    // Short path if the bag is empty    if (m_headList == null)        return new List<T>().GetEnumerator(); // empty list
    bool lockTaken = false;    try    {        // 首先凍結(jié)整個(gè) ConcurrentBag集合        FreezeBag(ref lockTaken);        // 然后ToList 再拿到 List的 IEnumerator        return ToList().GetEnumerator();    }    finally    {        UnfreezeBag(lockTaken);    }}

使用FreezeBag()方法，凍結(jié)整個(gè)ConcurrentBag<T>集合。因?yàn)樾枰杉系腖ist<T>副本，生成副本期間不能有其它線程更改損壞數(shù)據(jù)。
將ConcurrrentBag<T>生成List<T>副本。因?yàn)镃oncurrentBag<T>存儲(chǔ)數(shù)據(jù)的方式比較特殊，直接實(shí)現(xiàn)迭代器模式困難，考慮到線程安全和邏輯，最佳的辦法是生成一個(gè)副本。
完成以上操作以后，就可以使用UnfreezeBag()方法解凍整個(gè)集合。

首先獲取全局鎖，通過Monitor.Enter(GlobalListsLock, ref lockTaken);這樣一條語句，這樣其它線程就不能凍結(jié)集合。
然后獲取所有線程中ThreadLocalList的鎖，通過`AcquireAllLocks()方法來遍歷獲取。這樣其它線程就不能對(duì)它進(jìn)行操作損壞數(shù)據(jù)。
等待已經(jīng)進(jìn)入了操作流程線程結(jié)束，通過WaitAllOperations()方法來實(shí)現(xiàn)，該方法會(huì)遍歷每一個(gè)ThreadLocalList對(duì)象的m_currentOp屬性，確保全部處于None操作。