Java并发包中的同步队列SynchronousQueue实现原理

2021-11-11 08:23:21

作者：一粟

介绍

Java 6的并发编程包中的SynchronousQueue是一个没有数据缓冲的BlockingQueue，生产者线程对其的插入操作put必须等待消费者的移除操作take，反过来也一样。

不像ArrayBlockingQueue或LinkedListBlockingQueue，SynchronousQueue内部并没有数据缓存空间，你不能调用peek()方法来看队列中是否有数据元素，因为数据元素只有当你试着取走的时候才可能存在，不取走而只想偷窥一下是不行的，当然遍历这个队列的操作也是不允许的。队列头元素是第一个排队要插入数据的线程，而不是要交换的数据。数据是在配对的生产者和消费者线程之间直接传递的，并不会将数据缓冲数据到队列中。可以这样来理解：生产者和消费者互相等待对方，握手，然后一起离开。

SynchronousQueue的一个使用场景是在线程池里。Executors.newCachedThreadPool()就使用了SynchronousQueue，这个线程池根据需要（新任务到来时）创建新的线程，如果有空闲线程则会重复使用，线程空闲了60秒后会被回收。

实现原理

阻塞队列的实现方法有许多：

阻塞算法实现

阻塞算法实现通常在内部采用一个锁来保证多个线程中的put()和take()方法是串行执行的。采用锁的开销是比较大的，还会存在一种情况是线程A持有线程B需要的锁，B必须一直等待A释放锁，即使A可能一段时间内因为B的优先级比较高而得不到时间片运行。所以在高性能的应用中我们常常希望规避锁的使用。

查看源代码 打印帮助
01 public class NativeSynchronousQueue<E> {
02     boolean putting = false;
03     E item = null;
04  
05     public synchronized E take() throws InterruptedException {
06         while (item == null)
07             wait();
08         E e = item;
09         item = null;
10         notifyAll();
11         return e;
12     }
13  
14     public synchronized void put(E e) throws InterruptedException {
15         if (e==null) return;
16         while (putting)
17             wait();
18         putting = true;
19         item = e;
20         notifyAll();
21         while (item!=null)
22             wait();
23         putting = false;
24         notifyAll();
25     }
26 }

信号量实现

经典同步队列实现采用了三个信号量，代码很简单，比较容易理解：

查看源代码 打印帮助
01 public class SemaphoreSynchronousQueue<E> {
02     E item = null;
03     Semaphore sync = new Semaphore(0);
04     Semaphore send = new Semaphore(1);
05     Semaphore recv = new Semaphore(0);
06  
07     public E take() throws InterruptedException {
08         recv.acquire();
09         E x = item;
10         sync.release();
11         send.release();
12         return x;
13     }
14  
15     public void put (E x) throws InterruptedException{
16         send.acquire();
17         item = x;
18         recv.release();
19         sync.acquire();
20     }
21 }

在多核机器上，上面方法的同步代价仍然较高，操作系统调度器需要上千个时间片来阻塞或唤醒线程，而上面的实现即使在生产者put()时已经有一个消费者在等待的情况下，阻塞和唤醒的调用仍然需要。

Java 5实现

查看源代码 打印帮助
01 public class Java5SynchronousQueue<E> {
02     ReentrantLock qlock = new ReentrantLock();
03     Queue waitingProducers = new Queue();
04     Queue waitingConsumers = new Queue();
05  
06     static class Node extends AbstractQueuedSynchronizer {
07         E item;
08         Node next;
09  
10         Node(Object x) { item = x; }
11         void waitForTake() { /* (uses AQS) */ }
12            E waitForPut() { /* (uses AQS) */ }
13     }
14  
15     public E take() {
16         Node node;
17         boolean mustWait;
18         qlock.lock();
19         node = waitingProducers.pop();
20         if(mustWait = (node == null))
21            node = waitingConsumers.push(null);
22          qlock.unlock();
23  
24         if (mustWait)
25            return node.waitForPut();
26         else
27             return node.item;
28     }
29  
30     public void put(E e) {
31          Node node;
32          boolean mustWait;
33          qlock.lock();
34          node = waitingConsumers.pop();
35          if (mustWait = (node == null))
36              node = waitingProducers.push(e);
37          qlock.unlock();
38  
39          if (mustWait)
40              node.waitForTake();
41          else
42             node.item = e;
43     }
44 }

Java 5的实现相对来说做了一些优化，只使用了一个锁，使用队列代替信号量也可以允许发布者直接发布数据，而不是要首先从阻塞在信号量处被唤醒。

Java6实现

Java 6的SynchronousQueue的实现采用了一种性能更好的无锁算法 — 扩展的“Dual stack and Dual queue”算法。性能比Java5的实现有较大提升。竞争机制支持公平和非公平两种：非公平竞争模式使用的数据结构是后进先出栈(Lifo Stack)；公平竞争模式则使用先进先出队列（Fifo Queue），性能上两者是相当的，一般情况下，Fifo通常可以支持更大的吞吐量，但Lifo可以更大程度的保持线程的本地化。

代码实现里的Dual Queue或Stack内部是用链表(LinkedList)来实现的，其节点状态为以下三种情况：

  1. 持有数据 – put()方法的元素
  2. 持有请求 – take()方法
  3. 空

这个算法的特点就是任何操作都可以根据节点的状态判断执行，而不需要用到锁。

其核心接口是Transfer，生产者的put或消费者的take都使用这个接口，根据第一个参数来区别是入列（栈）还是出列（栈）。

查看源代码 打印帮助
01 /**
02     * Shared internal API for dual stacks and queues.
03     */
04    static abstract class Transferer {
05        /**
06         * Performs a put or take.
07         *
08         * @param e if non-null, the item to be handed to a consumer;
09         *          if null, requests that transfer return an item
10         *          offered by producer.
11         * @param timed if this operation should timeout
12         * @param nanos the timeout, in nanoseconds
13         * @return if non-null, the item provided or received; if null,
14         *         the operation failed due to timeout or interrupt --
15         *         the caller can distinguish which of these occurred
16         *         by checking Thread.interrupted.
17         */
18        abstract Object transfer(Object e, boolean timed, long nanos);
19    }

TransferQueue实现如下(摘自Java 6源代码)，入列和出列都基于Spin和CAS方法：

查看源代码 打印帮助
01 /**
02     * Puts or takes an item.
03     */
04    Object transfer(Object e, boolean timed, long nanos) {
05        /* Basic algorithm is to loop trying to take either of
06         * two actions:
07         *
08         * 1. If queue apparently empty or holding same-mode nodes,
09         *    try to add node to queue of waiters, wait to be
10         *    fulfilled (or cancelled) and return matching item.
11         *
12         * 2. If queue apparently contains waiting items, and this
13         *    call is of complementary mode, try to fulfill by CAS'ing
14         *    item field of waiting node and dequeuing it, and then
15         *    returning matching item.
16         *
17         * In each case, along the way, check for and try to help
18         * advance head and tail on behalf of other stalled/slow
19         * threads.
20         *
21         * The loop starts off with a null check guarding against
22         * seeing uninitialized head or tail values. This never
23         * happens in current SynchronousQueue, but could if
24         * callers held non-volatile/final ref to the
25         * transferer. The check is here anyway because it places
26         * null checks at top of loop, which is usually faster
27         * than having them implicitly interspersed.
28         */
29  
30        QNode s = null; // constructed/reused as needed
31        boolean isData = (e != null);
32  
33        for (;;) {
34            QNode t = tail;
35            QNode h = head;
36            if (t == null || h == null)         // saw uninitialized value
37                continue;                       // spin
38  
39            if (h == t || t.isData == isData) { // empty or same-mode
40                QNode tn = t.next;
41                if (t != tail)                  // inconsistent read
42                    continue;
43                if (tn != null) {               // lagging tail
44                    advanceTail(t, tn);
45                    continue;
46                }
47                if (timed &amp;&amp; nanos &lt;= 0)        // can't wait
48                    return null;
49                if (s == null)
50                    s = new QNode(e, isData);
51                if (!t.casNext(null, s))        // failed to link in
52                    continue;
53  
54                advanceTail(t, s);              // swing tail and wait
55                Object x = awaitFulfill(s, e, timed, nanos);
56                if (x == s) {                   // wait was cancelled
57                    clean(t, s);
58                    return null;
59                }
60  
61                if (!s.isOffList()) {           // not already unlinked
62                    advanceHead(t, s);          // unlink if head
63                    if (x != null)              // and forget fields
64                        s.item = s;
65                    s.waiter = null;
66                }
67                return (x != null)? x : e;
68  
69            } else {                            // complementary-mode
70                QNode m = h.next;               // node to fulfill
71                if (t != tail || m == null || h != head)
72                    continue;                   // inconsistent read
73  
74                Object x = m.item;
75                if (isData == (x != null) ||    // m already fulfilled
76                    x == m ||                   // m cancelled
77                    !m.casItem(x, e)) {         // lost CAS
78                    advanceHead(h, m);          // dequeue and retry
79                    continue;
80                }
81  
82                advanceHead(h, m);              // successfully fulfilled
83                LockSupport.unpark(m.waiter);
84                return (x != null)? x : e;
85            }
86        }
87    }
上一篇：Access转成Sql 2008步骤，同时解决自动编号问题，主键，id数值不重置。


下一篇：Postman教程-变量、集合的基本使用