SynchronousQueue详解

SynchronousQueue详解

简介

SynchronousQueue是BlockingQueue的一种，因此SynchronousQueue是线程安全的。SynchronousQueue和其余的BlockingQueue不一样的是SynchronousQueue的capacity是0。即SynchronousQueue不存储任何元素。

也就是说SynchronousQueue的每一次insert操做，必须等待其余线性的remove操做。而每个remove操做也必须等待其余线程的insert操做。

这种特性可让咱们想起了Exchanger。和Exchanger不一样的是，使用SynchronousQueue能够在两个线程中传递同一个对象。一个线程放对象，另一个线程取对象。

举例说明

咱们举一个多线程中传递对象的例子。仍是举生产者消费者的例子，在生产者中咱们建立一个对象，在消费者中咱们取出这个对象。先看一下用CountDownLatch该怎么作：

@Test
    public void useCountdownLatch() throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(2);
        AtomicReference<Object> atomicReference= new AtomicReference<>();
        CountDownLatch countDownLatch = new CountDownLatch(1);

        Runnable producer = () -> {
            Object object=new Object();
            atomicReference.set(object);
            log.info("produced {}",object);
            countDownLatch.countDown();
        };

        Runnable consumer = () -> {
            try {
                countDownLatch.await();
                Object object = atomicReference.get();
                log.info("consumed {}",object);
            } catch (InterruptedException ex) {
                log.error(ex.getMessage(),ex);
            }
        };

        executor.submit(producer);
        executor.submit(consumer);

        executor.awaitTermination(50000, TimeUnit.MILLISECONDS);
        executor.shutdown();
    }

上例中，咱们使用AtomicReference来存储要传递的对象，而且定义了一个型号量为1的CountDownLatch。

在producer中，咱们存储对象，而且countDown。

在consumer中，咱们await,而后取出对象。

输出结果：

[pool-1-thread-1] INFO com.flydean.SynchronousQueueUsage - produced java.lang.Object@683d1b4b
[pool-1-thread-2] INFO com.flydean.SynchronousQueueUsage - consumed java.lang.Object@683d1b4b

能够看到传入和输出了同一个对象。

上面的例子咱们也能够用SynchronousQueue来改写：

@Test
    public void useSynchronousQueue() throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(2);
        SynchronousQueue<Object> synchronousQueue=new SynchronousQueue<>();

        Runnable producer = () -> {
            Object object=new Object();
            try {
                synchronousQueue.put(object);
            } catch (InterruptedException ex) {
                log.error(ex.getMessage(),ex);
            }
            log.info("produced {}",object);
        };

        Runnable consumer = () -> {
            try {
                Object object = synchronousQueue.take();
                log.info("consumed {}",object);
            } catch (InterruptedException ex) {
                log.error(ex.getMessage(),ex);
            }
        };

        executor.submit(producer);
        executor.submit(consumer);

        executor.awaitTermination(50000, TimeUnit.MILLISECONDS);
        executor.shutdown();
    }

上面的例子中，若是咱们使用synchronousQueue，则能够不用手动同步，也不须要额外的存储。

总结

若是咱们须要在代码中用到这种线程中传递对象的状况，那么使用synchronousQueue吧。

本文的例子https://github.com/ddean2009/learn-java-collections

欢迎关注个人公众号:程序那些事，更多精彩等着您！
更多内容请访问 www.flydean.com