快捷導(dǎo)航

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

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

這篇文章主要介紹了Java并發(fā)編程回環(huán)屏障CyclicBarrier，文章繼續(xù)上文所介紹的Java并發(fā)編程同步器CountDownLatch展開主題相關(guān)內(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)屏障的意思，它可以讓一組線程全部達(dá)到一個狀態(tài)后再全部同時執(zhí)行。這里之所以叫作回環(huán)是因?yàn)楫?dāng)所有等待線程執(zhí)行完畢，并重置CyclicBarrier 的狀態(tài)后它可以被重用。之所以叫作屏障是因?yàn)榫€程調(diào)用await方法后就會被阻塞，這個阻塞點(diǎn)就稱為屏障點(diǎn)，等所有線程都調(diào)用了 await方法后，線程們就會沖破屏障，繼續(xù)向下運(yùn)行。在介紹原理前先介紹幾個實(shí)例以便加深理解。在下面的例子中，我們要實(shí)現(xiàn)的是，使用兩個線程去執(zhí)行一個被分解的任務(wù) A，當(dāng)兩個線程把自己的任務(wù)都執(zhí)行完畢后再對它們的結(jié)果進(jìn)行匯總處理。

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

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

   }
}

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

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

假設(shè)一個任務(wù)由階段1、階段2和階段3組成，每個線程要串行地執(zhí)行階段1、階段2和階段3，當(dāng)多個線程執(zhí)行該任務(wù)時，必須要保證所有線程的階段1全部完成后才能進(jìn)入階段2執(zhí)行，當(dāng)所有線程的階段2全部完成后才能進(jìn)入階段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();
                }
            }
        });

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

    }
}

如上代碼中，每個子線程在執(zhí)行完階段1后都調(diào)用了await方法，等到所有線程都到達(dá)屏障點(diǎn)后才會一塊往下執(zhí)行，這就保證了所有線程都完成了階段1后才會開始執(zhí)行階段2。

到此這篇關(guān)于Java并發(fā)編程回環(huán)屏障CyclicBarrier的文章就介紹到這了,更多相關(guān)Java 回環(huán)屏障CyclicBarrier內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: