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, );
//如果m_needSync，则说明已经发起冻结操作，需要加锁保证线程安全
if ((list.Count < ) || this.m_needSync)
{
list.m_currentOp = ;
Monitor.Enter(list, ref lockTaken);
}
list.Add(item, lockTaken);
}
finally
{
list.m_currentOp = ;
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();
}
冻结bag
5、转化成数组
/// <summary>
/// 转化为数组
/// </summary>
/// <returns></returns>
[__DynamicallyInvokable]
public T[] ToArray()
{
T[] localArray;
//没有数据返回空数组
if (this.m_headList == null)
{
return new T[];
}
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;
}
转化成数组