Java並發編程回環屏障CyclicBarrier實例分析-java教程-PHP中文網

CyclicBarrier

前面介紹的CountDownLatch在解決多個執行緒同步方面相對於呼叫執行緒的join方法已經有了不少最佳化。但是CountDownLatch的計數器是一次性的，也就是等到計數器值變為0後，再呼叫CountDownLatch的await和countdown方法都會立刻返回，這就起不到線程同步的效果了。所以為了滿足計數器可以重置的需要，JDK開發組提供了CyclicBarrier類，而CyclicBarrier類別的功能並不限於CountDownLatch的功能。從字面意思理解 CyclicBarrier 是回環屏障的意思，它可以讓一組線程全部達到一個狀態後再全部同時執行。這裡之所以叫作回環是因為當所有等待執行緒執行完畢，並重置CyclicBarrier 的狀態後它可以被重複使用。之所以叫作屏障是因為線程調用await方法後就會被阻塞，這個阻塞點就稱為屏障點，等所有線程都調用了 await方法後，線程們就會衝破屏障，繼續向下運行。在介紹原理前先介紹幾個實例以便加深理解。在下面的例子中，我們要實現的是，使用兩個執行緒去執行一個被分解的任務 A，當兩個執行緒把自己的任務都執行完畢後再對它們的結果進行匯總處理。

import java.util.concurrent.CyclicBarrier; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; public class CycleBarrierTest { //创建一个线程数固定为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(); } }

登入後複製

Java並發編程回環屏障CyclicBarrier實例分析

如上程式碼建立了一個CyclicBarrier對象，其第一個參數為計數器初始值，第二個數Runable是當計數值為0時需要執行的任務。在main函數裡面先建立了一個大小為2的執行緒池。在執行緒池中加入兩個子任務，每個子任務在完成自身邏輯後會呼叫該方法。一開始計數器值為2，當第一個執行緒呼叫await方法時，計數器值會遞減為1，由於此時計數器值不為0，所以目前執行緒就到了屏障點而被阻塞。然後第二個執行緒呼叫await時，會進入屏障，計數器值也會遞減，現在計數器值為0，這時就會去執行CyclicBarrier構造函數中的任務，執行完畢後退出屏障點，並且喚醒被阻塞的第二個線程。這時候第一個執行緒也會退出屏障點繼續往下運行。

上面的例子說明了多個執行緒之間是相互等待的，假如計數器值為N，那麼隨後呼叫await 方法的N1個執行緒都會因為到達屏障點而被阻塞，當第N個執行緒呼叫await後，計數器值為0了，這時候第N個執行緒才會發出通知喚醒前面的N1個執行緒。也就是當全部執行緒都到達屏障點時才能一塊繼續往下執行。透過使用CountDownLatch，此範例可以獲得類似的輸出結果。下面再舉個例子來說明CyclicBarrier的可重複使用性。

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

import java.util.concurrent.CyclicBarrier; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; public class CycleBarrierTest1 { //创建一个线程数固定为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(); } }

登入後複製

Java並發編程回環屏障CyclicBarrier實例分析