引言

那么在程序的世界中是如何對(duì)這種協(xié)調(diào)關(guān)系進(jìn)行描述的呢？今天就和大家聊聊Java大神Doug Lea在并發(fā)包中如何通過CountDownLatch和CyclicBarrier實(shí)現(xiàn)任務(wù)協(xié)調(diào)的代碼描述。

CountDownLatch

我相信大家都知道好代碼的一個(gè)重要特性就是代碼中類、變量等的命名可以做到顧名思義，也就是說看到命名就可以大概知道這個(gè)類或者變量表達(dá)了怎樣的業(yè)務(wù)語義。就拿 CountDownLatch 來說，它的命名形象的表示了其能力屬性，Count代表著計(jì)數(shù)，Down代表著計(jì)數(shù)器的遞減操作，而Latch表示計(jì)數(shù)器遞減后的結(jié)果動(dòng)作。CountDownLatch結(jié)合起來的字面意思就是計(jì)數(shù)器遞減后打開門栓，通過后面內(nèi)容的描述，回過頭來看大家肯定會(huì)覺得這個(gè)命名十分之形象。

好了通過它的類的名稱，我們猜測(cè)了它的功能是通過計(jì)數(shù)器的遞減操作來控制線程，那我們?cè)倏纯垂俜矫枋鍪遣皇沁@個(gè)意思。

/**
* A synchronization aid that allows one or more threads to wait until
* a set of operations being performed in other threads completes.
*
* <p>A {@code CountDownLatch} is initialized with a given <em>count</em>.
* The {@link #await await} methods block until the current count reaches
* zero due to invocations of the {@link #countDown} method, after which
* all waiting threads are released and any subsequent invocations of
* {@link #await await} return immediately. This is a one-shot phenomenon
* -- the count cannot be reset. If you need a version that resets the
* count, consider using a {@link CyclicBarrier}.
*...
*/

上面注釋的大致意思就是CountDownLatch是一個(gè)線程同步器，它允許一個(gè)或者多個(gè)線程阻塞等待直到其他線程中業(yè)務(wù)執(zhí)行完成。CountDownLatch可以通過一個(gè)計(jì)數(shù)器進(jìn)行初始化，他可以讓那個(gè)等待的線程被阻塞，直到對(duì)應(yīng)的計(jì)數(shù)器被置為0。當(dāng)計(jì)數(shù)器置為0后，阻塞的線程被釋放。另外它是一個(gè)一次性使用的同步器，計(jì)數(shù)器無法被重置。

通過JDK的官方描述我們可以明確CountDownLatch三個(gè)核心特征：

1、它是一種線程同步器，用以協(xié)調(diào)線程的執(zhí)行觸發(fā)時(shí)機(jī)；

2、它本質(zhì)是一個(gè)計(jì)數(shù)器，是控制線程的號(hào)令槍；

3、它是一次性使用的，用完即失效。

知道了CountDownLatch是一個(gè)什么東東之后，我們?cè)僖黄饋砜聪滤氖褂脠?chǎng)景是什么，我們?cè)谑裁礃拥那闆r下可以使用它幫我們解決一些代碼中的問題。

使用場(chǎng)景

就像上文描述的，CountDownLatch就像是田徑賽場(chǎng)上裁判員發(fā)射的發(fā)令槍，所有參賽的選手準(zhǔn)備就緒后，發(fā)令槍一響，所有運(yùn)動(dòng)員聞聲而動(dòng)。那么在Java多線程場(chǎng)景中，CountDownLatch就是線程協(xié)調(diào)者，它的計(jì)數(shù)器在沒有減為0之前。假設(shè)有這樣一個(gè)業(yè)務(wù)場(chǎng)景，在一個(gè)監(jiān)控告警平臺(tái)中，需要從告警服務(wù)中查詢告警信息以及從工單服務(wù)中查詢工單信息，然后再分析哪些告警沒有轉(zhuǎn)工單。按照老系統(tǒng)的做法，參見如下簡(jiǎn)化后的偽代碼：

List<Alarm> alarmList = alarmService.getAlarm();
List<WorkOrder> workOrderList = workOrderService.getWorkOrder();
List<Alarm> notTransferToWorkOrder = analysis(alarmList, workOrderList);

大家能看出來這段偽代碼有什么需要進(jìn)行優(yōu)化的地方嗎？我們來一起分析一下。這段代碼在數(shù)據(jù)量不大的時(shí)候可能沒什么影響，但是一旦告警以及工單的數(shù)據(jù)量大的時(shí)候，獲取告警信息或者獲取工單信息都可能出現(xiàn)數(shù)據(jù)查詢慢的問題，那就會(huì)導(dǎo)致這個(gè)分析任務(wù)就會(huì)出現(xiàn)性能瓶頸的問題。那么我們應(yīng)該怎么進(jìn)行優(yōu)化呢?從業(yè)務(wù)以及代碼我們可以看的出來，獲取告警信息以及獲取工單信息，實(shí)際上并沒有業(yè)務(wù)上面的耦合性，在上述代碼中他們是順序執(zhí)行的，因此要進(jìn)行性能優(yōu)化，可以考慮將它們進(jìn)行并行執(zhí)行。

那么修改優(yōu)化后的偽代碼如下所示：

Executor executor = Executors.newFixedThreadPool(2);
executor.execute(()-> { alarmList = alarmService.getAlarm(); });
executor.execute(()-> { workOrderList = workOrderService.getWorkOrder(); });
 
List<Alarm> notTransferToWorkOrder = analysis(alarmList, workOrderList);

我們通過使用線程池的方式，在獲取告警信息以及工單信息的時(shí)候并發(fā)執(zhí)行，不再像之前的執(zhí)行完獲取告警信息再執(zhí)行獲取工單信息，這樣效率更高。但是這樣的實(shí)現(xiàn)方式還是存在問題，由于在線的線程中執(zhí)行操作，并不知道其實(shí)際的執(zhí)行結(jié)果，這就不好判斷執(zhí)行數(shù)據(jù)分析的具體時(shí)機(jī)。這個(gè)時(shí)候CountDownLatch就派上用場(chǎng)了，利用它可以實(shí)現(xiàn)線程揀的等待，條件滿足后再放開執(zhí)行后續(xù)的邏輯。這就好比公司組織團(tuán)建，約定好了早上8點(diǎn)半在公司大門集合，那么司機(jī)師傅肯定要等到所有參加團(tuán)建的同時(shí)都到齊后才會(huì)發(fā)車。

使用CountDownLatch之后的偽代碼如下所示：

Executor executor = Executors.newFixedThreadPool(2);
CountDownLatch latch = new CountDownLatch(2);
executor.execute(()-> { alarmList = alarmService.getAlarm();
                      latch.countDown();
                      });
executor.execute(()-> { workOrderList = workOrderService.getWorkOrder(); 
                      latch.countDown();
                      });
latch.await();
List<Alarm> notTransferToWorkOrder = analysis(alarmList, workOrderList);

底層實(shí)現(xiàn)原理

初始化

在使用CountDownLatch之前我們得先進(jìn)行初始化，在初始化的過程中實(shí)際做了兩件事情，一個(gè)是創(chuàng)建了一個(gè)AQS的同步隊(duì)列，另外一個(gè)是將AQS中的state設(shè)置成了count，這個(gè)state是AQS的核心變量（AQS是并發(fā)包的底層實(shí)現(xiàn)基礎(chǔ)，關(guān)于它的分析我們放到下一篇文章中進(jìn)行）。

從代碼中我們可以看的出來實(shí)際創(chuàng)建了Sync內(nèi)部類實(shí)例，而Sync繼承了AQS，同時(shí)重寫了AQS加鎖解鎖的方法，并通過Sync的對(duì)象，調(diào)用AQS的方法，阻塞線程的運(yùn)行。Sync內(nèi)部類的代碼如下所示，其中tryAcquireShared方法重寫了AQS的模板方法，主要用來獲取共享鎖，在CountDownLatch內(nèi)部主要通過判斷獲取到的state的值是否為0來決定到底有沒有獲取到鎖。如果獲取到的state為0，則表示獲取鎖成功，此時(shí)線程不會(huì)阻塞，反之則獲取鎖失敗，線程會(huì)阻塞。

private static final class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 4982264981922014374L;
 
        Sync(int count) {
            setState(count);
        }
 
        int getCount() {
            return getState();
        }
		//嘗試加共享鎖（通過state判斷）
        protected int tryAcquireShared(int acquires) {
            return (getState() == 0) ? 1 : -1;
        }
		//嘗試釋放共享鎖（通過state判斷）
        protected boolean tryReleaseShared(int releases) {
            // Decrement count; signal when transition to zero
            for (;;) {
                int c = getState();
                if (c == 0)
                    return false;
                int nextc = c-1;
                if (compareAndSetState(c, nextc))
                    return nextc == 0;
            }
        }
    }

計(jì)數(shù)器遞減

如上文場(chǎng)景中介紹的代碼，每個(gè)線程在執(zhí)行完成自身業(yè)務(wù)后執(zhí)行countDown操作，表示該線程已經(jīng)準(zhǔn)備完成。同時(shí)檢查count值是否為0。如果為0則需要喚醒所有等待的線程。如下代碼所示，實(shí)際上它調(diào)用的是父類AQS的releaseShared方法。

public void countDown() {
        sync.releaseShared(1);
    }

tryReleaseShared這個(gè)方法實(shí)際是進(jìn)行嘗試釋放鎖的操作，如果此次count遞減為0，然后釋放所有的線程。

public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }

大致的代碼執(zhí)行邏輯可參見下圖：

阻塞線程

await的作用就是將當(dāng)前線程阻塞住，直到count值減為0才會(huì)放開執(zhí)行。它實(shí)際調(diào)用了內(nèi)部類的tryAcquireSharedNanos方法，這個(gè)方法實(shí)際是Sync類的父類AQS中的方法。

public void await() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }

AQS提供了可以響應(yīng)中斷的獲取公平鎖的實(shí)現(xiàn)的方式。tryAcquireShared在上文已經(jīng)進(jìn)行了介紹，該方法的作用是嘗試獲取共享鎖，如果獲取失敗，則線程將會(huì)被加入到AQS的同步隊(duì)列中進(jìn)行等待，也就是所謂的線程阻塞。

public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);
    }

CyclicBarrier

我們還是從CyclicBarrier的字面意思來先進(jìn)行理解，Cyclic是循環(huán)的意思而Barrier則表示柵欄、障礙的意思，字面的意思就是可循環(huán)的柵欄。還是老套路，在進(jìn)行CyclicBarrier之前，我們先來看下JDK是怎么描述的。

/**
* A synchronization aid that allows a set of threads to all wait for
* each other to reach a common barrier point. CyclicBarriers are
* useful in programs involving a fixed sized party of threads that
* must occasionally wait for each other. The barrier is called
* <em>cyclic</em> because it can be re-used after the waiting threads
* are released.
*
* <p>A {@code CyclicBarrier} supports an optional {@link Runnable} command
* that is run once per barrier point, after the last thread in the party
* arrives, but before any threads are released.
* This <em>barrier action</em> is useful
* for updating shared-state before any of the parties continue.
*...
**/

通過JDK的描述，我們可以看得出來，CyclicBarrier也是一個(gè)線程同步協(xié)調(diào)器，用以協(xié)調(diào)一組進(jìn)程的執(zhí)行。當(dāng)指定個(gè)數(shù)的線程到達(dá)柵欄后，可以放開柵欄，結(jié)束線程阻塞狀態(tài)。這么看上去它和CountDownLatch作用差不多了，實(shí)際上還是有區(qū)別的，CyclicBarrier是可循環(huán)使用的，而CountDownLatch卻是一次性的。我們來看下CyclicBarrier的核心屬性。

//柵欄入口的鎖
private final ReentrantLock lock = new ReentrantLock();
//線程等待條件
private final Condition trip = lock.newCondition();
//攔截的線程數(shù)量
private final int parties;
//在下一個(gè)柵欄代數(shù)到來前執(zhí)行的任務(wù)
private final Runnable barrierCommand;
//當(dāng)前的柵欄代數(shù)
private Generation generation = new Generation();

CyclicBarrier 的源碼實(shí)現(xiàn)和 CountDownLatch 大同小異，CountDownLatch 基于 AQS 的共享模式的使用，而 CyclicBarrier 基于 Condition 來實(shí)現(xiàn)的。

CyclicBarrier內(nèi)部維護(hù)了parties和count變量，parties表示每次參與到一個(gè)Generation中需要被攔截的線程數(shù)量，而count是內(nèi)部計(jì)數(shù)器，在初始化的時(shí)候count與parties相等，當(dāng)每次調(diào)用await方法的時(shí)候計(jì)數(shù)器count就會(huì)減1，這和上文中的countDown類似。

使用場(chǎng)景

還是以上文中的業(yè)務(wù)場(chǎng)景為例我們?cè)俜治鲆幌?，上文中我們通過CountDownLatch實(shí)現(xiàn)了查詢告警信息與查詢工單信息的線程協(xié)調(diào)問題，但是新的問題又出現(xiàn)了。因?yàn)楦婢畔⒑凸涡畔⒍际菍?shí)時(shí)在產(chǎn)生的，而使用CountDownLatch的實(shí)現(xiàn)方式只能完成一次的線程協(xié)調(diào)，后續(xù)產(chǎn)生的告警信息以及工單信息如果還有需要查詢到之后再進(jìn)行數(shù)據(jù)分析的話，它就愛莫能助了。也就是說，如果需要進(jìn)行持續(xù)的線程之間的互相等待完成之后再執(zhí)行后續(xù)的業(yè)務(wù)操作的話，這個(gè)時(shí)候就需要使用CyclicBarrier 來實(shí)現(xiàn)我們的需求了。

底層實(shí)現(xiàn)原理

初始化

CyclicBarrier 存在兩種的構(gòu)造函數(shù)，一種是構(gòu)建CyclicBarrier 的時(shí)候指定每次需要進(jìn)行協(xié)調(diào)的線程個(gè)數(shù)以及解除阻塞之后需要進(jìn)行后續(xù)任務(wù)的執(zhí)行，另一種只是設(shè)置需要協(xié)調(diào)的線程個(gè)數(shù)不設(shè)置后續(xù)執(zhí)行的任務(wù)。

public CyclicBarrier(int parties, Runnable barrierAction) {
        if (parties <= 0) throw new IllegalArgumentException();
        this.parties = parties;
        this.count = parties;
        this.barrierCommand = barrierAction;
    }
 
 public CyclicBarrier(int parties) {
        this(parties, null);
    }

阻塞等待

對(duì)于CyclicBarrier 來說，其最核心的等待方法實(shí)現(xiàn)就是dowait方法，具體代碼如下所示：

private int dowait(boolean timed, long nanos)
        throws InterruptedException, BrokenBarrierException,
               TimeoutException {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            final Generation g = generation;
 
            if (g.broken)
                throw new BrokenBarrierException();
 
            if (Thread.interrupted()) {
                breakBarrier();
                throw new InterruptedException();
            }
 
            int index = --count;
            //如果count計(jì)算為0，則需要喚醒所有線程并進(jìn)入到下一階段的線程協(xié)調(diào)期
            if (index == 0) {  // tripped
                boolean ranAction = false;
                try {
                    final Runnable command = barrierCommand;
                    if (command != null)
                        command.run();
                    ranAction = true;
                    nextGeneration();
                    return 0;
                } finally {
                    if (!ranAction)
                        breakBarrier();
                }
            }
 
            //計(jì)數(shù)器不為0，繼續(xù)進(jìn)行循環(huán)
            for (;;) {
                try {
                    if (!timed)
                        trip.await();
                    else if (nanos > 0L)
                        nanos = trip.awaitNanos(nanos);
                } catch (InterruptedException ie) {
                    if (g == generation && ! g.broken) {
                        breakBarrier();
                        throw ie;
                    } else {
                        // We're about to finish waiting even if we had not
                        // been interrupted, so this interrupt is deemed to
                        // "belong" to subsequent execution.
                        Thread.currentThread().interrupt();
                    }
                }
 
                if (g.broken)
                    throw new BrokenBarrierException();
 
                if (g != generation)
                    return index;
 
                if (timed && nanos <= 0L) {
                    breakBarrier();
                    throw new TimeoutException();
                }
            }
        } finally {
            lock.unlock();
        }
    }

我們可以看到在dowait方法中進(jìn)行了count的遞減操作，檢查count的值是否為0，如果在初始化的時(shí)候定義好了要執(zhí)行的任務(wù)，那么在count為0的時(shí)候就進(jìn)行任務(wù)執(zhí)行，任務(wù)執(zhí)行完成之后調(diào)用nextGeneration進(jìn)行下一次的線程協(xié)調(diào)周期，同時(shí)喚醒所有線程并重置計(jì)數(shù)器。

總結(jié)

本文分別從使用場(chǎng)景以及底層實(shí)現(xiàn)的角度分別介紹了線程同步協(xié)調(diào)神器CountDownLatch和CyclicBarrier，雖然它們都可以起到協(xié)調(diào)線程的作用但是實(shí)際上它們還是有區(qū)別的。CountDownLatch比較適合一個(gè)線程與其他多個(gè)線程之間的同步協(xié)調(diào)場(chǎng)景，而CyclicBarrier則適合一組線程之間的互相等待。另外CountDownLatch是一次性產(chǎn)品，而CyclicBarrier的計(jì)數(shù)器是可以重復(fù)使用的，可以進(jìn)行自動(dòng)重置計(jì)數(shù)器。

到此這篇關(guān)于JDK源碼之線程并發(fā)協(xié)調(diào)神器CountDownLatch和CyclicBarrier詳解的文章就介紹到這了,更多相關(guān)Java 線程并發(fā)協(xié)調(diào)內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論