最近再做一個高併發的伺服器處理程式，伺服器要用多執行緒處理大資料量計算，然後將計算結果封裝成訊息放入佇列中，然後另起幾個執行緒專門負責處理訊息佇列中的訊息分發給不同客戶端，這樣瓶頸就出來了，N多執行緒都在頻繁鎖訊息佇列，這樣導致隊裡的利用效率下降一半，無論是入隊還是出隊都要對佇列進行全部枷鎖，恩。後來仔細分析後得出結論，弄一個雙鎖[隊頭鎖，隊尾鎖]佇列，恩！
 網上也有不少大牛對此進行研究，恩。
原文轉自：Parallel Labs
URL：http://www.parallellabs.com/2010/10/25/practical-concurrent-queue-algorithm/
同時另有高人研究：http://www.cs.rochester.edu/research/synchronization/pseudocode/queues.html#nbq
程式碼如下：

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
}