1 简介
SynchronousQueue是这样一种阻塞队列，其中每个put必须等待一个take，反之亦然。同步队列没有任何内部容量，甚至连一个队列的容量都没有。不能在同步队列上进行peek，因为仅在试图要取得元素时，该元素才存在，除非另一个线程试图移除某个元素，否则也不能（使用任何方法）添加元素，也不能迭代队列，因为其中没有元素可用于迭代。队列的头是尝试添加到队列中的首个已排队线程元素，如果没有已排队线程，则不添加元素并且头为 null。
对于其他Collection方法（例如 contains），SynchronousQueue作为一个空集合，此队列不允许 null 元素。
同步队列类似于CSP和Ada中使用的rendezvous信道。它非常适合于传递性设计，在这种设计中，在一个线程中运行的对象要将某些信息、事件或任务传递给在另一个线程中运行的对象，它就必须与该对象同步。
对于正在等待的生产者和使用者线程而言，此类支持可选的公平排序策略，默认情况下不保证这种排序。
但是，使用公平设置为true所构造的队列可保证线程以FIFO的顺序进行访问。公平通常会降低吞吐量，但是可以减小可变性并避免得不到服务。
2 使用示例
1 import static org.junit.Assert.assertEquals; 2 3 import java.util.concurrent.CountDownLatch; 4 import java.util.concurrent.ExecutorService; 5 import java.util.concurrent.Executors; 6 import java.util.concurrent.SynchronousQueue; 7 import java.util.concurrent.ThreadLocalRandom; 8 import java.util.concurrent.TimeUnit; 9 import java.util.concurrent.atomic.AtomicInteger; 10 11 import org.junit.Test; 12 13 /** 14 * synchronousqueue的使用场景 ==== 线程间共享元素 15 * 假设有两个线程，一个生产者和一个消费者，当生产者设置一个共享变量的值时，我们希望向消费者线程 16 * 发出这个信号，然后消费者线程将从共享变量取值。 17 * @author ko 18 * 19 */ 20 public class Sqt { 21 22 /** 23 * 利用AtomicInteger+CountDownLatch实现 24 */ 25 @Test 26 public void doingByCountDownLatch(){ 27 ExecutorService executor = Executors.newFixedThreadPool(2); 28 AtomicInteger sharedState = new AtomicInteger();// 共享变量值 29 // 协调这两个线程，以防止情况当消费者访问共享变量值 30 CountDownLatch countDownLatch = new CountDownLatch(1); 31 32 // 生产商将设置一个随机整数到sharedstate变量，并countdown()方法， 33 // 信号给消费者，它可以从sharedstate取一个值 34 Runnable producer = () -> {// 这好像是java8的匿名内部类的新写法 35 Integer producedElement = ThreadLocalRandom.current().nextInt(); 36 sharedState.set(producedElement); 37 System.out.println("生产者给变量设值："+producedElement); 38 countDownLatch.countDown(); 39 }; 40 41 // 消费者会等待countdownlatch执行到await()方法，获取许可后，再从生产者里获取变量sharedstate值 42 Runnable consumer = () -> { 43 try { 44 countDownLatch.await(); 45 Integer consumedElement = sharedState.get(); 46 System.out.println("消费者获取到变量："+consumedElement); 47 } catch (InterruptedException ex) { 48 ex.printStackTrace(); 49 } 50 }; 51 52 executor.execute(producer); 53 executor.execute(consumer); 54 try { 55 executor.awaitTermination(500, TimeUnit.MILLISECONDS); 56 } catch (InterruptedException e) { 57 e.printStackTrace(); 58 } 59 executor.shutdown(); 60 assertEquals(countDownLatch.getCount(), 0); 61 } 62 63 /** 64 * 仅使用SynchronousQueue就可以实现 65 */ 66 @Test 67 public void doingBySynchronousQueue(){ 68 ExecutorService executor = Executors.newFixedThreadPool(2); 69 SynchronousQueue<Integer> queue = new SynchronousQueue<>(); 70 71 // 生产者 72 Runnable producer = () -> { 73 Integer producedElement = ThreadLocalRandom.current().nextInt(); 74 try { 75 queue.put(producedElement); 76 System.out.println("生产者设值："+producedElement); 77 } catch (InterruptedException ex) { 78 ex.printStackTrace(); 79 } 80 }; 81 82 // 消费者 83 Runnable consumer = () -> { 84 try { 85 Integer consumedElement = queue.take(); 86 System.out.println("消费者取值："+consumedElement); 87 } catch (InterruptedException ex) { 88 ex.printStackTrace(); 89 } 90 }; 91 92 executor.execute(producer); 93 executor.execute(consumer); 94 try { 95 executor.awaitTermination(500, TimeUnit.MILLISECONDS); 96 } catch (InterruptedException e) { 97 e.printStackTrace(); 98 } 99 executor.shutdown(); 100 assertEquals(queue.size(), 0); 101 } 102 }
3 实现原理
阻塞队列的实现方法有许多。
3.1 阻塞算法实现
阻塞算法实现通常在内部采用一个锁来保证多个线程中的put()和take()方法是串行执行的。采用锁的开销是比较大的，还会存在一种情况是线程A持有线程B需要的锁，B必须一直等待A释放锁，即使A可能一段时间内因为B的优先级比较高而得不到时间片运行。所以在高性能的应用中我们常常希望规避锁的使用。
1 public class NativeSynchronousQueue<E> { 2 boolean putting = false; 3 E item = null; 4 5 public synchronized E take() throws InterruptedException { 6 while (item == null) 7 wait(); 8 E e = item; 9 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 }
3.2 信号量实现
经典同步队列实现采用了三个信号量，代码很简单，比较容易理解。
1 public class SemaphoreSynchronousQueue<E> { 2 E item = null; 3 Semaphore sync = new Semaphore(0); 4 Semaphore send = new Semaphore(1); 5 Semaphore recv = new Semaphore(0); 6 7 public E take() throws InterruptedException { 8 recv.acquire(); 9 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()时已经有一个消费者在等待的情况下，阻塞和唤醒的调用仍然需要。
3.3 Java 5实现
1 public class Java5SynchronousQueue<E> { 2 ReentrantLock qlock = new ReentrantLock(); 3 Queue waitingProducers = new Queue(); 4 Queue waitingConsumers = new Queue(); 5 6 static class Node extends AbstractQueuedSynchronizer { 7 E item; 8 Node next; 9 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的实现相对来说做了一些优化，只使用了一个锁，使用队列代替信号量也可以允许发布者直接发布数据，而不是要首先从阻塞在信号量处被唤醒。
3.4 Java 6实现
Java 6的SynchronousQueue的实现采用了一种性能更好的无锁算法 — 扩展的“Dual stack and Dual queue”算法。性能比Java5的实现有较大提升。竞争机制支持公平和非公平两种：非公平竞争模式使用的数据结构是后进先出栈(Lifo Stack)；公平竞争模式则使用先进先出队列（Fifo Queue），性能上两者是相当的，一般情况下，Fifo通常可以支持更大的吞吐量，但Lifo可以更大程度的保持线程的本地化。
代码实现里的Dual Queue或Stack内部是用链表(LinkedList)来实现的，其节点状态为以下三种情况：
持有数据 – put()方法的元素
持有请求 – take()方法
这个算法的特点就是任何操作都可以根据节点的状态判断执行，而不需要用到锁。
其核心接口是Transfer，生产者的put或消费者的take都使用这个接口，根据第一个参数来区别是入列（栈）还是出列（栈）。
1 /** 2 * Shared internal API for dual stacks and queues. 3 */ 4 static abstract class Transferer { 5 /** 6 * Performs a put or take. 7 * 8 * @param e if non-null, the item to be handed to a consumer; 9 * 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方法：
1 /** 2 * Puts or takes an item. 3 */ 4 Object transfer(Object e, boolean timed, long nanos) { 5 /* Basic algorithm is to loop trying to take either of 6 * two actions: 7 * 8 * 1. If queue apparently empty or holding same-mode nodes, 9 * 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 && nanos <= 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 }
