Java CyclicBarrier源碼層分析與應用

更新時間：2023年12月30日 11:12:08 作者：The-Venus

這篇文章主要介紹了Java CyclicBarrier的源碼層分析與應用,CyclicBarrier也叫同步屏障,可以讓一組線程達到一個屏障時被阻塞,直到最后一個線程達到屏障,感興趣的的朋友可以參考下

前言

在多線程編程中，同步工具是確保線程之間協(xié)同工作的重要組成部分。

CyclicBarrier（循環(huán)屏障）是Java中的一個強大的同步工具，它允許一組線程在達到某個共同點之前互相等待。

在本文中，我們將深入探討CyclicBarrier的源碼實現(xiàn)以及提供一些示例，以幫助您更好地理解和應用這個有趣的同步工具。

CyclicBarrier源碼解析以及示例

主要成員變量

public class CyclicBarrier {
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition trip = lock.newCondition();
    private final int parties;
    private int count;
    private final Runnable barrierCommand;
}

lock: 用于控制并發(fā)訪問的重入鎖。
trip: 條件變量，用于在屏障點上等待。
parties: 表示需要等待的線程數(shù)。
count: 表示當前已經到達屏障點的線程數(shù)。
barrierCommand: 在所有線程到達屏障點之后執(zhí)行的命令，可以為null。

核心方法

await方法

public int await() throws InterruptedException, BrokenBarrierException {
    try {
        lock.lock();
        if (Thread.interrupted())
            throw new InterruptedException();
        int index = --count;
        if (index == 0) { // 如果是最后一個到達的線程
            boolean ranAction = false;
            try {
                final Runnable command = barrierCommand;
                if (command != null)
                    command.run();
                ranAction = true;
                return 0;
            } finally {
                if (!ranAction)
                    breakBarrier(); // 執(zhí)行失敗，重置屏障狀態(tài)
            }
        }
        while (index > 0) {
            try {
                trip.await();
            } catch (InterruptedException ie) {
                if (index == 1 && !broken)
                    breakBarrier();
                throw ie;
            }
        }
        if (broken)
            throw new BrokenBarrierException();
        return index;
    } finally {
        lock.unlock();
    }
}

上述代碼主要完成以下幾個任務：

減小計數(shù)器，表示有一個線程到達了屏障點。
如果是最后一個到達的線程，執(zhí)行屏障命令（如果有），然后喚醒所有等待的線程。
如果不是最后一個到達的線程，進入等待狀態(tài)，直到被喚醒。
處理中斷異常和屏障破壞異常。

應用場景

任務分解與合并

當一個大任務可以分解為多個子任務，每個子任務獨立執(zhí)行，但在某個點上需要等待所有子任務完成后再繼續(xù)執(zhí)行父任務。CyclicBarrier可以用來同步這些子任務的執(zhí)行，確保它們在特定的屏障點上等待，然后一起繼續(xù)執(zhí)行。

應用示例

假設我們有一個大型的數(shù)據(jù)處理任務，需要將數(shù)據(jù)分解為若干子任務并行處理，然后在所有子任務完成后進行結果的合并。CyclicBarrier 可以用來同步子任務的執(zhí)行，確保在所有子任務都完成后再進行合并操作。

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
public class TaskDecompositionAndMergeExample {
    private static final int NUM_SUBTASKS = 3;
    private static final CyclicBarrier barrier = new CyclicBarrier(NUM_SUBTASKS, () -> {
        System.out.println("All subtasks have been completed. Merging results...");
    });
    public static void main(String[] args) {
        for (int i = 0; i < NUM_SUBTASKS; i++) {
            final int subtaskId = i;
            new Thread(() -> {
                // Perform individual subtask
                System.out.println("Subtask " + subtaskId + " is processing.");
                // Simulate some computation for the subtask
                try {
                    Thread.sleep((long) (Math.random() * 1000));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("Subtask " + subtaskId + " has completed.");
                try {
                    // Wait for all subtasks to complete
                    barrier.await();
                } catch (InterruptedException | BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
}

并行計算

在并行計算中，當多個計算節(jié)點完成局部計算后，需要將它們的結果合并。CyclicBarrier可以用來等待所有計算節(jié)點完成局部計算，然后執(zhí)行合并操作。

應用示例

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
public class ParallelComputingExample {
    private static final int NUM_THREADS = 4;
    private static final CyclicBarrier barrier = new CyclicBarrier(NUM_THREADS, () -> {
        System.out.println("All threads have completed the computation. Merging results...");
    });
    public static void main(String[] args) {
        for (int i = 0; i < NUM_THREADS; i++) {
            final int threadId = i;
            new Thread(() -> {
                // Perform individual computation
                System.out.println("Thread " + threadId + " is performing computation.");
                // Simulate some computation for the thread
                try {
                    Thread.sleep((long) (Math.random() * 1000));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("Thread " + threadId + " has completed computation.");
                try {
                    // Wait for all threads to complete computation
                    barrier.await();
                } catch (InterruptedException | BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
}

游戲開發(fā)

在多線程游戲開發(fā)中，可能存在多個線程分別負責不同的任務，比如渲染、物理模擬、AI計算等。

在每一幀結束時，這些線程需要同步，確保下一幀開始時所有任務都已完成。CyclicBarrier可以在每一幀結束時等待所有任務完成，然后統(tǒng)一開始下一幀的計算。

比如我們在打匹配游戲的時候，十個人必須全部加載到100%，才可以開局。否則只要有一個人沒有加載到100%,那這個游戲就不能開始。先加載完成的玩家必須等待最后一個玩家加載成功才可以。

應用示例

public class CyclicBarrierDemo {
    private static CyclicBarrier cyclicBarrier;
    static class CyclicBarrierThread extends Thread{
        @Override
        public void run() {
            System.out.println("玩家 " + Thread.currentThread().getName() + " 加載100%");
            //等待
            try {
                cyclicBarrier.await();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
    public static void main(String[] args){
        cyclicBarrier = new CyclicBarrier(10, new Runnable() {
            public void run() {
                System.out.println("玩家都加載好了，開始游戲....");
            }
        });
        for(int i = 0 ; i < 10 ; i++){
            new CyclicBarrierThread().start();
        }
    }
}

輸出結果

玩家 Thread-0 加載100%
玩家 Thread-2 加載100%
玩家 Thread-3 加載100%
玩家 Thread-6 加載100%
玩家 Thread-1 加載100%
玩家 Thread-4 加載100%
玩家 Thread-5 加載100%
玩家 Thread-8 加載100%
玩家 Thread-7 加載100%
玩家 Thread-9 加載100%
玩家都加載好了，開始游戲....

數(shù)據(jù)加載

在某些應用中，可能需要同時加載多個數(shù)據(jù)源，但要確保所有數(shù)據(jù)加載完成后再繼續(xù)執(zhí)行。CyclicBarrier可以用來等待所有數(shù)據(jù)加載完成，然后執(zhí)行后續(xù)操作。

應用示例

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
public class DataLoaderExample {
    private static final int NUM_THREADS = 3;
    private static final CyclicBarrier barrier = new CyclicBarrier(NUM_THREADS, () -> {
        System.out.println("All data loading threads have completed. Initiating further processing...");
    });
    public static void main(String[] args) {
        for (int i = 0; i < NUM_THREADS; i++) {
            final int threadId = i;
            new Thread(() -> {
                // Simulate data loading
                System.out.println("Thread " + threadId + " is loading data.");
                // Simulate data loading time
                try {
                    Thread.sleep((long) (Math.random() * 1000));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("Thread " + threadId + " has completed data loading.");
                try {
                    // Wait for all data loading threads to complete
                    barrier.await();
                } catch (InterruptedException | BrokenBarrierException e) {
                    e.printStackTrace();
                }
                // Perform further processing after data loading is complete
                System.out.println("Thread " + threadId + " is performing further processing.");
            }).start();
        }
    }
}

并發(fā)工具的協(xié)同

CyclicBarrier可以與其他并發(fā)工具一起使用，例如 ExecutorService 和 CountDownLatch，以實現(xiàn)更復雜的多線程控制邏輯。

應用示例

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
public class CyclicBarrierExample {
    private static final int NUM_THREADS = 3;
    private static final CyclicBarrier barrier = new CyclicBarrier(NUM_THREADS, () -> {
        System.out.println("All threads have reached the barrier. Let's continue!");
    });
    public static void main(String[] args) {
        for (int i = 0; i < NUM_THREADS; i++) {
            new Thread(() -> {
                try {
                    // Perform individual tasks
                    System.out.println(Thread.currentThread().getName() + " is performing individual tasks.");
                    // Wait for all threads to reach the barrier
                    barrier.await();
                    // Continue with collective tasks after reaching the barrier
                    System.out.println(Thread.currentThread().getName() + " is performing collective tasks.");
                } catch (InterruptedException | BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
}

CyclicBarrier和CountDownLatch的區(qū)別

循環(huán)性

CyclicBarrier 具有循環(huán)的特性，可以被重復使用。一旦所有線程都到達屏障點，它會自動重置并再次等待下一輪。這使得 CyclicBarrier 更適合用于一組線程多次協(xié)同工作的場景。
CountDownLatch 是一次性的，一旦計數(shù)到達零，就無法重新設置。如果需要多次等待，就需要創(chuàng)建新的 CountDownLatch 實例。

計數(shù)器的變化

在 CyclicBarrier 中，計數(shù)器的遞減是由到達屏障點的線程執(zhí)行的，而且在所有線程都到達之前，任何線程都不會繼續(xù)執(zhí)行。
在 CountDownLatch 中，計數(shù)器的遞減是由任意線程執(zhí)行的，而且線程在遞減計數(shù)器后可以繼續(xù)執(zhí)行，不必等待其他線程。

用途

CyclicBarrier 通常用于一組線程并行執(zhí)行任務，然后在某個點上等待彼此，然后再一起繼續(xù)執(zhí)行下一輪任務。例如，任務分解與合并、并行計算等場景。
CountDownLatch 用于等待一組線程完成某個任務后再執(zhí)行其他任務。例如，主線程等待所有工作線程完成工作后再繼續(xù)執(zhí)行。