前言

SynchronousQueue是一個比較特別的隊列，由于在線程池方面有所應(yīng)用，為了更好的理解線程池的實現(xiàn)原理，筆者花了些時間學(xué)習(xí)了一下該隊列源碼(JDK1.8)，此隊列源碼中充斥著大量的CAS語句，理解起來是有些難度的，為了方便日后回顧，本篇文章會以簡潔的圖形化方式展示該隊列底層的實現(xiàn)原理。

SynchronousQueue簡單使用

經(jīng)典的生產(chǎn)者-消費者模式，操作流程是這樣的：

有多個生產(chǎn)者，可以并發(fā)生產(chǎn)產(chǎn)品，把產(chǎn)品置入隊列中，如果隊列滿了，生產(chǎn)者就會阻塞；

有多個消費者，并發(fā)從隊列中獲取產(chǎn)品，如果隊列空了，消費者就會阻塞；

如下面的示意圖所示：

SynchronousQueue 也是一個隊列來的，但它的特別之處在于它內(nèi)部沒有容器，一個生產(chǎn)線程，當(dāng)它生產(chǎn)產(chǎn)品（即put的時候），如果當(dāng)前沒有人想要消費產(chǎn)品(即當(dāng)前沒有線程執(zhí)行take)，此生產(chǎn)線程必須阻塞，等待一個消費線程調(diào)用take操作，take操作將會喚醒該生產(chǎn)線程，同時消費線程會獲取生產(chǎn)線程的產(chǎn)品（即數(shù)據(jù)傳遞），這樣的一個過程稱為一次配對過程(當(dāng)然也可以先take后put,原理是一樣的)。

我們用一個簡單的代碼來驗證一下，如下所示：

package com.concurrent;
import java.util.concurrent.SynchronousQueue;
public class SynchronousQueueDemo {
    public static void main(String[] args) throws InterruptedException {
        final SynchronousQueue<Integer> queue = new SynchronousQueue<Integer>();
        Thread putThread = new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println("put thread start");
                try {
                    queue.put(1);
                } catch (InterruptedException e) {
                }
                System.out.println("put thread end");
            }
        });
        Thread takeThread = new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println("take thread start");
                try {
                    System.out.println("take from putThread: " + queue.take());
                } catch (InterruptedException e) {
                }
                System.out.println("take thread end");
            }
        });
        putThread.start();
        Thread.sleep(1000);
        takeThread.start();
    }
}

一種輸出結(jié)果如下:

put thread start
take thread start
take from putThread: 1
put thread end
take thread end

從結(jié)果可以看出，put線程執(zhí)行queue.put(1) 后就被阻塞了，只有take線程進行了消費，put線程才可以返回?？梢哉J(rèn)為這是一種線程與線程間一對一傳遞消息的模型。

SynchronousQueue實現(xiàn)原理

不像ArrayBlockingQueue、LinkedBlockingDeque之類的阻塞隊列依賴AQS實現(xiàn)并發(fā)操作，SynchronousQueue直接使用CAS實現(xiàn)線程的安全訪問。由于源碼中充斥著大量的CAS代碼，不易于理解，所以按照筆者的風(fēng)格，接下來會使用簡單的示例來描述背后的實現(xiàn)模型。

隊列的實現(xiàn)策略通常分為公平模式和非公平模式，接下來將分別進行說明。

公平模式下的模型：

公平模式下，底層實現(xiàn)使用的是TransferQueue這個內(nèi)部隊列，它有一個head和tail指針，用于指向當(dāng)前正在等待匹配的線程節(jié)點。 初始化時，TransferQueue的狀態(tài)如下：

接著我們進行一些操作：

1、線程put1執(zhí)行 put(1)操作，由于當(dāng)前沒有配對的消費線程，所以put1線程入隊列，自旋一小會后睡眠等待，這時隊列狀態(tài)如下：

2、接著，線程put2執(zhí)行了put(2)操作，跟前面一樣，put2線程入隊列，自旋一小會后睡眠等待，這時隊列狀態(tài)如下：

3、這時候，來了一個線程take1，執(zhí)行了 take操作，由于tail指向put2線程，put2線程跟take1線程配對了(一put一take)，這時take1線程不需要入隊，但是請注意了，這時候，要喚醒的線程并不是put2，而是put1。為何？ 大家應(yīng)該知道我們現(xiàn)在講的是公平策略，所謂公平就是誰先入隊了，誰就優(yōu)先被喚醒，我們的例子明顯是put1應(yīng)該優(yōu)先被喚醒。至于讀者可能會有一個疑問，明明是take1線程跟put2線程匹配上了，結(jié)果是put1線程被喚醒消費，怎么確保take1線程一定可以和次首節(jié)點(head.next)也是匹配的呢？其實大家可以拿個紙畫一畫，就會發(fā)現(xiàn)真的就是這樣的。 公平策略總結(jié)下來就是：隊尾匹配隊頭出隊。 執(zhí)行后put1線程被喚醒，take1線程的 take()方法返回了1(put1線程的數(shù)據(jù))，這樣就實現(xiàn)了線程間的一對一通信，這時候內(nèi)部狀態(tài)如下：

4、最后，再來一個線程take2，執(zhí)行take操作，這時候只有put2線程在等候，而且兩個線程匹配上了，線程put2被喚醒， take2線程take操作返回了2(線程put2的數(shù)據(jù))，這時候隊列又回到了起點，如下所示：

以上便是公平模式下，SynchronousQueue的實現(xiàn)模型?？偨Y(jié)下來就是：隊尾匹配隊頭出隊，先進先出，體現(xiàn)公平原則。

非公平模式下的模型：

我們還是使用跟公平模式下一樣的操作流程，對比兩種策略下有何不同。非公平模式底層的實現(xiàn)使用的是TransferStack， 一個棧，實現(xiàn)中用head指針指向棧頂，接著我們看看它的實現(xiàn)模型:

1、線程put1執(zhí)行 put(1)操作，由于當(dāng)前沒有配對的消費線程，所以put1線程入棧，自旋一小會后睡眠等待，這時棧狀態(tài)如下：

2、接著，線程put2再次執(zhí)行了put(2)操作，跟前面一樣，put2線程入棧，自旋一小會后睡眠等待，這時棧狀態(tài)如下：

3、這時候，來了一個線程take1，執(zhí)行了take操作，這時候發(fā)現(xiàn)棧頂為put2線程，匹配成功，但是實現(xiàn)會先把take1線程入棧，然后take1線程循環(huán)執(zhí)行匹配put2線程邏輯，一旦發(fā)現(xiàn)沒有并發(fā)沖突，就會把棧頂指針直接指向 put1線程

4、最后，再來一個線程take2，執(zhí)行take操作，這跟步驟3的邏輯基本是一致的，take2線程入棧，然后在循環(huán)中匹配put1線程，最終全部匹配完畢，棧變?yōu)榭?，恢?fù)初始狀態(tài)，如下圖所示：

可以從上面流程看出，雖然put1線程先入棧了，但是卻是后匹配，這就是非公平的由來。

總結(jié)

SynchronousQueue由于其獨有的線程一一配對通信機制，在大部分平常開發(fā)中，可能都不太會用到，但線程池技術(shù)中會有所使用，由于內(nèi)部沒有使用AQS，而是直接使用CAS，所以代碼理解起來會比較困難，但這并不妨礙我們理解底層的實現(xiàn)模型，在理解了模型的基礎(chǔ)上，有興趣的話再查閱源碼，就會有方向感，看起來也會比較容易，希望本文有所借鑒意義。

到此這篇關(guān)于java并發(fā)編程中的SynchronousQueue實現(xiàn)原理解析的文章就介紹到這了,更多相關(guān)SynchronousQueue實現(xiàn)原理內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論