SynchronousQueue詳解

時間 2020-05-12

標籤 synchronousqueue 詳解简体版

原文原文鏈接

SynchronousQueue詳解java

簡介

SynchronousQueue是BlockingQueue的一種，因此SynchronousQueue是線程安全的。SynchronousQueue和其餘的BlockingQueue不一樣的是SynchronousQueue的capacity是0。即SynchronousQueue不存儲任何元素。git

也就是說SynchronousQueue的每一次insert操做，必須等待其餘線性的remove操做。而每個remove操做也必須等待其餘線程的insert操做。github

這種特性可讓咱們想起了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。 atom

在producer中，咱們存儲對象，而且countDown。線程

在consumer中，咱們await,而後取出對象。3d

輸出結果：code

[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，則能夠不用手動同步，也不須要額外的存儲。