快捷導(dǎo)航

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

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

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

CyclicBarrier

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

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

public class CycleBarrierTest {

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

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

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

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

public class CycleBarrierTest1 {

    //創(chuàng)建一個(gè)線程數(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();

    }
}

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

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

您可能感興趣的文章: