理解Java并发工具类SynchronousQueue

SynchronousQueue类是JDK5中引入的一个同步队列，这个类比较特殊，因为它虽然是一个队列但实际上并不真正的存储数据，仅仅维护一个线程配对的队列列表等待其入队和出队，与上篇文章中我们已经介绍过Exchanger类的功能比较类似，准确点讲与上篇我们使用Exchanger类构建的生产者消费者的队列非常类似，只不过SynchronousQueue更加完善提供了队列里面一些基本的操作。

SynchronousQueue基本原理是构建在配对或者叫直接交接的机制上其与Java里面另外常见的两个阻塞队列LinkedBlockingQueue和ArrayBlockingQueue是不同的，在SynchronousQueue里面，生产者要对自己生产的消息负责，如果自己生产的消息没有消费者消费，那么生产者就要一直阻塞直到有消费者和自己配对成功，直接接头成功才算交易完成。反之也一样，一个消费者去消费消息，如果找不到配对的生产者那么其也要一直阻塞，直到有生产者和自己匹配成功。

这种机制看似非常的不合理，但其确实有其使用的场景，尤其是消息或者任务的传递，这种模式更像CSP的并发模型Communicating Sequential Process，这里引申一下一些语言里面的并发机制，最典型有两种：

Scala里面的Actor模型

Go语言的里面CSP模型

主要的不同点在于关于消息发送方和接收方：

Actor：注重的处理单元，也就是Actor，而不是消息传送方式。发送消息时，都需要知道对方是谁，这里的“都需要知道对方是谁”的意思，当ActorX要给ActorY发消息时，必须明确知道ActorY的地址。ActorY接收到消息时，就能够知道消息发送者（ActorX）的地址。返回消息给发送者时，只需要按发送者的地址往回传消息就行。

CSP：注重的是消息传送方式（channel），不关心发送的人和接收的人是谁。 向channel写消息的人，不知道消息的接收者是谁；读消息的人，也不知道消息的写入者是谁。

SynchronousQueue其实非常类似CSP的模型，生产者和消费者都不知道对方，重点在于队列这个通道，下面我们通过一个简单的例子看一下如何使用SynchronousQueue：

public static void main(String[] args) {

        SynchronousQueue<Integer> synchronousQueue=new SynchronousQueue();



        new Thread(()->{
            int count=0;
            while (true){

                try {

                    System.out.println(Thread.currentThread().getName()+" 放入了"+count);
                    synchronousQueue.put(count);

                    count++;
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }


        }).start();


        new Thread(()->{

            while (true){

                try {
                    Thread.sleep(2000);
                    int count=synchronousQueue.take();
                    System.out.println(Thread.currentThread().getName()+" 消费了"+count);


                    count++;
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }


        }).start();
    }

输出结果：

Thread-0 放入了0
Thread-1 消费了0
Thread-0 放入了1
Thread-1 消费了1
Thread-0 放入了2
Thread-1 消费了2
Thread-0 放入了3
Thread-1 消费了3
Thread-0 放入了4
Thread-1 消费了4
Thread-0 放入了5
Thread-1 消费了5
Thread-0 放入了6
Thread-1 消费了6

举个生活中的例子，在饭店里面，负责收拾饭后餐桌上的盘子上的阿姨把盘子放到清洗间的货架上等待洗盘子的阿姨洗盘子，这里面货架就充当了队列容器，而LinkedBlockingQueue和ArrayBlockingQueue就是基于这种方式实现的，这样的好处是生产者与消费者解耦了，但如果队列满了，生产者就要等待，反之队列空了，消费者就要等待。

而SynchronousQueue则是相当于这里没有货架，收拾餐桌盘子的阿姨在把盘子拿到清洗间之后直接交给洗盘子的阿姨，注意这里是一对一的配对，如果没有配对成功另一方必须一直等待直到有人生产或者消费者，看起来这种方法很奇怪，但其减少了生产者到消费者之间的延迟，因为是直接交付。如果不是基于这种方式，那么则就必须先入队在出队，出队操作必须等待入队操作完成之后才能进行。

关于对列的一些常见方法如下：

add         增加一个元索                      如果队列已满，则抛出一个IIIegaISlabEepeplian异常
remove   移除并返回队列头部的元素     如果队列为空，则抛出一个NoSuchElementException异常
element  返回队列头部的元素              如果队列为空，则抛出一个NoSuchElementException异常
offer       添加一个元素并返回true        如果队列已满，则返回false
poll         移除并返问队列头部的元素     如果队列为空，则返回null
peek       返回队列头部的元素              如果队列为空，则返回null
put         添加一个元素                       如果队列满，则阻塞
take        移除并返回队列头部的元素

SynchronousQueue虽说也是一个无界队列，但其支持的api方法非常有限，常用的有put和take方法，这两个方法如果不能成对出现，那么任何一方都会先自旋一小会，然后进入阻塞直到配对成功处理。因为不存储具体数据所以size和iteritor等一些方法都是无效的，具体的可以参考官网文档，此外SynchronousQueue支持公平和非公平模式底层分别通过队列和栈来实现，我们可以通过构造函数来声明公平模式。

在Java里面线程池Executors.newCachedThreadPool就是使用SynchronousQueue类在内部构建，创建的线程数量仅仅是按需所取，使用SynchronousQueue的目的就是保证“对于提交的任务，如果有空闲线程，则使用空闲线程来处理；否则新建一个线程来处理任务”。

本文主要介绍了Java里面比较特殊的一种阻塞队列SynchronousQueue，其设计的主要目的是为了处理一些handoff（切换/传递）的设计，指的是一个线程的信息，事件或者任务要传递到另一个线程中处理，这种处理方式与CSP的并发机制比较类似，并且在队列和线程数量为1的时候相比其他的阻塞队列，其吞吐量非常出色，所以我们可以在一些适合的场景下使用这个特别的同步阻塞队列。