盤(pán)點(diǎn)并發(fā)編程中幾個(gè)實(shí)用的線程同步技術(shù)

作者：志哥聊技術(shù) 2024-10-18 11:29:15

本文主要圍繞 Java 多線程中常見(jiàn)的并發(fā)工具類進(jìn)行了簡(jiǎn)單的用例介紹，這些工具類都可以實(shí)現(xiàn)線程同步的效果，底層原理實(shí)現(xiàn)主要是基于 AQS 隊(duì)列式同步器來(lái)實(shí)現(xiàn)，關(guān)于 AQS 我們會(huì)在后期的文章中再次介紹。

01、背景介紹

在前幾篇文章中，我們講到了線程池實(shí)現(xiàn)原理、阻塞隊(duì)列技術(shù)等核心組件，其實(shí) JDK 給開(kāi)發(fā)者還提供了比synchronized更加高級(jí)的線程同步組件，比如 CountDownLatch、CyclicBarrier、Semaphore、Exchanger 等并發(fā)工具類。

下面我們一起來(lái)了解一下這些常用的并發(fā)工具類！

02、常用并發(fā)工具類

2.1、CountDownLatch

CountDownLatch是 JDK5 之后加入的一種并發(fā)流程控制工具類，它允許一個(gè)或多個(gè)線程一直等待，直到其他線程運(yùn)行完成后再執(zhí)行。

它的工作原理主要是通過(guò)一個(gè)計(jì)數(shù)器來(lái)實(shí)現(xiàn)，初始化的時(shí)候需要指定線程的數(shù)量；每當(dāng)一個(gè)線程完成了自己的任務(wù)，計(jì)數(shù)器的值就相應(yīng)得減 1；當(dāng)計(jì)數(shù)器到達(dá) 0 時(shí)，表示所有的線程都已經(jīng)執(zhí)行完畢，處于等待的線程就可以恢復(fù)繼續(xù)執(zhí)行任務(wù)。

根據(jù)CountDownLatch的工作原理，它的應(yīng)用場(chǎng)景一般可以劃分為兩種：

場(chǎng)景一：某個(gè)線程需要在其他 n 個(gè)線程執(zhí)行完畢后，再繼續(xù)執(zhí)行

場(chǎng)景二：多個(gè)工作線程等待某個(gè)線程的命令，同時(shí)執(zhí)行同一個(gè)任務(wù)

下面我們先來(lái)看下兩個(gè)簡(jiǎn)單的示例。

示例1：某個(gè)線程等待 n 個(gè)工作線程

比如某項(xiàng)任務(wù)，先采用多線程去執(zhí)行，最后需要在主線程中進(jìn)行匯總處理，這個(gè)時(shí)候CountDownLatch就可以發(fā)揮作用了，具體應(yīng)用如下！

public class CountDownLatchTest {

    public static void main(String[] args) throws InterruptedException {
        // 采用 10 個(gè)工作線程去執(zhí)行任務(wù)
        final int threadCount = 10;
        CountDownLatch countDownLatch = new CountDownLatch(threadCount);
        for (int i = 0; i < threadCount; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    // 執(zhí)行具體任務(wù)
                    System.out.println("thread name:" +  Thread.currentThread().getName() + "，執(zhí)行完畢！");
                    // 計(jì)數(shù)器減 1
                    countDownLatch.countDown();
                }
            }).start();
        }

        // 阻塞等待 10 個(gè)工作線程執(zhí)行完畢
        countDownLatch.await();
        System.out.println("所有任務(wù)線程已執(zhí)行完畢，準(zhǔn)備進(jìn)行結(jié)果匯總");
    }
}

運(yùn)行結(jié)果如下：

thread name:Thread-0，執(zhí)行完畢！
thread name:Thread-2，執(zhí)行完畢！
thread name:Thread-1，執(zhí)行完畢！
thread name:Thread-3，執(zhí)行完畢！
thread name:Thread-4，執(zhí)行完畢！
thread name:Thread-5，執(zhí)行完畢！
thread name:Thread-6，執(zhí)行完畢！
thread name:Thread-7，執(zhí)行完畢！
thread name:Thread-8，執(zhí)行完畢！
thread name:Thread-9，執(zhí)行完畢！
所有任務(wù)線程執(zhí)行完畢，準(zhǔn)備進(jìn)行結(jié)果匯總

示例2：n 個(gè)工作線程等待某個(gè)線程

比如田徑賽跑，10 個(gè)同學(xué)準(zhǔn)備開(kāi)跑，但是需要等工作人員發(fā)出槍聲才允許開(kāi)跑，使用CountDownLatch可以實(shí)現(xiàn)這一功能，具體應(yīng)用如下！

public class CountDownLatchTest {

    public static void main(String[] args) throws InterruptedException {
        // 使用一個(gè)計(jì)數(shù)器
        CountDownLatch countDownLatch = new CountDownLatch(1);
        final int threadCount = 10;
        // 采用 10 個(gè)工作線程去執(zhí)行任務(wù)
        for (int i = 0; i < threadCount; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        // 阻塞等待計(jì)數(shù)器為 0
                        countDownLatch.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    // 發(fā)起某個(gè)服務(wù)請(qǐng)求，省略
                    System.out.println("thread name:" +  Thread.currentThread().getName() + "，開(kāi)始執(zhí)行！");

                }
            }).start();
        }

        Thread.sleep(1000);
        System.out.println("thread name:" +  Thread.currentThread().getName() + " 準(zhǔn)備開(kāi)始！");
        // 將計(jì)數(shù)器減 1，運(yùn)行完成后為 0
        countDownLatch.countDown();
    }
}

運(yùn)行結(jié)果如下：

thread name:main 準(zhǔn)備開(kāi)始！
thread name:Thread-0，開(kāi)始執(zhí)行！
thread name:Thread-1，開(kāi)始執(zhí)行！
thread name:Thread-2，開(kāi)始執(zhí)行！
thread name:Thread-3，開(kāi)始執(zhí)行！
thread name:Thread-5，開(kāi)始執(zhí)行！
thread name:Thread-6，開(kāi)始執(zhí)行！
thread name:Thread-8，開(kāi)始執(zhí)行！
thread name:Thread-7，開(kāi)始執(zhí)行！
thread name:Thread-4，開(kāi)始執(zhí)行！
thread name:Thread-9，開(kāi)始執(zhí)行！

從上面的示例可以很清晰的看到，CountDownLatch類似于一個(gè)倒計(jì)數(shù)器，當(dāng)計(jì)數(shù)器為 0 的時(shí)候，調(diào)用await()方法的線程會(huì)被解除等待狀態(tài)，然后繼續(xù)執(zhí)行。

CountDownLatch類的主要方法，有以下幾個(gè)：

public CountDownLatch(int count)：核心構(gòu)造方法，初始化的時(shí)候需要指定線程數(shù)
countDown()：每調(diào)用一次，計(jì)數(shù)器值 -1，直到 count 被減為 0，表示所有線程全部執(zhí)行完畢
await()：等待計(jì)數(shù)器變?yōu)?0，即等待所有異步線程執(zhí)行完畢，否則一直阻塞
await(long timeout, TimeUnit unit)：支持指定時(shí)間內(nèi)的等待，避免永久阻塞，await()的一個(gè)重載方法

從以上的分析可以得出，當(dāng)計(jì)數(shù)器為 1 的時(shí)候，即由一個(gè)線程來(lái)通知其他線程，效果等同于對(duì)象的wait()和notifyAll()；當(dāng)計(jì)時(shí)器大于 1 的時(shí)候，可以實(shí)現(xiàn)多個(gè)工作線程完成任務(wù)后通知一個(gè)或者多個(gè)等待線程繼續(xù)工作，CountDownLatch可以看成是一種進(jìn)階版的等待/通知機(jī)制，在實(shí)際中應(yīng)用比較多見(jiàn)。

2.2、CyclicBarrier

CyclicBarrier從字面上很容易理解，表示可循環(huán)使用的屏障，它真正的作用是讓一組線程到達(dá)一個(gè)屏障時(shí)被阻塞，直到滿足要求的線程數(shù)都到達(dá)屏障時(shí)，屏障才會(huì)解除，此時(shí)所有被屏障阻塞的線程就可以繼續(xù)執(zhí)行。

下面我們還是先看一個(gè)簡(jiǎn)單的示例，以便于更好的理解這個(gè)工具類。

public class CyclicBarrierTest {

    public static void main(String[] args) {
        // 設(shè)定參與線程的個(gè)數(shù)為 5
        int threadCount = 5;
        CyclicBarrier cyclicBarrier = new CyclicBarrier(threadCount, new Runnable() {
            @Override
            public void run() {
                System.out.println("所有的線程都已經(jīng)準(zhǔn)備就緒...");
            }
        });
        for (int i = 0; i < threadCount; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    System.out.println("thread name:" +  Thread.currentThread().getName() + "，已達(dá)到屏障！");
                    try {
                        cyclicBarrier.await();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                    System.out.println("thread name:" +  Thread.currentThread().getName() + "，阻塞解除，繼續(xù)執(zhí)行！");
                }
            }).start();
        }
    }
}

輸出結(jié)果：

thread name:Thread-0，已達(dá)到屏障！
thread name:Thread-1，已達(dá)到屏障！
thread name:Thread-2，已達(dá)到屏障！
thread name:Thread-3，已達(dá)到屏障！
thread name:Thread-4，已達(dá)到屏障！
所有的線程都已經(jīng)準(zhǔn)備就緒...
thread name:Thread-4，阻塞解除，繼續(xù)執(zhí)行！
thread name:Thread-0，阻塞解除，繼續(xù)執(zhí)行！
thread name:Thread-3，阻塞解除，繼續(xù)執(zhí)行！
thread name:Thread-1，阻塞解除，繼續(xù)執(zhí)行！
thread name:Thread-2，阻塞解除，繼續(xù)執(zhí)行！

從上面的示例可以很清晰的看到，CyclicBarrier中設(shè)定的線程數(shù)相當(dāng)于一個(gè)屏障，當(dāng)所有的線程數(shù)達(dá)到時(shí)，此時(shí)屏障就會(huì)解除，線程繼續(xù)執(zhí)行剩下的邏輯。

CyclicBarrier類的主要方法，有以下幾個(gè)：

public CyclicBarrier(int parties)：構(gòu)造方法，parties參數(shù)表示參與線程的個(gè)數(shù)
public CyclicBarrier(int parties, Runnable barrierAction)：核心構(gòu)造方法，barrierAction參數(shù)表示線程到達(dá)屏障時(shí)的回調(diào)方法
public void await()：核心方法，每個(gè)線程調(diào)用await()方法告訴CyclicBarrier我已經(jīng)到達(dá)了屏障，然后當(dāng)前線程被阻塞，直到屏障解除，繼續(xù)執(zhí)行剩下的邏輯

從以上的示例中，可以看到CyclicBarrier與CountDownLatch有很多的相似之處，都能夠?qū)崿F(xiàn)線程之間的等待，但是它們的側(cè)重點(diǎn)不同：

CountDownLatch一般用于一個(gè)或多個(gè)線程，等待其他的線程執(zhí)行完任務(wù)后再執(zhí)行
CyclicBarrier一般用于一組線程等待至某個(gè)狀態(tài)，當(dāng)狀態(tài)解除之后，這一組線程再繼續(xù)執(zhí)行
CyclicBarrier中的計(jì)數(shù)器可以反復(fù)使用，而CountDownLatch用完之后只能重新初始化

2.3、Semaphore

Semaphore通常我們把它稱之為信號(hào)計(jì)數(shù)器，它可以保證同一時(shí)刻最多有 N 個(gè)線程能訪問(wèn)某個(gè)資源，比如同一時(shí)刻最多允許 10 個(gè)用戶訪問(wèn)某個(gè)服務(wù)，同一時(shí)刻最多創(chuàng)建 100 個(gè)數(shù)據(jù)庫(kù)連接等等。

Semaphore可以用于控制并發(fā)的線程數(shù)，實(shí)際應(yīng)用場(chǎng)景非常的廣，比如流量控制、服務(wù)限流等等。

下面我們看一個(gè)簡(jiǎn)單的示例。

public class SemaphoreTest {

    public static void main(String[] args) {
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

        // 同一時(shí)刻僅允許最多3個(gè)線程獲取許可
        final Semaphore semaphore = new Semaphore(3);
        // 初始化 5 個(gè)線程生成
        for (int i = 0; i < 5; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        // 如果超過(guò)了許可數(shù)量,其他線程將在此等待
                        semaphore.acquire();
                        System.out.println(format.format(new Date()) +  " thread name:" +  Thread.currentThread().getName() + " 獲取許可，開(kāi)始執(zhí)行任務(wù)");
                        // 假設(shè)執(zhí)行某項(xiàng)任務(wù)的耗時(shí)
                        Thread.sleep(2000);
                    } catch (Exception e) {
                        e.printStackTrace();
                    } finally {
                        // 使用完后釋放許可
                        semaphore.release();
                    }
                }
            }).start();
        }
    }
}

輸出結(jié)果：

2023-11-22 17:32:01 thread name:Thread-0 獲取許可，開(kāi)始執(zhí)行任務(wù)
2023-11-22 17:32:01 thread name:Thread-1 獲取許可，開(kāi)始執(zhí)行任務(wù)
2023-11-22 17:32:01 thread name:Thread-2 獲取許可，開(kāi)始執(zhí)行任務(wù)
2023-11-22 17:32:03 thread name:Thread-4 獲取許可，開(kāi)始執(zhí)行任務(wù)
2023-11-22 17:32:03 thread name:Thread-3 獲取許可，開(kāi)始執(zhí)行任務(wù)

從上面的示例可以很清晰的看到，同一時(shí)刻前 3 個(gè)線程獲得了許可優(yōu)先執(zhí)行， 2 秒過(guò)后許可被釋放，剩下的 2 個(gè)線程獲取釋放的許可繼續(xù)執(zhí)行。

Semaphore類的主要方法，有以下幾個(gè)：

public Semaphore(int permits)：構(gòu)造方法，permits參數(shù)表示同一時(shí)間能訪問(wèn)某個(gè)資源的線程數(shù)量
acquire()：獲取一個(gè)許可，在獲取到許可之前或者被其他線程調(diào)用中斷之前，線程將一直處于阻塞狀態(tài)
tryAcquire(long timeout, TimeUnit unit)：表示在指定時(shí)間內(nèi)嘗試獲取一個(gè)許可，如果獲取成功，返回true；反之false
release()：釋放一個(gè)許可，同時(shí)喚醒一個(gè)獲取許可不成功的阻塞線程。

通過(guò)permits參數(shù)的設(shè)定，可以實(shí)現(xiàn)限制多個(gè)線程同時(shí)訪問(wèn)服務(wù)的效果，當(dāng)permits參數(shù)為 1 的時(shí)候，表示同一時(shí)刻只有一個(gè)線程能訪問(wèn)服務(wù)，相當(dāng)于一個(gè)互斥鎖，效果等同于synchronized。

使用Semaphore的時(shí)候，通常需要先調(diào)用acquire()或者tryAcquire()獲取許可，然后通過(guò)try ... finally模塊在finally中釋放許可。

例如如下方式，嘗試在 3 秒內(nèi)獲取許可，如果沒(méi)有獲取就退出，防止程序一直阻塞。

// 嘗試 3 秒內(nèi)獲取許可
if(semaphore.tryAcquire(3, TimeUnit.SECONDS)){
    try {
       // ...業(yè)務(wù)邏輯
    }  finally {
        // 釋放許可
        semaphore.release();
    }
}

2.4、Exchanger

Exchanger從字面上很容易理解表示交換，它主要用途在兩個(gè)線程之間進(jìn)行數(shù)據(jù)交換，注意也只能在兩個(gè)線程之間進(jìn)行數(shù)據(jù)交換。

Exchanger提供了一個(gè)exchange()同步交換方法，當(dāng)兩個(gè)線程調(diào)用exchange()方法時(shí)，無(wú)論調(diào)用時(shí)間先后，會(huì)互相等待線程到達(dá)exchange()方法同步點(diǎn)，此時(shí)兩個(gè)線程進(jìn)行交換數(shù)據(jù)，將本線程產(chǎn)出數(shù)據(jù)傳遞給對(duì)方。

簡(jiǎn)單的示例如下。

public class ExchangerTest {

    public static void main(String[] args) {
        // 交換同步器
        Exchanger<String> exchanger = new Exchanger<>();

        // 線程1
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    String value = "A";
                    System.out.println("thread name:" +  Thread.currentThread().getName() + " 原數(shù)據(jù)：" + value);
                    String newValue = exchanger.exchange(value);
                    System.out.println("thread name:" +  Thread.currentThread().getName() + " 交換后的數(shù)據(jù)：" + newValue);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();

        // 線程2
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    String value = "B";
                    System.out.println("thread name:" +  Thread.currentThread().getName() + " 原數(shù)據(jù)：" + value);
                    String newValue = exchanger.exchange(value);
                    System.out.println("thread name:" +  Thread.currentThread().getName() + " 交換后的數(shù)據(jù)：" + newValue);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }
}

輸出結(jié)果：

thread name:Thread-0 原數(shù)據(jù)：A
thread name:Thread-1 原數(shù)據(jù)：B
thread name:Thread-0 交換后的數(shù)據(jù)：B
thread name:Thread-1 交換后的數(shù)據(jù)：A

從上面的示例可以很清晰的看到，當(dāng)線程Thread-0和Thread-1都到達(dá)了exchange()方法的同步點(diǎn)時(shí)，進(jìn)行了數(shù)據(jù)交換。

Exchanger類的主要方法，有以下幾個(gè)：

exchange(V x)：等待另一個(gè)線程到達(dá)此交換點(diǎn)，然后將給定的對(duì)象傳送給該線程，并接收該線程的對(duì)象，除非當(dāng)前線程被中斷，否則一直阻塞等待
exchange(V x, long timeout, TimeUnit unit)：表示在指定的時(shí)間內(nèi)等待另一個(gè)線程到達(dá)此交換點(diǎn)，如果超時(shí)會(huì)自動(dòng)退出并拋超時(shí)異常

如果多個(gè)線程調(diào)用exchange()方法，數(shù)據(jù)交換可能會(huì)出現(xiàn)混亂，因此實(shí)際上Exchanger應(yīng)用并不多見(jiàn)。