Java并發(fā)編程回環(huán)屏障CyclicBarrier

更新時間：2022年04月15日 17:03:21 作者：派大大大星?

這篇文章主要介紹了Java并發(fā)編程回環(huán)屏障CyclicBarrier，文章繼續(xù)上文所介紹的Java并發(fā)編程同步器CountDownLatch展開主題相關內(nèi)容，需要的小伙伴可以參考一下

CyclicBarrier

前面介紹的CountDownLatch在解決多個線程同步方面相對于調(diào)用線程的join方法已經(jīng)有了不少優(yōu)化。但是CountDownLatch的計數(shù)器是一次性的，也就是等到計數(shù)器值變?yōu)?后，再調(diào)用CountDownLatch的await和countdown方法都會立刻返回，這就起不到線程同步的效果了。所以為了滿足計數(shù)器可以重置的需要，JDK開發(fā)組提供了CyclicBarrier類，并且CyclicBarrier類的功能并不限于CountDownLatch的功能。從字面意思理解 CyclicBarrier 是回環(huán)屏障的意思，它可以讓一組線程全部達到一個狀態(tài)后再全部同時執(zhí)行。這里之所以叫作回環(huán)是因為當所有等待線程執(zhí)行完畢，并重置CyclicBarrier 的狀態(tài)后它可以被重用。之所以叫作屏障是因為線程調(diào)用await方法后就會被阻塞，這個阻塞點就稱為屏障點，等所有線程都調(diào)用了 await方法后，線程們就會沖破屏障，繼續(xù)向下運行。在介紹原理前先介紹幾個實例以便加深理解。在下面的例子中，我們要實現(xiàn)的是，使用兩個線程去執(zhí)行一個被分解的任務 A，當兩個線程把自己的任務都執(zhí)行完畢后再對它們的結果進行匯總處理。

import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CycleBarrierTest {

   
   //創(chuàng)建一個線程數(shù)固定為2的線程池
   private static CyclicBarrier cyclicBarrier = new CyclicBarrier(2, new Runnable() {
       @Override
       public void run() {
           System.out.println(Thread.currentThread() + " task1 merge result");
       }
   });

   public static void main(String[] args) throws InterruptedException{
       ExecutorService executorService = Executors.newFixedThreadPool(2);

       
       //添加線程A到線程池
       executorService.submit(new Runnable() {
           @Override
           public void run() {
               try {
                   System.out.println(Thread.currentThread() + "task1");
                   System.out.println(Thread.currentThread() + "enter in  barrier");
                   cyclicBarrier.await();
                   System.out.println(Thread.currentThread() + "enter out  barrier");
               } catch (Exception e) {
                   e.printStackTrace();
               }
           }
       });

       //添加線程B到線程池
       executorService.submit(new Runnable() {
           @Override
           public void run() {
               try {
                   System.out.println(Thread.currentThread() + "task2");
                   System.out.println(Thread.currentThread() + "enter in  barrier");
                   cyclicBarrier.await();
                   System.out.println(Thread.currentThread() + "enter out  barrier");
               } catch (Exception e) {
                   e.printStackTrace();
               }
           }
       });

       //關閉線程池
       executorService.shutdown();

   }
}

如上代碼創(chuàng)建了一個CyclicBarrier對象，其第一個參數(shù)為計數(shù)器初始值，第二個數(shù)Runable是當計數(shù)值為0時需要執(zhí)行的任務。在main函數(shù)里面首先創(chuàng)建了一個大小為2的線程池。然后添加兩個子任務到線程池，每個子任務在執(zhí)行完自己的邏輯后會調(diào)用方法。一開始計數(shù)器值為2，當?shù)谝粋€線程調(diào)用await方法時，計數(shù)器值會遞減為1，由于此時計數(shù)器值不為0，所以當前線程就到了屏障點而被阻塞。然后第二個線程調(diào)用await時，會進入屏障，計數(shù)器值也會遞減，現(xiàn)在計數(shù)器值為0，這時就會去執(zhí)行 CyclicBarrier構造函數(shù)中的任務，執(zhí)行完畢后退出屏障點，并且喚醒被阻塞的第二個線程。這時候第一個線程也會退出屏障點繼續(xù)向下運行。

上面的例子說明了多個線程之間是相互等待的，假如計數(shù)器值為N，那么隨后調(diào)用 await 方法的N1個線程都會因為到達屏障點而被阻塞，當?shù)贜個線程調(diào)用await后，計數(shù)器值為0了，這時候第N個線程才會發(fā)出通知喚醒前面的N1個線程。也就是當全部線程都到達屏障點時才能一塊繼續(xù)向下執(zhí)行。對于這個例子來說，使用CountDownLatch也可以得到類似的輸出結果。下面再舉個例子來說明CyclicBarrier的可復用性。

假設一個任務由階段1、階段2和階段3組成，每個線程要串行地執(zhí)行階段1、階段2和階段3，當多個線程執(zhí)行該任務時，必須要保證所有線程的階段1全部完成后才能進入階段2執(zhí)行，當所有線程的階段2全部完成后才能進入階段3執(zhí)行。下面使用 CyclicBarrier 來完成這個需求。

import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CycleBarrierTest1 {

    //創(chuàng)建一個線程數(shù)固定為2的線程池
    private static CyclicBarrier cyclicBarrier = new CyclicBarrier(2);

    public static void main(String[] args) throws InterruptedException{
        ExecutorService executorService = Executors.newFixedThreadPool(2);


        //添加線程A到線程池
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    System.out.println(Thread.currentThread() + "step1");
                    cyclicBarrier.await();
                    System.out.println(Thread.currentThread() + "step2");
                    cyclicBarrier.await();
                    System.out.println(Thread.currentThread() + "step3");
                    cyclicBarrier.await();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });

        //添加線程B到線程池
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    System.out.println(Thread.currentThread() + "step1");
                    cyclicBarrier.await();
                    System.out.println(Thread.currentThread() + "step2");
                    cyclicBarrier.await();
                    System.out.println(Thread.currentThread() + "step3");
                    cyclicBarrier.await();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });

        //關閉線程池
        executorService.shutdown();

    }
}