多執行緒佇列（Concurrent Queue）的使用場合非常多，高效能伺服器中的訊息佇列，並行演算法中的Work Stealing等都離不開它。對於一個佇列來說有兩個最主要的動作：新增（enqueue）和刪除（dequeue）節點。在一個（或多個）執行緒在對一個佇列進行enqueue操作的同時可能會有一個（或多個）執行緒對這個佇列進行dequeue操作。因為enqueue和dequeue都是對同一個佇列裡的節點進行操作，為了保證執行緒安全，一般在實現中都會在佇列的結構體中加入一個佇列鎖（典型的如pthread_mutex_t q_lock），在進行enqueue和dequeue時都會先鎖住這個鎖以鎖住整個佇列然後再進行相關的操作。這樣的設計如果實現的好的話一般效能就會很不錯了。以連結串列實現的佇列的結構體一般是這樣的：

struct queue_t {
    node_t *head;
    node_t *tail;
    pthread_mutex_t q_lock;
};

但是，這其中其實有一個潛在的效能瓶頸：enqueue和dequeue操作都要鎖住整個佇列，這線上程少的時候可能沒什麼問題，但是隻要執行緒數一多，這個鎖競爭所產生的效能瓶頸就會越來越嚴重。那麼我們可不可以想辦法優化一下這個演算法呢？當然可以！如果我們仔細想一想enqueue和dequeue的具體操作就會發現他們的操作其實不一定是衝突的。例如：如果所有的enqueue操作都是往佇列的尾部插入新節點，而所有的dequeue操作都是從佇列的頭部刪除節點，那麼enqueue和dequeue大部分時候都是相互獨立的，我們大部分時候根本不需要鎖住整個佇列，白白損失效能！那麼一個很自然就能想到的演算法優化方案就呼之欲出了：我們可以把那個佇列鎖拆成兩個：一個佇列頭部鎖（head lock)和一個佇列尾部鎖(tail lock)。這樣這樣的設計思路是對了，但是如果再仔細思考一下它的實現的話我們會發現其實不太容易，因為有兩個特殊情況非常的tricky（難搞）：第一種就是往空佇列裡插入第一個節點的時候，第二種就是從只剩最後一個節點的佇列中刪除那個“最後的果實”的時候。

為什麼難搞呢？當我們向空佇列中插入第一個節點的時候，我們需要同時修改佇列的head和tail指標，使他們同時指向這個新插入的節點，換句話說，我們此時即需要拿到head lock又需要拿到tail lock。而另一種情況是對只剩一個節點的佇列進行dequeue的時候，我們也是需要同時修改head和tail指標使他們指向NULL，亦即我們需要同時獲得head和tail lock。有經驗的同學會立刻發現我們進入危險區了！是什麼危險呢？死鎖！多執行緒程式設計中最臭名昭著的一種bug就是死鎖了。例如，如果執行緒A在鎖住了資源1後還想要獲取資源2，而執行緒B在鎖住了資源2後還想要獲取資源1，這時兩個執行緒誰都不能獲得自己想要的那個資源，兩個執行緒就死鎖了。所以我們要小心奕奕的設計這個演算法以避免死鎖，例如保證enqueue和dequeue對head lock和tail lock的請求順序（lock ordering）是一致的等等。但是這樣設計出來的演算法很容易就會包含多次的加鎖/解鎖操作，這些都會造成不必要的開銷，尤其是線上程數很多的情況下反而可能導致效能的下降。我的親身經歷就是在32執行緒時這個思路設計出來的演算法效能反而下降了10%左右，原因就是加鎖/解鎖的開銷增加了。

好在有聰明人早在96年就想到了一個更妙的演算法。這個演算法也是用了head和tail兩個鎖，但是它有一個關鍵的地方是它在佇列初始化的時候head和tail指標不為空，而是指向一個空節點。在enqueue的時候只要向佇列尾部新增新節點就好了。而dequeue的情況稍微複雜點，它要返回的不是頭節點，而是head->next，即頭節點的下一個節點。先來看虛擬碼：

typedef struct node_t {
    TYPE value; 
    node_t *next
} NODE;

typedef struct queue_t {
    NODE *head; 
    NODE *tail;
    LOCK q_h_lock;
    LOCK q_t_lock;
} Q;

initialize(Q *q) {
   node = new_node()   // Allocate a free node
   node->next = NULL   // Make it the only node in the linked list
   q->head = q->tail = node	// Both head and tail point to it
   q->q_h_lock = q->q_t_lock = FREE   // Locks are initially free
}

enqueue(Q *q, TYPE value) {
   node = new_node()       // Allocate a new node from the free list
   node->value = value	  // Copy enqueued value into node
   node->next = NULL       // Set next pointer of node to NULL
   lock(&q->q_t_lock)	  // Acquire t_lock in order to access Tail
      q->tail->next = node // Link node at the end of the queue
      q->tail = node       // Swing Tail to node
   unlock(&q->q_t_lock)    // Release t_lock
｝

dequeue(Q *q, TYPE *pvalue) {
   lock(&q->q_h_lock)   // Acquire h_lock in order to access Head
      node = q->head    // Read Head
      new_head = node->next	     // Read next pointer
      if new_head == NULL         // Is queue empty?
         unlock(&q->q_h_lock)     // Release h_lock before return
         return FALSE             // Queue was empty
      endif
      *pvalue = new_head->value   // Queue not empty, read value
      q->head = new_head  // Swing Head to next node
   unlock(&q->q_h_lock)   // Release h_lock
   free(node)			  // Free node
   return TRUE			  // Queue was not empty, dequeue succeeded
}

發現玄機了麼？是的，這個演算法中佇列總會包含至少一個節點。dequeue每次返回的不是頭節點，而是頭節點的下一個節點中的資料：如果head->next不為空的話就把這個節點的資料取出來作為返回值，同時再把head指標指向這個節點，此時舊的頭節點就可以被free掉了。這個在佇列初始化時插入空節點的技巧使得enqueue和dequeue徹底相互獨立了。但是，還有一個小地方在實現的時候需要注意：對第一個空節點的next指標的讀寫。想象一下，當一個執行緒對一個空佇列進行第一次enqueue操作時剛剛執行完第25行的程式碼（對該空節點的next指標進行寫操作）；而此時另一個執行緒對這個佇列進行第一次dequeue操作時恰好執行到第33行（對該空節點的next指標進行讀操作），它們其實還是有衝突！不過，好在一般來講next指標是32位資料，而現代的CPU已經能保證多執行緒程式中記憶體對齊了的32位資料讀寫操作的原子性，而一般來講編譯器會自動幫你對齊32位資料，所以這個不是問題。唯一需要注意的是我們要確保enqueue執行緒是先讓要新增的新節點包含好資料再把新節點插入連結串列（也就是不能先插入空節點，再往節點中填入資料），那麼dequeue執行緒就不會拿到空的節點。其實我們也可以把q_t_lock理解成生產者的鎖，q_h_lock理解成消費者的鎖，這樣生產者（們）和消費者（們）的操作就相互獨立了，只有在多個生產者對同一佇列進行新增操作時，以及多個消費者對同一佇列進行刪除操作時才需要加鎖以使訪問互斥。

通過使用這個演算法，我成功的把一個32執行緒程式的效能提升了11%！可見多執行緒中的鎖競爭對效能影響之大！此演算法出自一篇著名的論文：M. Michael and M. Scott. Simple, Fast, and Practical Non-Blocking and Blocking Concurren Queue Algorithms. 如果還想做更多優化的話可以參考這篇論文實現相應的Non Blocking版本的演算法，效能還能有更多提升。當然了，這個演算法早已被整合到java.util.concurrent裡了（即LinkedBlockingQueue），其他的並行庫例如Intel的TBB多半也有類似的演算法，如果大家能用上現成的庫的話就不要再重複造輪子了。為什麼別造並行演算法的輪子呢？因為高效能的並行演算法實在太難正確地實現了，尤其是Non Blocking，Lock Free之類的“火箭工程”。有多難呢？Doug Lea提到java.util.concurrent中一個Non Blocking的演算法的實現大概需要1年的時間，總共約500行程式碼。所以，對最廣大的程式設計師來說，別去寫Non Blocking, Lock Free的程式碼，只管用就行了，我看見網上很多的Non Blocking阿，無鎖程式設計的演算法實現啊什麼的都非常地害怕，誰敢去用他們貼出來的這些程式碼啊？我之所以推薦這個two lock的演算法是因為它的實現相對Non Blocking之類的來說容易多了，非常具備實用價值。雖然這篇論文出現的很早，但是我在看了幾個開源軟體中多執行緒佇列的實現之後發現他們很多還是用的本文最開始提到的那種一個鎖的演算法。如果你想要實現更高效能的多執行緒佇列的話，試試這個演算法吧！

Update: 多執行緒佇列演算法有很多種，大家應根據不同的應用場合選取最優演算法（例如是CPU密集型還是IO密集型）。本文所列的演算法應用在這樣一個多執行緒程式中：每個執行緒都擁有一個佇列，每個佇列可能被本執行緒進行dequeue操作，也可以被其他執行緒進行dequeue（即work stealing），執行緒數不超過CPU核心數，是一個典型的CPU/MEM密集型客戶端單寫者多讀者場景。

多執行緒佇列（Concurrent Queue）的使用場合非常多，高效能伺服器中的訊息佇列，並行演算法中的Work Stealing等都離不開它。對於一個佇列來說有兩個最主要的動作：新增（enqueue）和刪除（dequeue）節點。在一個（或多個）執行緒在對一個佇列進行enqueue操作的同時可能會有一個（或多個）執行緒對這個佇列進行dequeue操作。因為enqueue和dequeue都是對同一個佇列裡的節點進行操作，為了保證執行緒安全，一般在實現中都會在佇列的結構體中加入一個佇列鎖（典型的如pthread_mutex_t q_lock），在進行enqueue和dequeue時都會先鎖住這個鎖以鎖住整個佇列然後再進行相關的操作。這樣的設計如果實現的好的話一般效能就會很不錯了。以連結串列實現的佇列的結構體一般是這樣的：

struct queue_t {
    node_t *head;
    node_t *tail;
    pthread_mutex_t q_lock;
};

但是，這其中其實有一個潛在的效能瓶頸：enqueue和dequeue操作都要鎖住整個佇列，這線上程少的時候可能沒什麼問題，但是隻要執行緒數一多，這個鎖競爭所產生的效能瓶頸就會越來越嚴重。那麼我們可不可以想辦法優化一下這個演算法呢？當然可以！如果我們仔細想一想enqueue和dequeue的具體操作就會發現他們的操作其實不一定是衝突的。例如：如果所有的enqueue操作都是往佇列的尾部插入新節點，而所有的dequeue操作都是從佇列的頭部刪除節點，那麼enqueue和dequeue大部分時候都是相互獨立的，我們大部分時候根本不需要鎖住整個佇列，白白損失效能！那麼一個很自然就能想到的演算法優化方案就呼之欲出了：我們可以把那個佇列鎖拆成兩個：一個佇列頭部鎖（head lock)和一個佇列尾部鎖(tail lock)。這樣這樣的設計思路是對了，但是如果再仔細思考一下它的實現的話我們會發現其實不太容易，因為有兩個特殊情況非常的tricky（難搞）：第一種就是往空佇列裡插入第一個節點的時候，第二種就是從只剩最後一個節點的佇列中刪除那個“最後的果實”的時候。

為什麼難搞呢？當我們向空佇列中插入第一個節點的時候，我們需要同時修改佇列的head和tail指標，使他們同時指向這個新插入的節點，換句話說，我們此時即需要拿到head lock又需要拿到tail lock。而另一種情況是對只剩一個節點的佇列進行dequeue的時候，我們也是需要同時修改head和tail指標使他們指向NULL，亦即我們需要同時獲得head和tail lock。有經驗的同學會立刻發現我們進入危險區了！是什麼危險呢？死鎖！多執行緒程式設計中最臭名昭著的一種bug就是死鎖了。例如，如果執行緒A在鎖住了資源1後還想要獲取資源2，而執行緒B在鎖住了資源2後還想要獲取資源1，這時兩個執行緒誰都不能獲得自己想要的那個資源，兩個執行緒就死鎖了。所以我們要小心奕奕的設計這個演算法以避免死鎖，例如保證enqueue和dequeue對head lock和tail lock的請求順序（lock ordering）是一致的等等。但是這樣設計出來的演算法很容易就會包含多次的加鎖/解鎖操作，這些都會造成不必要的開銷，尤其是線上程數很多的情況下反而可能導致效能的下降。我的親身經歷就是在32執行緒時這個思路設計出來的演算法效能反而下降了10%左右，原因就是加鎖/解鎖的開銷增加了。

好在有聰明人早在96年就想到了一個更妙的演算法。這個演算法也是用了head和tail兩個鎖，但是它有一個關鍵的地方是它在佇列初始化的時候head和tail指標不為空，而是指向一個空節點。在enqueue的時候只要向佇列尾部新增新節點就好了。而dequeue的情況稍微複雜點，它要返回的不是頭節點，而是head->next，即頭節點的下一個節點。先來看虛擬碼：

typedef struct node_t {
    TYPE value; 
    node_t *next
} NODE;

typedef struct queue_t {
    NODE *head; 
    NODE *tail;
    LOCK q_h_lock;
    LOCK q_t_lock;
} Q;

initialize(Q *q) {
   node = new_node()   // Allocate a free node
   node->next = NULL   // Make it the only node in the linked list
   q->head = q->tail = node	// Both head and tail point to it
   q->q_h_lock = q->q_t_lock = FREE   // Locks are initially free
}

enqueue(Q *q, TYPE value) {
   node = new_node()       // Allocate a new node from the free list
   node->value = value	  // Copy enqueued value into node
   node->next = NULL       // Set next pointer of node to NULL
   lock(&q->q_t_lock)	  // Acquire t_lock in order to access Tail
      q->tail->next = node // Link node at the end of the queue
      q->tail = node       // Swing Tail to node
   unlock(&q->q_t_lock)    // Release t_lock
｝

dequeue(Q *q, TYPE *pvalue) {
   lock(&q->q_h_lock)   // Acquire h_lock in order to access Head
      node = q->head    // Read Head
      new_head = node->next	     // Read next pointer
      if new_head == NULL         // Is queue empty?
         unlock(&q->q_h_lock)     // Release h_lock before return
         return FALSE             // Queue was empty
      endif
      *pvalue = new_head->value   // Queue not empty, read value
      q->head = new_head  // Swing Head to next node
   unlock(&q->q_h_lock)   // Release h_lock
   free(node)			  // Free node
   return TRUE			  // Queue was not empty, dequeue succeeded
}

發現玄機了麼？是的，這個演算法中佇列總會包含至少一個節點。dequeue每次返回的不是頭節點，而是頭節點的下一個節點中的資料：如果head->next不為空的話就把這個節點的資料取出來作為返回值，同時再把head指標指向這個節點，此時舊的頭節點就可以被free掉了。這個在佇列初始化時插入空節點的技巧使得enqueue和dequeue徹底相互獨立了。但是，還有一個小地方在實現的時候需要注意：對第一個空節點的next指標的讀寫。想象一下，當一個執行緒對一個空佇列進行第一次enqueue操作時剛剛執行完第25行的程式碼（對該空節點的next指標進行寫操作）；而此時另一個執行緒對這個佇列進行第一次dequeue操作時恰好執行到第33行（對該空節點的next指標進行讀操作），它們其實還是有衝突！不過，好在一般來講next指標是32位資料，而現代的CPU已經能保證多執行緒程式中記憶體對齊了的32位資料讀寫操作的原子性，而一般來講編譯器會自動幫你對齊32位資料，所以這個不是問題。唯一需要注意的是我們要確保enqueue執行緒是先讓要新增的新節點包含好資料再把新節點插入連結串列（也就是不能先插入空節點，再往節點中填入資料），那麼dequeue執行緒就不會拿到空的節點。其實我們也可以把q_t_lock理解成生產者的鎖，q_h_lock理解成消費者的鎖，這樣生產者（們）和消費者（們）的操作就相互獨立了，只有在多個生產者對同一佇列進行新增操作時，以及多個消費者對同一佇列進行刪除操作時才需要加鎖以使訪問互斥。

通過使用這個演算法，我成功的把一個32執行緒程式的效能提升了11%！可見多執行緒中的鎖競爭對效能影響之大！此演算法出自一篇著名的論文：M. Michael and M. Scott. Simple, Fast, and Practical Non-Blocking and Blocking Concurren Queue Algorithms. 如果還想做更多優化的話可以參考這篇論文實現相應的Non Blocking版本的演算法，效能還能有更多提升。當然了，這個演算法早已被整合到java.util.concurrent裡了（即LinkedBlockingQueue），其他的並行庫例如Intel的TBB多半也有類似的演算法，如果大家能用上現成的庫的話就不要再重複造輪子了。為什麼別造並行演算法的輪子呢？因為高效能的並行演算法實在太難正確地實現了，尤其是Non Blocking，Lock Free之類的“火箭工程”。有多難呢？Doug Lea提到java.util.concurrent中一個Non Blocking的演算法的實現大概需要1年的時間，總共約500行程式碼。所以，對最廣大的程式設計師來說，別去寫Non Blocking, Lock Free的程式碼，只管用就行了，我看見網上很多的Non Blocking阿，無鎖程式設計的演算法實現啊什麼的都非常地害怕，誰敢去用他們貼出來的這些程式碼啊？我之所以推薦這個two lock的演算法是因為它的實現相對Non Blocking之類的來說容易多了，非常具備實用價值。雖然這篇論文出現的很早，但是我在看了幾個開源軟體中多執行緒佇列的實現之後發現他們很多還是用的本文最開始提到的那種一個鎖的演算法。如果你想要實現更高效能的多執行緒佇列的話，試試這個演算法吧！

Update: 多執行緒佇列演算法有很多種，大家應根據不同的應用場合選取最優演算法（例如是CPU密集型還是IO密集型）。本文所列的演算法應用在這樣一個多執行緒程式中：每個執行緒都擁有一個佇列，每個佇列可能被本執行緒進行dequeue操作，也可以被其他執行緒進行dequeue（即work stealing），執行緒數不超過CPU核心數，是一個典型的CPU/MEM密集型客戶端單寫者多讀者場景。