Java多線程 5.柵欄

1.Java多線程-認識Java線程

2.Java多線程-線程安全

3.Java多線程-線程協作

4.Java多線程-線程池

5.Java多線程-柵欄

6.Java多線程-Fork/Join

 

---

5.1 ReadMe

　　此文線程和任務可以理解為一個意思；

　　Java中一般通過CountDownLantch和CyclicBarrier來解決線程（任務）之間依賴的問題，柵欄特指CyclicBarrier類，因為CountDownLatch可以實現類似功能，所以在此放到一塊講解；

　　在任務A依賴任務B的這種場景可以使用Object的wait和notify來實現，但是如果任務A依賴任務B、C、D多個任務的場景，使用Object的wait和notify就難以實現，例如運動會10個人長跑（看作10個長跑任務），公布總排名這個任務就依賴至少9個長跑任務結束，這種場景適合使用CountDownLatch；

　　結合實際開發過程，更多的場景是A、B、C多個任務同時執行，但是A、B、C任務在執行過程中的某一個點相互依賴，例如一個需求分為前段開發和后端開發，前后端開始聯調的時間點就是2個任務相互依賴的點，這種場景適合使用CyclicBarrier；

5.2 CountDownLatch

　　Latch是門閂的意思，要打開門需要打開門上的所有門閂，CountDownLatch可以理解為一個有多個門閂的門，每個門閂需要一個的線程打開；

　　代碼實現舉例：

import java.security.SecureRandom;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * CountDownLatch實例
 */
public class DemoCountDownLatch {

    //參賽人數 / 門閂個數
    private static final int COUNT_RUNNER = 3;

    public static void main(String[] args) throws Exception {

        // 3人參加比賽 / 創建一個有3個門閂的門
        CountDownLatch countDownLatch = new CountDownLatch(COUNT_RUNNER);

        // 創建3人跑道 / 裝上門
        ExecutorService executorService = Executors.newFixedThreadPool(COUNT_RUNNER);

        for (int i = 0; i < 3; i++) {
            executorService.submit(new Runner("runner" + i, countDownLatch));
        }

        // 等待3人跑到終點 / 把3個門閂都鎖上
        countDownLatch.await();

        // 公布3人成績 / 打開門了
        System.out.println("all runner game over.");
    }

    static class Runner implements Runnable {

        private String name;

        private CountDownLatch countDownLatch;

        public Runner(String name, CountDownLatch countDownLatch) {
            this.name = name;
            this.countDownLatch = countDownLatch;
        }

        @Override
        public void run() {
            try {
                SecureRandom secureRandom = SecureRandom.getInstanceStrong();
                int runTime = Math.abs(secureRandom.nextInt()) % 10;
                TimeUnit.SECONDS.sleep(runTime);
                System.out.println(this.name + " game over cost " + runTime + " second.");
            } catch (Exception e) {
                e.printStackTrace();
            }

            // 跑到終點 / 打開門閂
            this.countDownLatch.countDown();
        }
    }
}

　 說明：

1. 一個CountDownLatch對象后只能使用一次，也就是說不能工作同一個CountDownLatch對象來重復控制線程的依賴問題;

2.上面的例子中如果有長跑運動員中途放棄比賽，是否永遠不能公布總的比賽成績？ CountDownLatch的await可以有入參（timeout, TimeUnit）表示最長等待時間;

5.3 CyclicBarrier

 CyclicBarrier是循環柵欄的的意思，循環表示同一個CyclicBarrier可以重復使用（區別於CountDownLatch），Barrier柵欄可以理解為線程相互依賴的那個點（例如前后端聯調時間點），各個線程在那個點相互等待，等所有線程到達點后才繼續執行；

代碼實現舉例：

import java.time.Instant;
import java.util.concurrent.*;

/**
 * 前端開發倩倩和后端開發厚厚開發一個需求
 * 兩人先獨自開發需求，等都開發完再一塊聯調功能
 */
public class DemoCyclicBarrier
{
    public static void main(String[] args) throws InterruptedException
    {

        // 創建柵欄，參數一為相互依賴的任務數；參數二為各任務到達依賴點后先執行的任務，等任務執行結束相互依賴的任務繼續執行
        CyclicBarrier cyclicBarrier = new CyclicBarrier(2, () -> {
            System.out.println("准備開始聯調吧...");
            lastSecond(3L);
        });

        // 創建線程池執行2個任務
        ExecutorService executorService = Executors.newFixedThreadPool(2);
        executorService.execute(new Coder(10L, "后端", cyclicBarrier));
        executorService.execute(new Coder(3L, "前端", cyclicBarrier));
    }

    /**
     * 線程持續second秒
     */
    private static void lastSecond(long second)
    {
        Instant instant = Instant.now();
        while (Instant.now().minusSeconds(second).isBefore(instant))
        {
        }
    }

    static class Coder implements Runnable
    {
        // 完成工作需要的時間
        private long workTime;

        private String name;

        private CyclicBarrier cyclicBarrier;

        public Coder(long workTime, String name, CyclicBarrier cyclicBarrier)
        {
            this.workTime = workTime;
            this.name = name;
            this.cyclicBarrier = cyclicBarrier;
        }

        @Override
        public void run()
        {
            try
            {
                System.out.println(this.name + " are coding...");
                lastSecond(this.workTime);
                System.out.println(this.name + " code end wait debugging..");
                // 完成工作/到達依賴的點/我這邊可以開始聯調了
                this.cyclicBarrier.await();
                System.out.println("we are debugging..");

            }
            catch (InterruptedException e)
            {
                // 當前線程被中斷
                e.printStackTrace();
            }
            catch (BrokenBarrierException e)
            {
                // 1.其他線程中斷；2.其他線程await方法超時；3.cyclicBarrier重置
                e.printStackTrace();
            }
            // catch (TimeoutException e)
            // {
            // //當前線程的await方法超時(await方法設置超時參數)
            // e.printStackTrace();
            // }
        }
    }
}

說明：

1.CyclicBarrier作為循環柵欄，同一個對象可以循環使用；

2.上面例子中前端開發人員很短時間開發結束，通過await()一直在等待后端開發結束，可以通過await(timeout, TimeUnit)來設置最長等待時間；

3. 可以通過CyclicBarrier的getNumberWaiting()查看到達依賴點的任務；

4.CyclicBarrier構造方法的第二個參數指定的任務A，在其他相互依賴的任務到達依賴點后，任務A優先執行，並且是執行結束，其他任務才繼續執行；

5.4 CyclicBarrier&CountDownLatch舉例

import java.time.Instant;
import java.util.concurrent.*;

public class Demo_CyclicBarrier_CountDownLatch
{
    private static final int COUNT_WORKER = 2;

    public static void main(String[] args) throws InterruptedException
    {
        CountDownLatch countDownLatch = new CountDownLatch(COUNT_WORKER);
        CyclicBarrier cyclicBarrier = new CyclicBarrier(COUNT_WORKER, () -> {
            System.out.println("准備開始聯調吧...");
            lastSecond(3L);
        });

        ExecutorService executorService = Executors.newFixedThreadPool(2);
        executorService.execute(new Coder(10L, "后端", countDownLatch, cyclicBarrier));
        executorService.execute(new Coder(3L, "前端", countDownLatch, cyclicBarrier));

        countDownLatch.await();
        System.out.println("開發聯調結束，需求交付...");
    }

    /**
     * 線程持續second秒
     */
    private static void lastSecond(long second)
    {
        Instant instant = Instant.now();
        while (Instant.now().minusSeconds(second).isBefore(instant))
        {
        }
    }

    static class Coder implements Runnable
    {
        // 開發聯調時間
        private long workTime;

        private String name;

        private CountDownLatch countDownLatch;

        private CyclicBarrier cyclicBarrier;

        public Coder(long workTime, String name, CountDownLatch countDownLatch, CyclicBarrier cyclicBarrier)
        {
            this.workTime = workTime;
            this.name = name;
            this.countDownLatch = countDownLatch;
            this.cyclicBarrier = cyclicBarrier;
        }

        @Override
        public void run()
        {
            try
            {
                System.out.println(this.name + " are coding...");
                lastSecond(this.workTime);
                System.out.println(this.name + " code end wait debugging..");

                this.cyclicBarrier.await();

                System.out.println(this.name + " are debugging..");
                lastSecond(this.workTime);

                System.out.println(this.name + " debug end..");
                this.countDownLatch.countDown();
            }
            catch (Exception e)
            {
                e.printStackTrace();
            }
        }
    }
}