基础知识（6）
多线程队列（Concurrent Queue）的使用场合非常多，高性能服务器中的消息队列，并行算法中的Work Stealing等都离不开它。对于一个队列来说有两个最主要的动作：添加（enqueue）和删除（dequeue）节点。在一个（或多个）线程在对一个队列进行enqueue操作的同时可能会有一个（或多个）线程对这个队列进行dequeue操作。因为enqueue和dequeue都是对同一个队列里的节点进行操作，为了保证线程安全，一般在实现中都会在队列的结构体中加入一个队列锁（典型的如pthread_mutex_t
q_lock），在进行enqueue和dequeue时都会先锁住这个锁以锁住整个队列然后再进行相关的操作。这样的设计如果实现的好的话一般性能就会很不错了。以链表实现的队列的结构体一般是这样的：
struct queue_t {
pthread_mutex_t q_
但是，这其中其实有一个潜在的性能瓶颈：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 {
node_t *next
typedef struct queue_t {
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
*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.&. 如果还想做更多优化的话可以参考这篇论文实现相应的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 {
pthread_mutex_t q_
但是，这其中其实有一个潜在的性能瓶颈：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 {
node_t *next
typedef struct queue_t {
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
*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.&. 如果还想做更多优化的话可以参考这篇论文实现相应的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密集型客户端单写者多读者场景。
&&相关文章推荐
参考知识库
* 以上用户言论只代表其个人观点，不代表CSDN网站的观点或立场
访问：34094次
排名：千里之外
原创：12篇
转载：83篇java（85）
一.同步队列BlockingQueue
前面的两篇博文：
详细阐述了多个任务之间的协同合作，需要使用wait、notify、notifyAll或者lock、condition、await、signal、signalAll方法来进行同步。实现起来比较复杂。因此java提供了同步队列（阻塞队列BlockingQueue）。
同步队列要求只能有一个任务对其进行操作（因此无需再对其使用同步操作，同步队列内部是同步的。）。当队列是空时，会导致取该队列的线程阻塞；当队列满（设置固定大小的队列）时，会导致写该队列的线程阻塞。
java的JUC（java.util.concurrent）包提供了BlockingQueue接口，并为其提供了三个实现：LinkedBlockingQueue（无界队列）、ArrayBlockingQueue（有固定大小的队列）、 SynchronousQueue（大小为1的队列）。
下面请看代码：
package org.fan.learn.thread.
* Created by fan on .
public class LiftOff implements Runnable {
private static int taskCunnt = 0;
private final int id = taskCunnt++;
private int countDown = 10;
private void status() {
System.out.print("Task(" + id + ")#" + (countDown & 0 ? countDown : "liftOff") + "
public void run() {
while (countDown-- & 0) {
Thread.yield();
System.out.println();
package org.fan.learn.thread.
import java.util.concurrent.*;
* Created by fan on .
class LiftOffRunner implements Runnable {
private BlockingQueue&LiftOff& blockingQ
public LiftOffRunner(BlockingQueue&LiftOff& blockingQueue) {
this.blockingQueue = blockingQ
public void add(LiftOff liftOff) {
blockingQueue.put(liftOff);
} catch (InterruptedException e) {
e.printStackTrace();
public void run() {
while (!Thread.interrupted()) {
LiftOff liftOff = blockingQueue.take();
liftOff.run();
} catch (InterruptedException e) {
e.printStackTrace();
System.out.println("Exiting Liftoff Runner");
public class TestBlockingQueue {
static void test(BlockingQueue&LiftOff& blockingQueue) throws InterruptedException {
LiftOffRunner liftOffRunner = new LiftOffRunner(blockingQueue);
Thread thread = new Thread(liftOffRunner);
thread.start();
for (int i = 0; i & 5; i++) {
liftOffRunner.add(new LiftOff());
TimeUnit.SECONDS.sleep(2);
thread.interrupt();
System.out.println("Finished Test");
public static void main(String[] args) {
test(new SynchronousQueue&LiftOff&());
} catch (InterruptedException e) {
e.printStackTrace();
//add方法，如果是ArrayBlockingQueue会抛 队列满的异常
//blockingQueue.add(liftOff);
1.add会造成异常。
Exception in thread "main" java.lang.IllegalStateException: Queue full
Task(0)#liftOff
Task(1)#liftOff
Task(2)#liftOff
at java.util.AbstractQueue.add(AbstractQueue.java:98)
at java.util.concurrent.ArrayBlockingQueue.add(ArrayBlockingQueue.java:312)
at org.fan.learn.thread.blockqueue.LiftOffRunner.add(TestBlockingQueue.java:20)
at org.fan.learn.thread.blockqueue.TestBlockingQueue.test(TestBlockingQueue.java:42)
at org.fan.learn.thread.blockqueue.TestBlockingQueue.main(TestBlockingQueue.java:51)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:497)
at com.intellij.rt.execution.application.AppMain.main(AppMain.java:144)
Process finished with exit code -1
2.正常结束
Task(0)#liftOff
Task(1)#liftOff
Task(2)#liftOff
Task(3)#liftOff
Task(4)#liftOff
Finished Test
java.lang.InterruptedException
Exiting Liftoff Runner
at java.util.concurrent.SynchronousQueue.take(SynchronousQueue.java:928)
at org.fan.learn.thread.blockqueue.LiftOffRunner.run(TestBlockingQueue.java:28)
at java.lang.Thread.run(Thread.java:745)
Process finished with exit code 0
管道跟队列差不多，但是比起BlockingQueue来，不如BlockingQueue更健壮。当管道是空时，它会阻塞读取管道的线程。
两个任务操作同一个管道时，一个写端（PipeWriter）、一个读端（PipeReader）。注意，只能其中一个任务创建管道，而另一个管道与刚才创建的管道进行关联。管道这种IO方式可以被interrrupt()中断，而普通的IO是不能被interrupt中断的。
&&相关文章推荐
参考知识库
* 以上用户言论只代表其个人观点，不代表CSDN网站的观点或立场
访问：63645次
积分：2220
积分：2220
排名：第15495名
原创：126篇
转载：11篇
评论：23条
(6)(4)(1)(2)(7)(5)(2)(11)(4)(13)(17)(13)(12)(5)(1)(8)(7)(6)(4)(1)(2)(1)(3)(4)(2)多线程队列（Concurrent Queue）的使用场合非常多，高性能服务器中的消息队列，并行算法中的Work Stealing等都离不开它。对于一个队列来说有两个最主要的动作：添加（enqueue）和删除（dequeue）节点。在一个（或多个）线程在对一 个队列进行enqueue操作的同时可能会有一个（或多个）线程对这个队列进行dequeue操作。因为enqueue和dequeue都是对同一个队列里的节点进行操作，为了保证线程安全，一般在实现中都会在队列的结构体中加入一个队列锁（典型的如pthread_mutex_tq_lock），在进行enqueue和dequeue时都会先锁住这个锁以锁住整个队列然后再进行相关的操作。这样的设计如果实现的好的话一般性能就会很不错了。以链表实现的队列的结构体一般是这样的：
struct queue_t {
&&&&node_t *
&&&&node_t *ta
&&&&pthread_mutex_t q_
但是，这其中其实有一个潜在的性能瓶颈：enqueue和dequeue操作都要锁住整个队列，这在线程少的时候可能没什么问题，但是只要线程数一多，这个锁竞争所产生的性能瓶颈就会越来越严重。那么我们可不可以想办法优化一下这个算法呢？当然可以！如果我们仔细想一想enqueue和dequeue的具体 操作就会发现他们的操作其实不一定是冲突的。例如：如果所有的enqueue操作都是往队列的尾部插入新节点，而所有的dequeue操作都是从队列的头 部删除节点，那么enqueue和dequeue大部分时候都是相互独立的，我们大部分时候根本不需要锁住整个队列，白白损失性能！那么一个很自然就能想
到的算法优化方案就呼之欲出了：我们可以把那个队列锁拆成两个：一个队列头部锁（head lock)和一个队列尾部锁(tail lock)。这样这样的设计思路是对了，但是如果再仔细思考一下它的实现的话我们会发现其实不太容易，因为有两个特殊情况非常的tricky（难搞）：第 一种就是往空队列里插入第一个节点的时候，第二种就是从只剩最后一个节点的队列中删除那个“最后的果实”的时候。
为什么难搞呢？当我们向空队列中插入第一个节点的时候，我们需要同时修改队列的head和tail指针，使他们同时指向这个新插入的节点，换句话说，我们此 时即需要拿到headlock又需要拿到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 {
&&&&node_t *next
typedef struct queue_t {
&&&&NODE *
&&&&NODE *
&&&&LOCK q_h_
&&&&LOCK q_t_
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.&. 如果还想做更多优化的话可以参考这篇论文实现相应的Non Blocking版本的算法，性能还能有更多提升。当然了，这个算法早已被集成到java.util.concurrent里了（即 LinkedBlockingQueue），其他的并行库例如Intel的TBB多半也有类似的算法，如果大家能用上现成的库的话就不要再重复造轮子了。 为什么别造并行算法的轮子呢？因为高性能的并行算法实在太难正确地实现了，尤其是Non
Blocking，Lock Free之类的“火箭工程”。有多难呢？Doug Lea提到java.util.concurrent中一个Non Blocking的算法的实现大概需要1年的时间，总共约500行代码。所以，对最广大的程序员来说，别去写Non Blocking, LockFree的代码，只管用就行了，我看见网上很多的Non Blocking阿，无锁编程的算法实现啊什么的都非常地害怕，谁敢去用他们贴出来的这些代码啊？我之所以推荐这个two lock的算法是因为它的实现相对Non Blocking之类的来说容易多了，非常具备实用价值。虽然这篇论文出现的很早，但是我在看了几个开源软件中多线程队列的实现之后发现他们很多还是用的本文最开始提到的那种一个锁的算法。如果你想要实现更高性能的多线程队列的话，试试这个算法吧！
Update: 多线程队列算法有很多种，大家应根据不同的应用场合选取最优算法（例如是CPU密集型还是IO密集型）。本文所列的算法应用在这样一个多线程程序中：每个线程都拥有一个队列，每个队列可能被本线程进行dequeue操作，也可以被其他线程进行dequeue（即work stealing），线程数不超过CPU核心数，是一个典型的CPU/MEM密集型客户端单写者多读者场景。
&&相关文章推荐
参考知识库
* 以上用户言论只代表其个人观点，不代表CSDN网站的观点或立场
访问：7011次
排名：千里之外