如果是自己實現(xiàn)一個簡單的隊列功能，我們該如何設(shè)計它的存儲結(jié)構(gòu)呢？一般來說有這兩種方式：數(shù)組或者鏈表，先來簡單分析下。
我們都知道，數(shù)組是固定空間的集合，意味著初始化的時候要指定數(shù)組大小，但是隊列的長度是隨時變化的，超出數(shù)組大小了怎么辦？這時候就必須要對數(shù)組進(jìn)行擴容。問題又來了，擴容要擴多少呢，少了不夠用多了浪費內(nèi)存空間。與之相反的，鏈表是動態(tài)空間類型的數(shù)據(jù)結(jié)構(gòu)，元素之間通過指針相連，不需要提前分配空間，需要多少分配多少。但隨之而來的問題是，大量的出隊入隊操作伴隨著大量對象的創(chuàng)建銷毀，GC的壓力又變得非常大。
回到主題，要實現(xiàn)一個高性能的線程安全隊列，我們試著回答以下問題：
存儲結(jié)構(gòu)是怎樣的如何初始化（初始容量給多少比較好？）常用操作（入隊出隊）如何實現(xiàn)線程安全是如何保證的通過源碼可以看到ConcurrentQueue采用了數(shù)組+鏈表的組合模式，充分吸收了2種結(jié)構(gòu)的優(yōu)點。
具體來說，它的總體結(jié)構(gòu)是一個鏈表，鏈表的每個節(jié)點是一個包含數(shù)組的特殊對象，我們稱之為Segment（段或節(jié)，原話是a queue is a linked list of small arrays, each node is called a segment.），它里面的數(shù)組是存儲真實數(shù)據(jù)的地方，容量固定大小是32，每一個Segment有指向下一個Segment的的指針，以此形成鏈表結(jié)構(gòu)。而隊列中維護(hù)了2個特殊的指針，他們分別指向隊列的首段（head segment）和尾段（tail segment），他們對入隊和出隊有著重要的作用。用一張圖來解釋隊列的內(nèi)部結(jié)構(gòu)：
嗯，畫圖畫到這里突然聯(lián)想到，搞成雙向鏈表的話是不是就神似B+樹的葉子節(jié)點？技術(shù)就是這么奇妙~
段的核心定義為：
/// <summary>
/// private class for ConcurrentQueue.
/// 鏈表節(jié)點（段）
/// </summary>
private class Segment
{
//實際存儲數(shù)據(jù)的容器
internal volatile T[] m_array;
//存儲對應(yīng)位置數(shù)據(jù)的狀態(tài)，當(dāng)數(shù)據(jù)的對應(yīng)狀態(tài)位標(biāo)記為true時該數(shù)據(jù)才是有效的
internal volatile VolatileBool[] m_state;
//下一段的指針
private volatile Segment m_next;
//當(dāng)前段在隊列中的索引
internal readonly long m_index;
//兩個位置指針
private volatile int m_low;
private volatile int m_high;
//所屬的隊列實例
private volatile ConcurrentQueue<T> m_source;
}
隊列的核心定義為：
/// <summary>
/// 線程安全的先進(jìn)先出集合，
/// </summary>
public class ConcurrentQueue<T> : IProducerConsumerCollection<T>, IReadOnlyCollection<T>
{
//首段
[NonSerialized]
private volatile Segment m_head;
//尾段
[NonSerialized]
private volatile Segment m_tail;
//每一段的大小
private const int SEGMENT_SIZE = 32;
//截取快照的操作數(shù)量
[NonSerialized]
internal volatile int m_numSnapshotTakers = 0;
}
先從初始化一個隊列開始看起。
創(chuàng)建隊列實例與普通Queue不同的是，ConcurrentQueue不再支持初始化時指定隊列大?。╟apacity），僅僅提供一個無參構(gòu)造函數(shù)和一個IEnumerable<T>參數(shù)的構(gòu)造函數(shù)。
/// <summary>
/// Initializes a new instance of the <see cref="ConcurrentQueue{T}"/> class.
/// </summary>
public ConcurrentQueue()
{
m_head = m_tail = new Segment(0, this);
}
無參構(gòu)造函數(shù)很簡單，創(chuàng)建了一個Segment實例并把首尾指針都指向它，此時隊列只包含一個Segment，它的索引是0，隊列容量是32。繼續(xù)看一下Segment是如何被初始化的：
/// <summary>
/// Create and initialize a segment with the specified index.
/// </summary>
internal Segment(long index, ConcurrentQueue<T> source)
{
m_array = new T[SEGMENT_SIZE];
m_state = new VolatileBool[SEGMENT_SIZE]; //all initialized to false
m_high = -1;
Contract.Assert(index >= 0);
m_index = index;
m_source = source;
}
Segment只提供了一個構(gòu)造函數(shù)，接受的參數(shù)分別是隊列索引和隊列實例，它創(chuàng)建了一個長度為32的數(shù)組，并創(chuàng)建了與之對應(yīng)的狀態(tài)數(shù)組，然后初始化了位置指針（m_low=0，m_high=-1，此時表示一個空的Segment）。到這里，一個并發(fā)隊列就創(chuàng)建好了。
使用集合創(chuàng)建隊列的過程和上面類似，只是多了兩個步驟：入隊和擴容，下面會重點描述這兩部分所以這里不再過多介紹。
先亮出源碼：
/// <summary>
/// Adds an object to the end of the <see cref="ConcurrentQueue{T}"/>.
/// </summary>
/// <param name="item">The object to add to the end of the <see
/// cref="ConcurrentQueue{T}"/>. The value can be a null reference
/// (Nothing in Visual Basic) for reference types.
/// </param>
public void Enqueue(T item)
{
SpinWait spin = new SpinWait();
while (true)
{
Segment tail = m_tail;
if (tail.TryAppend(item))
return;
spin.SpinOnce();
}
}
通過源碼可以看到，入隊操作是在隊尾（m_tail）進(jìn)行的，它嘗試在最后一個Segment中追加指定的元素，如果成功了就直接返回，失敗的話就自旋等待，直到成功為止。那什么情況下會失敗呢？這就要繼續(xù)看看是如何追加元素的：
internal bool TryAppend(T value)
{
//先判斷一下高位指針有沒有達(dá)到數(shù)組邊界（也就是數(shù)組是否裝滿了）
if (m_high >= SEGMENT_SIZE - 1)
{
return false;
}
int newhigh = SEGMENT_SIZE;
try
{ }
finally
{
//使用原子操作讓高位指針加1
newhigh = Interlocked.Increment(ref m_high);
//如果數(shù)組還有空位
if (newhigh <= SEGMENT_SIZE - 1)
{
//把數(shù)據(jù)放到數(shù)組中，同時更新狀態(tài)
m_array[newhigh] = value;
m_state[newhigh].m_value = true;
}
//數(shù)組滿了要觸發(fā)擴容
if (newhigh == SEGMENT_SIZE - 1)
{
Grow();
}
}
return newhigh <= SEGMENT_SIZE - 1;
}
所以，只有當(dāng)尾段m_tail裝滿的情況下追加元素才會失敗，這時候必須要等待下一個段產(chǎn)生，也就是擴容（細(xì)細(xì)品一下Grow這個詞真的很妙），自旋就是在等擴容完成才能有地方放數(shù)據(jù)。而在保存數(shù)據(jù)的時候，通過原子自增操作保證了同一個位置只會有一個數(shù)據(jù)被寫入，從而實現(xiàn)了線程安全。
注意：這里的裝滿并不是指數(shù)組每個位置都有數(shù)據(jù)，而是指最后一個位置已被使用。
繼續(xù)看一下擴容是怎么一個過程：
/// <summary>
/// Create a new segment and append to the current one
/// Update the m_tail pointer
/// This method is called when there is no contention
/// </summary>
internal void Grow()
{
//no CAS is needed, since there is no contention (other threads are blocked, busy waiting)
Segment newSegment = new Segment(m_index + 1, m_source); //m_index is Int64, we don't need to worry about overflow
m_next = newSegment;
Contract.Assert(m_source.m_tail == this);
m_source.m_tail = m_next;
}
在普通隊列中，擴容是通過創(chuàng)建一個更大的數(shù)組然后把數(shù)據(jù)拷貝過去實現(xiàn)擴容的，這個操作比較耗時。而在并發(fā)隊列中就非常簡單了，首先創(chuàng)建一個新Segment，然后把當(dāng)前Segment的next指向它，最后掛到隊列的末尾去就可以了，全部是指針操作非常高效。而且從代碼注釋中可以看到，這里不會出現(xiàn)線程競爭的情況，因為其他線程都因為位置不夠被阻塞都在自旋等待中。
元素出隊還是先亮出源碼：
public bool TryDequeue(out T result)
{
while (!IsEmpty)
{
Segment head = m_head;
if (head.TryRemove(out result))
return true;
//since method IsEmpty spins, we don't need to spin in the while loop
}
result = default(T);
return false;
}
可以看到只有在隊列不為空（IsEmpty==false）的情況下才會嘗試出隊操作，而出隊是在首段上進(jìn)行操作的。關(guān)于如何判斷隊列是否為空總結(jié)就一句話：當(dāng)首段m_head不包含任何數(shù)據(jù)且沒有下一段的時候隊列才為空，詳細(xì)的判斷過程源碼注釋中寫的很清楚，限于篇幅不詳細(xì)介紹。
出隊的本質(zhì)是從首段中移除低位指針?biāo)赶虻脑?，看一下具體實現(xiàn)步驟：
internal bool TryRemove(out T result)
{
SpinWait spin = new SpinWait();
int lowLocal = Low, highLocal = High;
//判斷當(dāng)前段是否為空
while (lowLocal <= highLocal)
{
//判斷低位指針位置是否可以移除
if (Interlocked.CompareExchange(ref m_low, lowLocal + 1, lowLocal) == lowLocal)
{
SpinWait spinLocal = new SpinWait();
//判斷元素是否有效
while (!m_state[lowLocal].m_value)
{
spinLocal.SpinOnce();
}
//取出元素
result = m_array[lowLocal];
//釋放引用關(guān)系
if (m_source.m_numSnapshotTakers <= 0)
{
m_array[lowLocal] = default(T);
}
//判斷當(dāng)前段的元素是否全部被移除了，要丟棄它
if (lowLocal + 1 >= SEGMENT_SIZE)
{
spinLocal = new SpinWait();
while (m_next == null)
{
spinLocal.SpinOnce();
}
Contract.Assert(m_source.m_head == this);
m_source.m_head = m_next;
}
return true;
}
else
{
//線程競爭失敗，自旋等待并重置
spin.SpinOnce();
lowLocal = Low; highLocal = High;
}
}//end of while
result = default(T);
return false;
}
首先，只有當(dāng)前Segment不為空的情況下才嘗試移除元素，否則就直接返回false。然后通過一個原子操作Interlocked.CompareExchange判斷當(dāng)前低位指針上是否有其他線程同時也在移除，如果有那就進(jìn)入自旋等待，沒有的話就從這個位置取出元素并把低位指針往前推進(jìn)一位。如果當(dāng)前隊列沒有正在進(jìn)行截取快照的操作，那取出元素后還要把這個位置給釋放掉。當(dāng)這個Segment的所有元素都被移除掉了，這時候要把它丟棄，簡單來說就是讓隊列的首段指針指向它的下一段即可，丟棄的這一段等著GC來收拾它。
這里稍微提一下Interlocked.CompareExchange，它的意思是比較和交換，也就是更為大家所熟悉的CAS（Compare-and-Swap），它主要做了以下2件事情：
比較m_low和lowLocal的值是否相等如果相等則m_low=lowLocal+1，如果不相等就什么都不做，不管是否相等，始終返回m_low的原始值整個操作是原子性的，對CPU而言就是一條指令，這樣就可以保證當(dāng)前位置只有一個線程執(zhí)行出隊操作。
獲取隊列中元素的數(shù)量還有一個TryPeek()方法和出隊類似，它是從隊首獲取一個元素但是無需移除該元素，可以看做Dequeue的簡化版，不再詳細(xì)介紹。
與普通Queue不同的是，ConcurrentQueue并沒有維護(hù)一個表示隊列中元素個數(shù)的計數(shù)器，那就意味著要得到這個數(shù)量必須實時去計算。我們看一下計算過程：
public int Count
{
get
{
Segment head, tail;
int headLow, tailHigh;
GetHeadTailPositions(out head, out tail, out headLow, out tailHigh);
if (head == tail)
{
return tailHigh - headLow + 1;
}
int count = SEGMENT_SIZE - headLow;
count += SEGMENT_SIZE * ((int)(tail.m_index - head.m_index - 1));
count += tailHigh + 1;
return count;
}
}
大致思路是，先計算（GetHeadTailPositions）出首段的低位指針和尾段的高位指針，這中間的總長度就是我們要的數(shù)量，然后分成3節(jié)依次累加每一個Segment包含的元素個數(shù)得到最終的隊列長度，可以看到這是一個開銷比較大的操作。正因為如此，微軟官方推薦使用IsEmpty屬性來判斷隊列是否為空，而不是使用隊列長度Count==0來判斷，使用ConcurrentStack也是一樣。
截取快照（take snapshot）所謂的take snapshot就是指一些格式轉(zhuǎn)換的操作，例如ToArray()、ToList()、GetEnumerator()這種類型的方法。在前面隊列的核心定義中我們提到有一個m_numSnapshotTakers字段，這時候就派上用場了。下面以比較典型的ToList()源碼舉例說明：
private List<T> ToList()
{
// Increments the number of active snapshot takers. This increment must happen before the snapshot is
// taken. At the same time, Decrement must happen after list copying is over. Only in this way, can it
// eliminate race condition when Segment.TryRemove() checks whether m_numSnapshotTakers == 0.
Interlocked.Increment(ref m_numSnapshotTakers);
List<T> list = new List<T>();
try
{
Segment head, tail;
int headLow, tailHigh;
GetHeadTailPositions(out head, out tail, out headLow, out tailHigh);
if (head == tail)
{
head.AddToList(list, headLow, tailHigh);
}
else
{
head.AddToList(list, headLow, SEGMENT_SIZE - 1);
Segment curr = head.Next;
while (curr != tail)
{
curr.AddToList(list, 0, SEGMENT_SIZE - 1);
curr = curr.Next;
}
tail.AddToList(list, 0, tailHigh);
}
}
finally
{
// This Decrement must happen after copying is over.
Interlocked.Decrement(ref m_numSnapshotTakers);
}
return list;
}
可以看到，ToList的邏輯和Count非常相似，都是先計算出兩個首尾位置指針，然后把隊列分為3節(jié)依次遍歷處理，最大的不同之處在于方法的開頭和結(jié)尾分別對m_numSnapshotTakers做了一個原子操作。在方法的第一行，使用Interlocked.Increment做了一次遞增，這時候表示隊列正在進(jìn)行一次截取快照操作，在處理完后又在finally中用Interlocked.Decrement做了一次遞減表示當(dāng)前操作已完成，這樣確保了在進(jìn)行快照時不被出隊影響。感覺這塊很難描述的特別好，所以保留了原始的英文注釋，大家慢慢體會。
到這里，基本把ConcurrentQueue的核心說清楚了。
總結(jié)一下回到文章開頭提出的幾個問題，現(xiàn)在應(yīng)該有了很清晰的答案：
存儲結(jié)構(gòu) -- 采用數(shù)組和鏈表的組合形式如何初始化 -- 創(chuàng)建固定大小的段，無需指定初始容量常用操作如何實現(xiàn) -- 尾段入隊，首段出隊線程安全問題 -- 使用SpinWait自旋等待和原子操作實現(xiàn)最后多說一句，很多人學(xué)Python過程中會遇到各種煩惱問題，沒有人解答容易放棄。小編是一名python開發(fā)工程師，這里有我自己整理了一套最新的python系統(tǒng)學(xué)習(xí)教程，包括從基礎(chǔ)的python腳本到web開發(fā)、爬蟲、數(shù)據(jù)分析、數(shù)據(jù)可視化、機器學(xué)習(xí)等。想要這些資料的可以關(guān)注小編，并在后臺私信小編：“01”即可領(lǐng)取。
