java多線程之並發工具類CountDownLatch,CyclicBarrier和Semaphore

CountDownLatch

CountDownLatch允許一個或多個線程等待其他線程完成操作。

假設一個Excel文件有多個sheet,我們需要去記錄每個sheet有多少行數據,

這時我們就可以使用CountDownLatch實現主線程等待所有sheet線程完成sheet的解析操作後,再繼續執行自己的任務。

public class CountDownLatchTest {
    private static class WorkThread extends Thread {
        private CountDownLatch cdl;
        public WorkThread(String name, CountDownLatch cdl) {
            super(name);
            this.cdl = cdl;
        }
        public void run() {
            System.out.println(this.getName() + "啟動瞭,時間為" + System.currentTimeMillis());
            System.out.println(this.getName() + "我要統計每個sheet的行數");
            try {
                cdl.await();
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(this.getName() + "執行完瞭,時間為" + System.currentTimeMillis());
        }
    }
    private static class sheetThread extends Thread {
        private CountDownLatch cdl;
        public sheetThread(String name, CountDownLatch cdl) {
            super(name);
            this.cdl = cdl;
        }
        public void run() {
            try {
                System.out.println(this.getName() + "啟動瞭,時間為" + System.currentTimeMillis());
                Thread.sleep(1000); //模擬任務執行耗時
                cdl.countDown();
                System.out.println(this.getName() + "執行完瞭,時間為" + System.currentTimeMillis() + " sheet的行數為:" + (int) (Math.random()*100));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    public static void main(String[] args) throws Exception {
        CountDownLatch cdl = new CountDownLatch(2);
        WorkThread wt0 = new WorkThread("WorkThread", cdl );
        wt0.start();
        sheetThread dt0 = new sheetThread("sheetThread1", cdl);
        sheetThread dt1 = new sheetThread("sheetThread2", cdl);
        dt0.start();
        dt1.start();
    }
}

執行結果:

WorkThread啟動瞭,時間為1640054503027
WorkThread我要統計每個sheet的行數
sheetThread1啟動瞭,時間為1640054503028
sheetThread2啟動瞭,時間為1640054503029
sheetThread2執行完瞭,時間為1640054504031 sheet的行數為:6
sheetThread1執行完瞭,時間為1640054504031 sheet的行數為:44
WorkThread執行完瞭,時間為1640054505036

可以看到,首先WorkThread執行await後開始等待,WorkThread在等待sheetThread1和sheetThread2都執行完自己的任務後,WorkThread立刻繼續執行後面的代碼。

CountDownLatch的構造函數接收一個int類型的參數作為計數器,如果你想等待N個點完成,這裡就傳入N。

當我們調用CountDownLatch的countDown方法時,N就會減1,CountDownLatch的await方法會阻塞當前線程,直到N變成零。

由於countDown方法可以用在任何地方,所以這裡說的N個點,可以是N個線程,也可以是1個線程裡的N個執行步驟。

用在多個線程時,隻需要把這個CountDownLatch的引用傳遞到線程裡即可。

我們繼續根據上面的測試案例流程,一步一步的分析CountDownLatch 源碼。

第一步看CountDownLatch的構造方法,傳入一個不能小於0的int類型的參數作為計數器

public CountDownLatch(int count) {
        if (count < 0) throw new IllegalArgumentException("count < 0");
        this.sync = new Sync(count);
    }
/**
     * Synchronization control For CountDownLatch.
     * Uses AQS state to represent count.
     */
    private static final class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 4982264981922014374L;
        Sync(int count) {
            setState(count);
        }
        int getCount() {
            return getState();
        }
        protected int tryAcquireShared(int acquires) {
            return (getState() == 0) ? 1 : -1;
        }
        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;
            }
        }
    }

看它的註釋,說的非常清楚,Sync就是CountDownLatch的同步控制器瞭,而它也是繼承瞭AQS,並且第3行註釋說到使用瞭AQS的state去代表count值。

第二步就是工作線程調用await()方法

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

如果線程中斷,拋出異常,否則開始調用tryAcquireShared(1),其內部類Sync的實現也非常簡單,就是判斷state也就是CountDownLatch的計數是否等於0,

如果等於0,則該方法返回1,第5行的if判斷不成立,否則該方法返回-1,第5行的if判斷成立,繼續執行doAcquireSharedInterruptibly(1)。

/**
     * Acquires in shared interruptible mode.
     * @param arg the acquire argument
     */
    private void doAcquireSharedInterruptibly(int arg)
        throws InterruptedException {
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            for (;;) {
                final Node p = node.predecessor();
                if (p == head) {
                    int r = tryAcquireShared(arg);
                    if (r >= 0) {
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        failed = false;
                        return;
                    }
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    throw new InterruptedException();
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

這個方法其實就是去獲取共享模式下的鎖,獲取失敗就park住。正如我們測試案例中的WorkThread線程應該次數就被park住瞭,那麼它又是何時被喚醒的呢?

下面就到countDown()方法瞭

public void countDown() {
        sync.releaseShared(1);
    }
public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }

tryReleaseShared(1)方法嘗試去釋放共享鎖

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;
            }
        }

在for循環中,先獲取CountDownLatch的計數也就是當前state,如果等於0返回false,否則將state更新為state-1,並返回最新的state是否等於0。

因此在我們的測試案例中,我們需要調用兩次countDown方法,才會將全局的state更新為0,然後繼續執行doReleaseShared()方法。

/**
     * Release action for shared mode -- signals successor and ensures
     * propagation. (Note: For exclusive mode, release just amounts
     * to calling unparkSuccessor of head if it needs signal.)
     */
    private void doReleaseShared() {
        /*
         * Ensure that a release propagates, even if there are other
         * in-progress acquires/releases.  This proceeds in the usual
         * way of trying to unparkSuccessor of head if it needs
         * signal. But if it does not, status is set to PROPAGATE to
         * ensure that upon release, propagation continues.
         * Additionally, we must loop in case a new node is added
         * while we are doing this. Also, unlike other uses of
         * unparkSuccessor, we need to know if CAS to reset status
         * fails, if so rechecking.
         */
        for (;;) {
            Node h = head;
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                if (ws == Node.SIGNAL) {
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue;            // loop to recheck cases
                    unparkSuccessor(h);
                }
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                // loop on failed CAS
            }
            if (h == head)                   // loop if head changed
                break;
        }
    }
/**
     * Wakes up node's successor, if one exists.
     *
     * @param node the node
     */
    private void unparkSuccessor(Node node) {
        /*
         * If status is negative (i.e., possibly needing signal) try
         * to clear in anticipation of signalling.  It is OK if this
         * fails or if status is changed by waiting thread.
         */
        int ws = node.waitStatus;
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);
        /*
         * Thread to unpark is held in successor, which is normally
         * just the next node.  But if cancelled or apparently null,
         * traverse backwards from tail to find the actual
         * non-cancelled successor.
         */
        Node s = node.next;
        if (s == null || s.waitStatus > 0) {
            s = null;
            for (Node t = tail; t != null && t != node; t = t.prev)
                if (t.waitStatus <= 0)
                    s = t;
        }
        if (s != null)
            LockSupport.unpark(s.thread);
    }

LockSupport.unpark(s.thread),喚醒線程的方法被調用後,WorkThread線程就可以繼續執行瞭。

至此我們簡單分析瞭整個測試案例中CountDownLatch的代碼流程。

Semaphore

Semaphore(信號量)是用來控制同時訪問特定資源的線程數量,相當於一個並發控制器,構造的時候傳入可供管理的信號量的數值,這個數值就是用來控制並發數量的,

每個線程執行前先通過acquire方法獲取信號,執行後通過release歸還信號 。每次acquire返回成功後,Semaphore可用的信號量就會減少一個,如果沒有可用的信號,

acquire調用就會阻塞,等待有release調用釋放信號後,acquire才會得到信號並返回。

下面我們看個測試案例

public class SemaphoreTest {
    public static void main(String[] args) {
        final Semaphore semaphore = new Semaphore(5);
        Runnable runnable = () -> {
            try {
                semaphore.acquire();
                System.out.println(Thread.currentThread().getName() + "獲得瞭信號量>>>>>,時間為" + System.currentTimeMillis());
                Thread.sleep(1000);
          System.out.println(Thread.currentThread().getName() + "釋放瞭信號量<<<<<,時間為" + System.currentTimeMillis());
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                semaphore.release();
            }
        };
        Thread[] threads = new Thread[10];
        for (int i = 0; i < threads.length; i++)
            threads[i] = new Thread(runnable);
        for (int i = 0; i < threads.length; i++)
            threads[i].start();
    }
}

執行結果:

Thread-0獲得瞭信號量>>>>>,時間為1640058647604
Thread-1獲得瞭信號量>>>>>,時間為1640058647604
Thread-2獲得瞭信號量>>>>>,時間為1640058647604
Thread-3獲得瞭信號量>>>>>,時間為1640058647605
Thread-4獲得瞭信號量>>>>>,時間為1640058647605
Thread-0釋放瞭信號量<<<<<,時間為1640058648606
Thread-1釋放瞭信號量<<<<<,時間為1640058648606
Thread-5獲得瞭信號量>>>>>,時間為1640058648607
Thread-4釋放瞭信號量<<<<<,時間為1640058648607
Thread-3釋放瞭信號量<<<<<,時間為1640058648607
Thread-7獲得瞭信號量>>>>>,時間為1640058648607
Thread-8獲得瞭信號量>>>>>,時間為1640058648607
Thread-2釋放瞭信號量<<<<<,時間為1640058648606
Thread-6獲得瞭信號量>>>>>,時間為1640058648607
Thread-9獲得瞭信號量>>>>>,時間為1640058648607
Thread-7釋放瞭信號量<<<<<,時間為1640058649607
Thread-6釋放瞭信號量<<<<<,時間為1640058649607
Thread-8釋放瞭信號量<<<<<,時間為1640058649607
Thread-9釋放瞭信號量<<<<<,時間為1640058649608
Thread-5釋放瞭信號量<<<<<,時間為1640058649607

我們使用for循環同時創建10個線程,首先是線程 0 1 2 3 4獲得瞭信號量,再後面的10行打印結果中,線程1到5分別釋放信號量,相同線程間隔也是1000毫秒,然後線程5 6 7 8 9才能繼續獲得信號量,而且保持最大獲取信號量的線程數小於等於5。

看下Semaphore的構造方法

public Semaphore(int permits) {
        sync = new NonfairSync(permits);
    }
public Semaphore(int permits, boolean fair) {
        sync = fair ? new FairSync(permits) : new NonfairSync(permits);
    }

它支持傳入一個int類型的permits,一個佈爾類型的fair,因此Semaphore也有公平模式與非公平模式。

/**
     * Synchronization implementation for semaphore.  Uses AQS state
     * to represent permits. Subclassed into fair and nonfair
     * versions.
     */
    abstract static class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 1192457210091910933L;
        Sync(int permits) {
            setState(permits);
        }
        final int getPermits() {
            return getState();
        }
        final int nonfairTryAcquireShared(int acquires) {
            for (;;) {
                int available = getState();
                int remaining = available - acquires;
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                    return remaining;
            }
        }
        protected final boolean tryReleaseShared(int releases) {
            for (;;) {
                int current = getState();
                int next = current + releases;
                if (next < current) // overflow
                    throw new Error("Maximum permit count exceeded");
                if (compareAndSetState(current, next))
                    return true;
            }
        }
        final void reducePermits(int reductions) {
            for (;;) {
                int current = getState();
                int next = current - reductions;
                if (next > current) // underflow
                    throw new Error("Permit count underflow");
                if (compareAndSetState(current, next))
                    return;
            }
        }
        final int drainPermits() {
            for (;;) {
                int current = getState();
                if (current == 0 || compareAndSetState(current, 0))
                    return current;
            }
        }
    }

第9行代碼可見Semaphore也是通過AQS的state來作為信號量的計數的

第12行 getPermits() 方法獲取當前的可用的信號量,即還有多少線程可以同時獲得信號量

第15行nonfairTryAcquireShared方法嘗試獲取共享鎖,邏輯就是直接將可用信號量減去該方法請求獲取的數量,更新state並返回該值。

第24行tryReleaseShared 方法嘗試釋放共享鎖,邏輯就是直接將可用信號量加上該方法請求釋放的數量,更新state並返回。

再看下Semaphore的公平鎖

/**
     * Fair version
     */
    static final class FairSync extends Sync {
        private static final long serialVersionUID = 2014338818796000944L;
        FairSync(int permits) {
            super(permits);
        }
        protected int tryAcquireShared(int acquires) {
            for (;;) {
                if (hasQueuedPredecessors())
                    return -1;
                int available = getState();
                int remaining = available - acquires;
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                    return remaining;
            }
        }
    }

看嘗試獲取共享鎖的方法中,多瞭個 if (hasQueuedPredecessors) 的判斷,在java多線程6:ReentrantLock,

分析過hasQueuedPredecessors其實就是判斷當前等待隊列中是否存在等待線程,並判斷第一個等待的線程(head.next)是否是當前線程。

CyclicBarrier

CyclicBarrier的字面意思是可循環使用(Cyclic)的屏障(Barrier)。它要做的事情是,讓一組線程到達一個屏障(也可以叫同步點)時被阻塞,直到最後一個線程到達屏障時,屏障才會開門,所有被屏障攔截的線程才會繼續運行。

一組線程同時被喚醒,讓我們想到瞭ReentrantLock的Condition,它的signalAll方法可以喚醒await在同一個condition的所有線程。

下面我們還是從一個簡單的測試案例先瞭解下CyclicBarrier的用法

public class CyclicBarrierTest extends Thread {
    private CyclicBarrier cb;
    private int sleepSecond;
    public CyclicBarrierTest(CyclicBarrier cb, int sleepSecond) {
        this.cb = cb;
        this.sleepSecond = sleepSecond;
    }
    public void run() {
        try {
            System.out.println(this.getName() + "開始, 時間為" + System.currentTimeMillis());
            Thread.sleep(sleepSecond * 1000);
            cb.await();
            System.out.println(this.getName() + "結束, 時間為" + System.currentTimeMillis());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    public static void main(String[] args) {
        Runnable runnable = new Runnable() {
            public void run() {
                System.out.println("CyclicBarrier的barrierAction開始運行, 時間為" + System.currentTimeMillis());
            }
        };
        CyclicBarrier cb = new CyclicBarrier(2, runnable);
        CyclicBarrierTest cbt0 = new CyclicBarrierTest(cb, 3);
        CyclicBarrierTest cbt1 = new CyclicBarrierTest(cb, 6);
        cbt0.start();
        cbt1.start();
    }
}

執行結果:

Thread-1開始, 時間為1640069673534
Thread-0開始, 時間為1640069673534
CyclicBarrier的barrierAction開始運行, 時間為1640069679536
Thread-1結束, 時間為1640069679536
Thread-0結束, 時間為1640069679536

可以看到Thread-0和Thread-1同時運行,而自定義的線程barrierAction是在6000毫秒後開始執行,說明Thread-0在await之後,等待瞭3000毫秒,和Thread-1一起繼續執行的。

看下CyclicBarrier 的一個更高級的構造函數

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

parties就是設定需要多少線程在屏障前等待,隻有調用await方法的線程數達到才能喚醒所有的線程,還有註意因為使用CyclicBarrier的線程都會阻塞在await方法上,所以在線程池中使用CyclicBarrier時要特別小心,如果線程池的線程過少,那麼就會發生死鎖。

Runnable barrierAction用於在線程到達屏障時,優先執行barrierAction,方便處理更復雜的業務場景。

/**
     * Main barrier code, covering the various policies.
     */
    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;
            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();
                }
            }
            // loop until tripped, broken, interrupted, or timed out
            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();
        }
    }

首先是ReentrantLock加鎖,全局的count值-1,然後判斷count是否等於0,如果不等於0,則循環,condition執行await等待,直到觸發、中斷、中斷或超時,如果count值等於0,先執行barrierAction線程,然後condition開始喚醒所有等待的線程。

簡單是使用之後,有人會覺得CyclicBarrierCountDownLatch有點像,其實它們兩者有些細微的差別:

1:CountDownLatch是在多個線程都進行瞭latch.countDown()後才會觸發事件,喚醒await()在latch上的線程,而執行countDown()的線程,是不會阻塞的;

CyclicBarrier是一個柵欄,用於同步所有調用await()方法的線程,線程執行瞭await()方法之後並不會執行之後的代碼,而隻有當執行await()方法的線程數等於指定的parties之後,這些執行瞭await()方法的線程才會同時運行。

2:CountDownLatch不能循環使用,計數器減為0就減為0瞭,不能被重置;CyclicBarrier本是就是支持循環使用parties,而且提供瞭reset()方法,可以重置計數器。

總結

本篇文章就到這裡瞭,希望能夠給你帶來幫助,也希望您能夠多多關註WalkonNet的更多內容!

推薦閱讀: