[.net]ConcurrentBag源码分析

ConcurrentBag根据操作线程，对不同线程分配不同的队列进行数据操作。这样，每个队列只有一个线程在操作，不会发生并发问题。其内部实现运用了net4.0新加入的ThreadLocal线程本地存储功能。各个队列间通过链表维护。

其内部结构如下：

 

1、获取线程本地队列：

/// <summary>
/// 获取当前线程的队列
/// </summary>
/// <param name="forceCreate">如果线程没有持有队列，是否新建</param>
/// <returns></returns>
private ThreadLocalList<T> GetThreadList(bool forceCreate)
{
    //尝试获取线程本地队列列表（参考ThreadLocal），此处的m_locals不同线程持有不同实例
    //如果获取为空，则说明线程是第一次执行此函数，需要分配一个队列
    ThreadLocalList<T> unownedList = this.m_locals.Value;
    if (unownedList != null)
    {
        return unownedList;
    }
    if (forceCreate)
    {
        //获取当前本地队列锁，防止在冻结队列时产生冲突（参考FreezeBag函数）
        object globalListsLock = this.GlobalListsLock;
        lock (globalListsLock)
        {
            //获取本地队列
            //如果没有创建过队列，则创建一个新的队列；否则尽量分配已有的线程终止的队列
            if (this.m_headList == null)
            {
                unownedList = new ThreadLocalList<T>(Thread.CurrentThread);
                this.m_headList = unownedList;
                this.m_tailList = unownedList;
            }
            else
            {
                //获取无主队列，不分配新队列
                unownedList = this.GetUnownedList();
                if (unownedList == null)
                {
                    unownedList = new ThreadLocalList<T>(Thread.CurrentThread);
                    this.m_tailList.m_nextList = unownedList;
                    this.m_tailList = unownedList;
                }
            }
            this.m_locals.Value = unownedList;
            return unownedList;
        }
    }
    return null;
}

2、获取无主队列

/// <summary>
/// 获取无主队列
/// 如果当前队列的持有线程已经终止，则为无主队列
/// </summary>
/// <returns></returns>
private ThreadLocalList<T> GetUnownedList()
{
    for (ThreadLocalList<T> list = this.m_headList; list != null; list = list.m_nextList)
    {
        if (list.m_ownerThread.ThreadState == System.Threading.ThreadState.Stopped)
        {
            list.m_ownerThread = Thread.CurrentThread;
            return list;
        }
    }
    return null;
}

3、插入操作代码分析

/// <summary>
/// 向Bag添加元素
/// </summary>
/// <param name="item"></param>
       
[__DynamicallyInvokable]
public void Add(T item)
{
    //获取当前线程持有的队列
    ThreadLocalList<T> threadList = this.GetThreadList(true);
    //向当前持有队列添加数据
    this.AddInternal(threadList, item);
}

/// <summary>
/// 向队列添加数据
/// </summary>
/// <param name="list"></param>
/// <param name="item"></param>
private void AddInternal(ThreadLocalList<T> list, T item)
{
    bool lockTaken = false;
    try
    {
        //CAS原子操作，设置标志位，与Steal和Freeze实现互斥
        Interlocked.Exchange(ref list.m_currentOp, 1);
        //如果m_needSync，则说明已经发起冻结操作，需要加锁保证线程安全
        if ((list.Count < 2) || this.m_needSync)
        {
            list.m_currentOp = 0;
            Monitor.Enter(list, ref lockTaken);
        }
        list.Add(item, lockTaken);
    }
    finally
    {
        list.m_currentOp = 0;
        if (lockTaken)
        {
            Monitor.Exit(list);
        }
    }
}

4、冻结Bag函数

/// <summary>
/// 冻结Bag，不能进行增,删,获取操作
/// </summary>
/// <param name="lockTaken"></param>
private void FreezeBag(ref bool lockTaken)
{
    //获取当前线程list锁
    Monitor.Enter(this.GlobalListsLock, ref lockTaken);
    //设置同步标志位，增,删,获取操作识别此标志位，只有获取锁才能执行
    this.m_needSync = true;
    //获取所有list的锁
    this.AcquireAllLocks();
    //等待所有操作执行完成
    this.WaitAllOperations();
}

5、转化成数组

/// <summary>
/// 转化为数组
/// </summary>
/// <returns></returns>
[__DynamicallyInvokable]
public T[] ToArray()
{
    T[] localArray;
    //没有数据返回空数组
    if (this.m_headList == null)
    {
        return new T[0];
    }
    bool lockTaken = false;
    try
    {
        //冻结bag
        this.FreezeBag(ref lockTaken);
        //转化成List后直接转成Array
        localArray = this.ToList().ToArray();
    }
    finally
    {
        this.UnfreezeBag(lockTaken);
    }
    return localArray;
}

/// <summary>
/// 转化成list
/// </summary>
/// <returns></returns>
private List<T> ToList()
{
    List<T> list = new List<T>();
    //获取所有list，遍历生成副本
    for (ThreadLocalList<T> list2 = this.m_headList; list2 != null; list2 = list2.m_nextList)
    {
        for (Node<T> node = list2.m_head; node != null; node = node.m_next)
        {
            list.Add(node.m_value);
        }
    }
    return list;
}