Java中SynchronousQueue的底層實(shí)現(xiàn)原理剖析

更新時間：2022年11月18日 14:46:57 作者：一燈架構(gòu)

BlockingQueue的實(shí)現(xiàn)類中，有一種阻塞隊(duì)列比較特殊，就是SynchronousQueue（同步移交隊(duì)列），隊(duì)列長度為0。本文就來剖析一下SynchronousQueue的底層實(shí)現(xiàn)原理，感興趣的可以了解一下

1. SynchronousQueue用法

先看一個SynchronousQueue的簡單用例：

/**
 * @author 一燈架構(gòu)
 * @apiNote SynchronousQueue示例
 **/
public class SynchronousQueueDemo {
    public static void main(String[] args) throws InterruptedException {
        // 1. 創(chuàng)建SynchronousQueue隊(duì)列
        BlockingQueue<Integer> synchronousQueue = new SynchronousQueue<>();

        // 2. 啟動一個線程，往隊(duì)列中放3個元素
        new Thread(() -> {
            try {
                System.out.println(Thread.currentThread().getName() + " 入隊(duì)列 1");
                synchronousQueue.put(1);
                Thread.sleep(1);
                System.out.println(Thread.currentThread().getName() + " 入隊(duì)列 2");
                synchronousQueue.put(2);
                Thread.sleep(1);
                System.out.println(Thread.currentThread().getName() + " 入隊(duì)列 3");
                synchronousQueue.put(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();

        // 3. 等待1000毫秒
        Thread.sleep(1000L);

        // 4. 再啟動一個線程，從隊(duì)列中取出3個元素
        new Thread(() -> {
            try {
                System.out.println(Thread.currentThread().getName() + " 出隊(duì)列 " + synchronousQueue.take());
                Thread.sleep(1);
                System.out.println(Thread.currentThread().getName() + " 出隊(duì)列 " + synchronousQueue.take());
                Thread.sleep(1);
                System.out.println(Thread.currentThread().getName() + " 出隊(duì)列 " + synchronousQueue.take());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
    }

}

輸出結(jié)果：

Thread-0 入隊(duì)列 1
Thread-1 出隊(duì)列 1
Thread-0 入隊(duì)列 2
Thread-1 出隊(duì)列 2
Thread-0 入隊(duì)列 3
Thread-1 出隊(duì)列 3

從輸出結(jié)果中可以看到，第一個線程Thread-0往隊(duì)列放入一個元素1后，就被阻塞了。直到第二個線程Thread-1從隊(duì)列中取走元素1后，Thread-0才能繼續(xù)放入第二個元素2。

由于SynchronousQueue是BlockingQueue的實(shí)現(xiàn)類，所以也實(shí)現(xiàn)類BlockingQueue中幾組抽象方法：

為了滿足不同的使用場景，BlockingQueue設(shè)計(jì)了很多的放數(shù)據(jù)和取數(shù)據(jù)的方法。

操作	拋出異常	返回特定值	阻塞	阻塞一段時間
放數(shù)據(jù)	add	offer	put	offer(e, time, unit)
取數(shù)據(jù)	remove	poll	take	poll(time, unit)
查看數(shù)據(jù)（不刪除）	element()	peek()	不支持	不支持

這幾組方法的不同之處就是：

當(dāng)隊(duì)列滿了，再往隊(duì)列中放數(shù)據(jù)，add方法拋異常，offer方法返回false，put方法會一直阻塞（直到有其他線程從隊(duì)列中取走數(shù)據(jù)），offer(e, time, unit)方法阻塞指定時間然后返回false。
當(dāng)隊(duì)列是空，再從隊(duì)列中取數(shù)據(jù)，remove方法拋異常，poll方法返回null，take方法會一直阻塞（直到有其他線程往隊(duì)列中放數(shù)據(jù)），poll(time, unit)方法阻塞指定時間然后返回null。
當(dāng)隊(duì)列是空，再去隊(duì)列中查看數(shù)據(jù)（并不刪除數(shù)據(jù)），element方法拋異常，peek方法返回null。

工作中使用最多的就是offer、poll阻塞指定時間的方法。

2. SynchronousQueue應(yīng)用場景

SynchronousQueue的特點(diǎn)：

隊(duì)列長度是0，一個線程往隊(duì)列放數(shù)據(jù)，必須等待另一個線程取走數(shù)據(jù)。同樣，一個線程從隊(duì)列中取數(shù)據(jù)，必須等待另一個線程往隊(duì)列中放數(shù)據(jù)。

這種特殊的實(shí)現(xiàn)邏輯有什么應(yīng)用場景呢？

我的理解就是，如果你希望你的任務(wù)需要被快速處理，就可以使用這種隊(duì)列。

Java線程池中的newCachedThreadPool（帶緩存的線程池）底層就是使用SynchronousQueue實(shí)現(xiàn)的。

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
            60L, TimeUnit.SECONDS,
            new SynchronousQueue<Runnable>());
}

newCachedThreadPool線程池的核心線程數(shù)是0，最大線程數(shù)是Integer的最大值，線程存活時間是60秒。

如果你使用newCachedThreadPool線程池，你提交的任務(wù)會被更快速的處理，因?yàn)槟忝看翁峤蝗蝿?wù)，都會有一個空閑的線程等著處理任務(wù)。如果沒有空閑的線程，也會立即創(chuàng)建一個線程處理你的任務(wù)。

你想想，這處理效率，杠杠滴！

當(dāng)然也有弊端，如果你提交了太多的任務(wù)，導(dǎo)致創(chuàng)建了大量的線程，這些線程都在競爭CPU時間片，等待CPU調(diào)度，處理任務(wù)速度也會變慢，所以在使用過程中也要綜合考慮。

3. SynchronousQueue源碼解析

3.1 SynchronousQueue類屬性

public class SynchronousQueue<E> extends AbstractQueue<E> implements BlockingQueue<E> {

    // 轉(zhuǎn)換器，取數(shù)據(jù)和放數(shù)據(jù)的核心邏輯都在這個類里面
    private transient volatile Transferer<E> transferer;

    // 默認(rèn)的構(gòu)造方法（使用非公平隊(duì)列）
    public SynchronousQueue() {
        this(false);
    }

    // 有參構(gòu)造方法，可以指定是否使用公平隊(duì)列
    public SynchronousQueue(boolean fair) {
        transferer = fair ? new TransferQueue<E>() : new TransferStack<E>();
    }

    // 轉(zhuǎn)換器實(shí)現(xiàn)類
    abstract static class Transferer<E> {
        abstract E transfer(E e, boolean timed, long nanos);
    }

    // 基于棧實(shí)現(xiàn)的非公平隊(duì)列
    static final class TransferStack<E> extends Transferer<E> {
    }

    // 基于隊(duì)列實(shí)現(xiàn)的公平隊(duì)列
    static final class TransferQueue<E> extends Transferer<E> {
    }

}

可以看到SynchronousQueue默認(rèn)的無參構(gòu)造方法，內(nèi)部使用的是基于棧實(shí)現(xiàn)的非公平隊(duì)列，當(dāng)然也可以調(diào)用有參構(gòu)造方法，傳參是true，使用基于隊(duì)列實(shí)現(xiàn)的公平隊(duì)列。

// 使用非公平隊(duì)列（基于棧實(shí)現(xiàn)）
BlockingQueue<Integer> synchronousQueue = new SynchronousQueue<>();
// 使用公平隊(duì)列（基于隊(duì)列實(shí)現(xiàn)）
BlockingQueue<Integer> synchronousQueue = new SynchronousQueue<>(true);

本次就常用的棧實(shí)現(xiàn)來剖析SynchronousQueue的底層實(shí)現(xiàn)原理。

3.2 棧底層結(jié)構(gòu)

棧結(jié)構(gòu)，是非公平的，遵循先進(jìn)后出。

使用個case測試一下：

/**
 * @author 一燈架構(gòu)
 * @apiNote SynchronousQueue示例
 **/
public class SynchronousQueueDemo {
    public static void main(String[] args) throws InterruptedException {
        // 1. 創(chuàng)建SynchronousQueue隊(duì)列
        SynchronousQueue<Integer> synchronousQueue = new SynchronousQueue<>();

        // 2. 啟動一個線程，往隊(duì)列中放1個元素
        new Thread(() -> {
            try {
                System.out.println(Thread.currentThread().getName() + " 入隊(duì)列 0");
                synchronousQueue.put(0);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();

        // 3. 等待1000毫秒
        Thread.sleep(1000L);

        // 4. 啟動一個線程，往隊(duì)列中放1個元素
        new Thread(() -> {
            try {
                System.out.println(Thread.currentThread().getName() + " 入隊(duì)列 1");
                synchronousQueue.put(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();

        // 5. 等待1000毫秒
        Thread.sleep(1000L);

        // 6. 再啟動一個線程，從隊(duì)列中取出1個元素
        new Thread(() -> {
            try {
                System.out.println(Thread.currentThread().getName() + " 出隊(duì)列 " + synchronousQueue.take());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();

        // 7. 等待1000毫秒
        Thread.sleep(1000L);

        // 8. 再啟動一個線程，從隊(duì)列中取出1個元素
        new Thread(() -> {
            try {
                System.out.println(Thread.currentThread().getName() + " 出隊(duì)列 " + synchronousQueue.take());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
    }

}

輸出結(jié)果：

Thread-0 入隊(duì)列 0
Thread-1 入隊(duì)列 1
Thread-2 出隊(duì)列 1
Thread-3 出隊(duì)列 0

從輸出結(jié)果中可以看出，符合棧結(jié)構(gòu)先進(jìn)后出的順序。

3.3 棧節(jié)點(diǎn)源碼

棧中的數(shù)據(jù)都是由一個個的節(jié)點(diǎn)組成的，先看一下節(jié)點(diǎn)類的源碼：

// 節(jié)點(diǎn)
static final class SNode {
    // 節(jié)點(diǎn)值（取數(shù)據(jù)的時候，該字段為null）
    Object item;
    // 存取數(shù)據(jù)的線程
    volatile Thread waiter;
    // 節(jié)點(diǎn)模式
    int mode;
    // 匹配到的節(jié)點(diǎn)
    volatile SNode match;
    // 后繼節(jié)點(diǎn)
    volatile SNode next;
}

item

節(jié)點(diǎn)值，只在存數(shù)據(jù)的時候用。取數(shù)據(jù)的時候，這個值是null。

waiter

存取數(shù)據(jù)的線程，如果沒有對應(yīng)的接收線程，這個線程會被阻塞。

mode

節(jié)點(diǎn)模式，共有3種類型：

類型值	類型描述	類型的作用
0	REQUEST	表示取數(shù)據(jù)
1	DATA	表示存數(shù)據(jù)
2	FULFILLING	表示正在等待執(zhí)行（比如取數(shù)據(jù)的線程，等待其他線程放數(shù)據(jù)）

3.4 put/take流程

放數(shù)據(jù)和取數(shù)據(jù)的邏輯，在底層復(fù)用的是同一個方法，以put/take方法為例，另外兩個放數(shù)據(jù)的方法，add和offer方法底層實(shí)現(xiàn)是一樣的。

先看一下數(shù)據(jù)流轉(zhuǎn)的過程，方便理解源碼。

還是以上面的case為例：

Thread0先往SynchronousQueue隊(duì)列中放入元素0
Thread1再往SynchronousQueue隊(duì)列放入元素1
Thread2從SynchronousQueue隊(duì)列中取出一個元素

第一步：Thread0先往SynchronousQueue隊(duì)列中放入元素0

把本次操作組裝成SNode壓入棧頂，item是元素0，waiter是當(dāng)前線程Thread0，mode是1表示放入數(shù)據(jù)。

第二步：Thread1再往SynchronousQueue隊(duì)列放入元素1

把本次操作組裝成SNode壓入棧頂，item是元素1，waiter是當(dāng)前線程Thread1，mode是1表示放入數(shù)據(jù)，next是SNode0。

第三步：Thread2從SynchronousQueue隊(duì)列中取出一個元素

這次的操作比較復(fù)雜，也是先把本次的操作包裝成SNode壓入棧頂。

item是null（取數(shù)據(jù)的時候，這個字段沒有值），waiter是null（當(dāng)前線程Thread2正在操作，所以不用賦值了），mode是2表示正在操作（即將跟后繼節(jié)點(diǎn)進(jìn)行匹配），next是SNode1。

然后，Thread2開始把棧頂?shù)膬蓚€節(jié)點(diǎn)進(jìn)行匹配，匹配成功后，就把SNode2賦值給SNode1的match屬性，喚醒SNode1中的Thread1線程，然后彈出SNode2節(jié)點(diǎn)和SNode1節(jié)點(diǎn)。

3.5 put/take源碼實(shí)現(xiàn)

先看一下put方法源碼：

// 放數(shù)據(jù)
public void put(E e) throws InterruptedException {
    // 不允許放null元素
    if (e == null)
        throw new NullPointerException();
    // 調(diào)用轉(zhuǎn)換器實(shí)現(xiàn)類，放元素
    if (transferer.transfer(e, false, 0) == null) {
        // 如果放數(shù)據(jù)失敗，就中斷當(dāng)前線程，并拋出異常
        Thread.interrupted();
        throw new InterruptedException();
    }
}

核心邏輯都在transfer方法中，代碼很長，理清邏輯后，也很容易理解。

// 取數(shù)據(jù)和放數(shù)據(jù)操作，共用一個方法
E transfer(E e, boolean timed, long nanos) {
    SNode s = null;
    // e為空，說明是取數(shù)據(jù)，否則是放數(shù)據(jù)
    int mode = (e == null) ? REQUEST : DATA;

    for (; ; ) {
        SNode h = head;
        // 1. 如果棧頂節(jié)點(diǎn)為空，或者棧頂節(jié)點(diǎn)類型跟本次操作相同（都是取數(shù)據(jù)，或者都是放數(shù)據(jù)）
        if (h == null || h.mode == mode) {
            // 2. 判斷節(jié)點(diǎn)是否已經(jīng)超時
            if (timed && nanos <= 0) {
                // 3. 如果棧頂節(jié)點(diǎn)已經(jīng)被取消，就刪除棧頂節(jié)點(diǎn)
                if (h != null && h.isCancelled())
                    casHead(h, h.next);
                else
                    return null;
                // 4. 把本次操作包裝成SNode，壓入棧頂
            } else if (casHead(h, s = snode(s, e, h, mode))) {
                // 5. 掛起當(dāng)前線程，等待被喚醒
                SNode m = awaitFulfill(s, timed, nanos);
                // 6. 如果這個節(jié)點(diǎn)已經(jīng)被取消，就刪除這個節(jié)點(diǎn)
                if (m == s) {
                    clean(s);
                    return null;
                }
                // 7. 把s.next設(shè)置成head
                if ((h = head) != null && h.next == s)
                    casHead(h, s.next);
                return (E) ((mode == REQUEST) ? m.item : s.item);
            }
            // 8. 如果棧頂節(jié)點(diǎn)類型跟本次操作不同，并且不是FULFILLING類型
        } else if (!isFulfilling(h.mode)) {
            // 9. 再次判斷如果棧頂節(jié)點(diǎn)已經(jīng)被取消，就刪除棧頂節(jié)點(diǎn)
            if (h.isCancelled())
                casHead(h, h.next);
                // 10. 把本次操作包裝成SNode（類型是FULFILLING），壓入棧頂
            else if (casHead(h, s = snode(s, e, h, FULFILLING | mode))) {
                // 11. 使用死循環(huán)，直到匹配到對應(yīng)的節(jié)點(diǎn)
                for (; ; ) {
                    // 12. 遍歷下個節(jié)點(diǎn)
                    SNode m = s.next;
                    // 13. 如果節(jié)點(diǎn)是null，表示遍歷到末尾，設(shè)置棧頂節(jié)點(diǎn)是null，結(jié)束。
                    if (m == null) {
                        casHead(s, null);
                        s = null;
                        break;
                    }
                    SNode mn = m.next;
                    // 14. 如果棧頂?shù)暮罄^節(jié)點(diǎn)跟棧頂節(jié)點(diǎn)匹配成功，就刪除這兩個節(jié)點(diǎn)，結(jié)束。
                    if (m.tryMatch(s)) {
                        casHead(s, mn);
                        return (E) ((mode == REQUEST) ? m.item : s.item);
                    } else
                        // 15. 如果沒有匹配成功，就刪除棧頂?shù)暮罄^節(jié)點(diǎn)，繼續(xù)匹配
                        s.casNext(m, mn);
                }
            }
        } else {
            // 16. 如果棧頂節(jié)點(diǎn)類型跟本次操作不同，并且是FULFILLING類型，
            // 就再執(zhí)行一遍上面第11步for循環(huán)中的邏輯（很少概率出現(xiàn)）
            SNode m = h.next;
            if (m == null)
                casHead(h, null);
            else {
                SNode mn = m.next;
                if (m.tryMatch(h))
                    casHead(h, mn);
                else
                    h.casNext(m, mn);
            }
        }
    }
}

transfer方法邏輯也很簡單，就是判斷本次操作類型是否跟棧頂節(jié)點(diǎn)相同，如果相同，就把本次操作壓入棧頂。否則就跟棧頂節(jié)點(diǎn)匹配，喚醒棧頂節(jié)點(diǎn)線程，彈出棧頂節(jié)點(diǎn)。

transfer方法中調(diào)用了awaitFulfill方法，作用是掛起當(dāng)前線程。

// 等待被喚醒
SNode awaitFulfill(SNode s, boolean timed, long nanos) {
    // 1. 計(jì)算超時時間
    final long deadline = timed ? System.nanoTime() + nanos : 0L;
    Thread w = Thread.currentThread();
    // 2. 計(jì)算自旋次數(shù)
    int spins = (shouldSpin(s) ?
            (timed ? maxTimedSpins : maxUntimedSpins) : 0);
    for (;;) {
        if (w.isInterrupted())
            s.tryCancel();
        // 3. 如果已經(jīng)匹配到其他節(jié)點(diǎn)，直接返回
        SNode m = s.match;
        if (m != null)
            return m;
        if (timed) {
            // 4. 超時時間遞減
            nanos = deadline - System.nanoTime();
            if (nanos <= 0L) {
                s.tryCancel();
                continue;
            }
        }
        // 5. 自旋次數(shù)減一
        if (spins > 0)
            spins = shouldSpin(s) ? (spins-1) : 0;
        else if (s.waiter == null)
            s.waiter = w;
        // 6. 開始掛起當(dāng)前線程
        else if (!timed)
            LockSupport.park(this);
        else if (nanos > spinForTimeoutThreshold)
            LockSupport.parkNanos(this, nanos);
    }
}

awaitFulfill方法的邏輯也很簡單，就是掛起當(dāng)前線程。

take方法底層使用的也是transfer方法：

// 取數(shù)據(jù)
public E take() throws InterruptedException {
    // // 調(diào)用轉(zhuǎn)換器實(shí)現(xiàn)類，取數(shù)據(jù)
    E e = transferer.transfer(null, false, 0);
    if (e != null)
        return e;
    // 沒取到，就中斷當(dāng)前線程
    Thread.interrupted();
    throw new InterruptedException();
}

4. 總結(jié)

SynchronousQueue是一種特殊的阻塞隊(duì)列，隊(duì)列長度是0，一個線程往隊(duì)列放數(shù)據(jù)，必須等待另一個線程取走數(shù)據(jù)。同樣，一個線程從隊(duì)列中取數(shù)據(jù)，必須等待另一個線程往隊(duì)列中放數(shù)據(jù)。
SynchronousQueue底層是基于棧和隊(duì)列兩種數(shù)據(jù)結(jié)構(gòu)實(shí)現(xiàn)的。
Java線程池中的newCachedThreadPool（帶緩存的線程池）底層就是使用SynchronousQueue實(shí)現(xiàn)的。
如果希望你的任務(wù)需要被快速處理，可以使用SynchronousQueue隊(duì)列。

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

您可能感興趣的文章: