併發程式設計欄目程式碼 GitHub package 地址： 點選開啟連結

部落格併發程式設計欄目 ： 點選開啟連結

對於某種需求，比如 A,B,C 三個併發的執行緒全部處理完後才執行 D 執行緒。

可以使用 join 控制等待其他執行緒完成。

比如多名參賽選手全部準備好後才起跑，即全部到達一個點後執行後面的方法。

Doug Lea 在併發包裡封裝了更好的處理類

 下面演示CountDownLatch、CyclicBarrier用法

CountDownLatch演示3個執行緒全部完成後執行後面的方法

package com.thread.concurrent_.basis;

import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * CountDownLatch ：
 * 一個同步輔助類，在完成一組正在其他執行緒中執行的操作之前，它允許一個或多個執行緒一直等待。
 * 用給定的計數 初始化 CountDownLatch。
 * 由於呼叫了 countDown()方法，所以在當前計數到達零之前，await方法會一直受阻塞。
 * 之後，會釋放所有等待的執行緒，await的所有後續呼叫都將立即返回。
 * 這種現象只出現一次——計數無法被重置。
 * <p>
 * CountDownLatch 很適合用來將一個任務分為n個獨立的部分，等這些部分都完成後繼續接下來的任務，
 * CountDownLatch 只能出發一次，計數值不能被重置。
 *
 * @author wei.Li by 14-8-22.
 */
public class CountDownLatchTest {

    //同步完成一個事件的執行的執行緒數量
    public static final int SYNCHRONIZED_DONE_THREAD_NUM = 3;

    //宣告同步輔助類
    private static CountDownLatch countDownLatch
            = new CountDownLatch(SYNCHRONIZED_DONE_THREAD_NUM);

    //執行緒池
    public static final ExecutorService EXECUTOR_SERVICE
            = Executors.newFixedThreadPool(SYNCHRONIZED_DONE_THREAD_NUM + 1);

    /**
     * 模擬多個執行緒執行任務
     */
    public static void analogThreads() {
        for (int i = 0; i < SYNCHRONIZED_DONE_THREAD_NUM; i++) {

            EXECUTOR_SERVICE.submit(new Callable<Object>() {
                @Override
                public Object call() throws Exception {

                    Thread.sleep(new Random().nextInt(20000));
                    //執行結束後計數器-1
                    countDownLatch.countDown();

                    System.out.println(this + " -> done ! " +
                            ".the current countDownLatch is " + countDownLatch.getCount());
                    
                    return true;
                }
            });
        }
        EXECUTOR_SERVICE.shutdown();
    }

    public static void main(String[] args) {

        System.out.println("~~~~~~~~ start ~~~~~~~~");

        //模擬多執行緒協作執行
        analogThreads();

        try {
            //等等計數器歸 0 ，即執行緒全部完成
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("~~~~~~~~ done ~~~~~~~~");

    }

}
 

執行結果

~~~~~~~~ start ~~~~~~~~
[email protected] -> done ! .the current countDownLatch is 2
[email protected] -> done ! .the current countDownLatch is 1
[email protected] -> done ! .the current countDownLatch is 0
~~~~~~~~ done ~~~~~~~~

CyclicBarrier演示所有選手準備好後才開始起跑

package com.thread.concurrent_.basis;

import java.io.IOException;
import java.util.Random;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 一個同步輔助類，它允許一組執行緒互相等待，直到到達某個公共屏障點 (common barrier point)。
 * 在涉及一組固定大小的執行緒的程式中，這些執行緒必須不時地互相等待，此時 CyclicBarrier 很有用。
 * 因為該 barrier 在釋放等待執行緒後可以重用，所以稱它為迴圈 的 barrier。
 * <p>
 * 需要所有的子任務都完成時，才執行主任務，這個時候就可以選擇使用CyclicBarrier。
 * <p>
 * 演示所有選手進入場地後，全部準備好後開始起跑！
 *
 * @author wei.Li by 14-8-23.
 */
public class CyclicBarrierTest {

    //參加跑步比賽的人數
    public static final int FOOTRACE_NUM = 5;

    //執行緒池
    private static ExecutorService executor
            = Executors.newFixedThreadPool(FOOTRACE_NUM);

    private static CyclicBarrier barrier
            = new CyclicBarrier(FOOTRACE_NUM);

    // 一個同步輔助類，它允許一組執行緒互相等待，直到到達某個公共屏障點 (common barrier point)
    static class Footrace implements Runnable {

        private CyclicBarrier barrier;

        private String name;

        public Footrace(CyclicBarrier barrier, String name) {
            super();
            this.barrier = barrier;
            this.name = name;
        }

        @Override
        public void run() {
            try {

                System.out.println(name + "  in the location...");
                Thread.sleep(1000 * (new Random()).nextInt(8));//準備中
                System.out.println(name + "  say : I am ready ...");

                // barrier的await方法，在所有參與者都已經在此 barrier 上呼叫 await 方法之前，將一直等待。
                barrier.await();
            } catch (InterruptedException | BrokenBarrierException e) {
                e.printStackTrace();
            }
            System.out.println(name + " go ！");
        }
    }

    public static void main(String[] args)
            throws IOException, InterruptedException {

        for (int i = 0; i < FOOTRACE_NUM; i++) {
            executor.submit(new Thread(
                    new Footrace(barrier, i + "號選手")
            ));
        }
        executor.shutdown();
    }
}

 

執行結果