快捷導(dǎo)航

Java CyclicBarrier源碼層分析與應(yīng)用

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

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

前言

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

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

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

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: 條件變量，用于在屏障點(diǎn)上等待。
parties: 表示需要等待的線程數(shù)。
count: 表示當(dāng)前已經(jīng)到達(dá)屏障點(diǎn)的線程數(shù)。
barrierCommand: 在所有線程到達(dá)屏障點(diǎn)之后執(zhí)行的命令，可以為null。

核心方法

await方法

public int await() throws InterruptedException, BrokenBarrierException {
    try {
        lock.lock();
        if (Thread.interrupted())
            throw new InterruptedException();
        int index = --count;
        if (index == 0) { // 如果是最后一個(gè)到達(dá)的線程
            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();
    }
}

上述代碼主要完成以下幾個(gè)任務(wù)：

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

應(yīng)用場(chǎng)景

任務(wù)分解與合并

當(dāng)一個(gè)大任務(wù)可以分解為多個(gè)子任務(wù)，每個(gè)子任務(wù)獨(dú)立執(zhí)行，但在某個(gè)點(diǎn)上需要等待所有子任務(wù)完成后再繼續(xù)執(zhí)行父任務(wù)。CyclicBarrier可以用來(lái)同步這些子任務(wù)的執(zhí)行，確保它們?cè)谔囟ǖ钠琳宵c(diǎn)上等待，然后一起繼續(xù)執(zhí)行。

應(yīng)用示例

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

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

并行計(jì)算

在并行計(jì)算中，當(dāng)多個(gè)計(jì)算節(jié)點(diǎn)完成局部計(jì)算后，需要將它們的結(jié)果合并。CyclicBarrier可以用來(lái)等待所有計(jì)算節(jié)點(diǎn)完成局部計(jì)算，然后執(zhí)行合并操作。

應(yīng)用示例

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ā)中，可能存在多個(gè)線程分別負(fù)責(zé)不同的任務(wù)，比如渲染、物理模擬、AI計(jì)算等。

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

比如我們?cè)诖蚱ヅ溆螒虻臅r(shí)候，十個(gè)人必須全部加載到100%，才可以開局。否則只要有一個(gè)人沒有加載到100%,那這個(gè)游戲就不能開始。先加載完成的玩家必須等待最后一個(gè)玩家加載成功才可以。

應(yīng)用示例

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

輸出結(jié)果

玩家 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ù)加載

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

應(yīng)用示例

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，以實(shí)現(xiàn)更復(fù)雜的多線程控制邏輯。

應(yīng)用示例

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)的特性，可以被重復(fù)使用。一旦所有線程都到達(dá)屏障點(diǎn)，它會(huì)自動(dòng)重置并再次等待下一輪。這使得 CyclicBarrier 更適合用于一組線程多次協(xié)同工作的場(chǎng)景。
CountDownLatch 是一次性的，一旦計(jì)數(shù)到達(dá)零，就無(wú)法重新設(shè)置。如果需要多次等待，就需要?jiǎng)?chuàng)建新的 CountDownLatch 實(shí)例。

計(jì)數(shù)器的變化

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

用途

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