快捷導航

Java中的SynchronousQueue隊列詳解

更新時間：2023年12月07日 09:00:33 作者：flydean程序那些事

這篇文章主要介紹了Java中的SynchronousQueue隊列詳解,SynchronousQueue是BlockingQueue的一種,所以SynchronousQueue是線程安全的,SynchronousQueue和其他的BlockingQueue不同的是SynchronousQueue的capacity是0,需要的朋友可以參考下

簡介

SynchronousQueue是BlockingQueue的一種，所以SynchronousQueue是線程安全的。

SynchronousQueue和其他的BlockingQueue不同的是SynchronousQueue的capacity是0。

即SynchronousQueue不存儲任何元素。

也就是說SynchronousQueue的每一次insert操作，必須等待其他線性的remove操作。

而每一個remove操作也必須等待其他線程的insert操作。

這種特性可以讓我們想起了Exchanger。和Exchanger不同的是，使用SynchronousQueue可以在兩個線程中傳遞同一個對象。

一個線程放對象，另外一個線程取對象。

舉例說明

我們舉一個多線程中傳遞對象的例子。

還是舉生產(chǎn)者消費者的例子，在生產(chǎn)者中我們創(chuàng)建一個對象，在消費者中我們?nèi)〕鲞@個對象。

先看一下用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,然后取出對象。

輸出結(jié)果：

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