Java并发包中SynchronousQueue的实现原理

作者: 小米

Java 6并发编程包中的SynchronousQueue是一个没有数据缓冲的BlockingQueue. 生产者线程的插入操作必须等待使用者的删除操作接受，反之亦然.

与ArrayBlockingQueue或LinkedListBlockingQueue不同，SynchronousQueue内部没有数据缓存空间，您无法调用peek（）方法来查看队列中是否存在数据元素，因为该数据元素仅在您尝试将其删除时存在，否仅仅接受窥视还不够. 当然，不允许遍历此队列的操作. 队列的头元素是排队插入数据而不是要交换的数据的第一个线程. 数据直接在成对的生产者线程和使用者线程之间传递，并且数据不缓冲到队列中. 可以这样理解: 生产者和消费者互相等待，握手，然后走在一起.

线程池中有一个SynchronousQueue的使用场景. Executors.newCachedThreadPool（）使用SynchronousQueue. 该线程池根据需要（当新任务到达时）创建新线程. 如果有空闲线程，它们将被重用. 线程空闲60秒后将被回收.

有很多方法可以实现阻塞队列:

阻塞算法实现通常在内部使用锁java queue的实现类，以确保多个线程中的put（）和take（）方法被串行执行. 使用锁的开销相对较大，并且可能存程A持有线程B所需的锁的情况. 即使由于B的优先级较高而导致A可能无法获得B，B也必须等待A释放锁. 一段时间. 运行到时间片. 因此，在高性能应用程序中，我们经常想避免使用锁.

public class NativeSynchronousQueue<E> {
    boolean putting = false;
    E item = null;
    public synchronized E take() throws InterruptedException {
        while (item == null)
            wait();
        E e = item;
        item = null;
        notifyAll();
        return e;
    }
    public synchronized void put(E e) throws InterruptedException {
        if (e==null) return;
        while (putting)
            wait();
        putting = true;
        item = e;
        notifyAll();
        while (item!=null)
            wait();
        putting = false;
        notifyAll();
    }
}

经典同步队列实现使用三个信号量. 该代码非常简单并且相对容易理解:

public class SemaphoreSynchronousQueue<E> {
    E item = null;
    Semaphore sync = new Semaphore(0);
    Semaphore send = new Semaphore(1);
    Semaphore recv = new Semaphore(0);
    public E take() throws InterruptedException {
        recv.acquire();
        E x = item;
        sync.release();
        send.release();
        return x;
    }
    public void put (E x) throws InterruptedException{
        send.acquire();
        item = x;
        recv.release();
        sync.acquire();
    }
}

在多核计算机上，上述方法的同步成本仍然很高. 操作系统调度程序需要数千个时间片来阻塞或唤醒线程，并且即使在生产者put（）处于等待状态的情况下，上述实现甚至有一个使用者java queue的实现类，即使在等待时，仍然需要阻塞和唤醒调用.

public class Java5SynchronousQueue<E> {
    ReentrantLock qlock = new ReentrantLock();
    Queue waitingProducers = new Queue();
    Queue waitingConsumers = new Queue();
    static class Node extends AbstractQueuedSynchronizer {
        E item;
        Node next;
        Node(Object x) { item = x; }
        void waitForTake() { /* (uses AQS) */ }
           E waitForPut() { /* (uses AQS) */ }
    }
    public E take() {
        Node node;
        boolean mustWait;
        qlock.lock();
        node = waitingProducers.pop();
        if(mustWait = (node == null))
           node = waitingConsumers.push(null);
         qlock.unlock();
        if (mustWait)
           return node.waitForPut();
        else
            return node.item;
    }
    public void put(E e) {
         Node node;
         boolean mustWait;
         qlock.lock();
         node = waitingConsumers.pop();
         if (mustWait = (node == null))
             node = waitingProducers.push(e);
         qlock.unlock();
         if (mustWait)
             node.waitForTake();
         else
            node.item = e;
    }
}

对Java 5的实现进行了相对优化，仅使用一个锁，并且使用队列而不是信号量还可以使发布者直接发布数据，而不是首先被唤醒以阻止信号量.

Java 6中SynchronousQueue的实现使用性能更好的无锁算法-扩展的“双栈和双队列”算法. 与Java5的实现相比，性能得到了极大的提高. 竞争机制支持公平和不公平: 非公平竞争模式使用的数据结构是后进先出堆栈（Lifo Stack）；公平竞争模式使用先进先出队列（Fifo Queue），两者的性能相当. 通常，Fifo通常可以支持更大的吞吐量，但是Lifo可以更大程度地保持线程本地化.

代码实现中的双队列或堆栈是通过链表（LinkedList）实现的，其节点状态在以下三种情况下:

put（）方法-take（）方法的数据元素保持请求为空

该算法的特点是可以根据节点的状态判断和执行任何操作，而无需使用锁.

核心接口是传输，供生产者的看跌期权或消费者的看跌期权使用. 根据第一个参数，可以区分是入队（堆栈）还是出队（堆栈）.

 /**
     * Shared internal API for dual stacks and queues.
     */
    static abstract class Transferer {
        /**
         * Performs a put or take.
         *
         * @param e if non-null, the item to be handed to a consumer;
         *          if null, requests that transfer return an item
         *          offered by producer.
         * @param timed if this operation should timeout
         * @param nanos the timeout, in nanoseconds
         * @return if non-null, the item provided or received; if null,
         *         the operation failed due to timeout or interrupt --
         *         the caller can distinguish which of these occurred
         *         by checking Thread.interrupted.
         */
        abstract Object transfer(Object e, boolean timed, long nanos);
    }

TransferQueue的实现如下（来自Java 6源代码），输入和输出均基于Spin和CAS方法:

     /**
         * Puts or takes an item.
         */
        Object transfer(Object e, boolean timed, long nanos) {
            /* Basic algorithm is to loop trying to take either of
             * two actions:
             *
             * 1. If queue apparently empty or holding same-mode nodes,
             *    try to add node to queue of waiters, wait to be
             *    fulfilled (or cancelled) and return matching item.
             *
             * 2. If queue apparently contains waiting items, and this
             *    call is of complementary mode, try to fulfill by CAS'ing
             *    item field of waiting node and dequeuing it, and then
             *    returning matching item.
             *
             * In each case, along the way, check for and try to help
             * advance head and tail on behalf of other stalled/slow
             * threads.
             *
             * The loop starts off with a null check guarding against
             * seeing uninitialized head or tail values. This never
             * happens in current SynchronousQueue, but could if
             * callers held non-volatile/final ref to the
             * transferer. The check is here anyway because it places
             * null checks at top of loop, which is usually faster
             * than having them implicitly interspersed.
             */
            QNode s = null; // constructed/reused as needed
            boolean isData = (e != null);
            for (;;) {
                QNode t = tail;
                QNode h = head;
                if (t == null || h == null)         // saw uninitialized value
                    continue;                       // spin
                if (h == t || t.isData == isData) { // empty or same-mode
                    QNode tn = t.next;
                    if (t != tail)                  // inconsistent read
                        continue;
                    if (tn != null) {               // lagging tail
                        advanceTail(t, tn);
                        continue;
                    }
                    if (timed && nanos <= 0)        // can't wait
                        return null;
                    if (s == null)
                        s = new QNode(e, isData);
                    if (!t.casNext(null, s))        // failed to link in
                        continue;
                    advanceTail(t, s);              // swing tail and wait
                    Object x = awaitFulfill(s, e, timed, nanos);
                    if (x == s) {                   // wait was cancelled
                        clean(t, s);
                        return null;
                    }
                    if (!s.isOffList()) {           // not already unlinked
                        advanceHead(t, s);          // unlink if head
                        if (x != null)              // and forget fields
                            s.item = s;
                        s.waiter = null;
                    }
                    return (x != null)? x : e;
                } else {                            // complementary-mode
                    QNode m = h.next;               // node to fulfill
                    if (t != tail || m == null || h != head)
                        continue;                   // inconsistent read
                    Object x = m.item;
                    if (isData == (x != null) ||    // m already fulfilled
                        x == m ||                   // m cancelled
                        !m.casItem(x, e)) {         // lost CAS
                        advanceHead(h, m);          // dequeue and retry
                        continue;
                    }
                    advanceHead(h, m);              // successfully fulfilled
                    LockSupport.unpark(m.waiter);
                    return (x != null)? x : e;
                }
            }
        }

SynchronousQueueScalable同步队列的Javadoc具有条件同步的非阻塞并发数据结构

本文来自电脑杂谈，转载请注明本文网址：
http://www.pc-fly.com/a/jisuanjixue/article-264327-1.html