快捷導(dǎo)航

Java中的SynchronousQueue隊(duì)列詳解

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

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

簡介

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

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

即SynchronousQueue不存儲(chǔ)任何元素。

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

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

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

一個(gè)線程放對(duì)象，另外一個(gè)線程取對(duì)象。

舉例說明

我們舉一個(gè)多線程中傳遞對(duì)象的例子。

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

先看一下用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來存儲(chǔ)要傳遞的對(duì)象，并且定義了一個(gè)型號(hào)量為1的CountDownLatch。

在producer中，我們存儲(chǔ)對(duì)象，并且countDown。

在consumer中，我們await,然后取出對(duì)象。

輸出結(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

可以看到傳入和輸出了同一個(gè)對(duì)象。

上面的例子我們也可以用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，則可以不用手動(dòng)同步，也不需要額外的存儲(chǔ)。