萬字圖解 Java 并發(fā)框架：Fork/Join、CountDownLatch、Semaphore、CyclicBarrier
Fork/Join 框架
Chaya：碼哥，什么是 Fork/Join 框架？
Fork/Join 是 Java 7 引入的并行計算框架，核心思想是 **"分而治之"**。它通過以下特性解決復雜計算問題：
- 自動任務拆分：將大任務遞歸拆分為子任務
- 工作竊取算法（Work-Stealing）：最大化線程利用率
- 輕量級線程管理：基于
ForkJoinPool的優(yōu)化線程池
Fork 就是把一個大任務切分為若干子任務并行的執(zhí)行，Join 就是合并這些子任務的執(zhí)行結(jié)果，最后得到這個大任務的結(jié) 果。
比如計算 1+2+…+10000，可以分割成 10 個子任務，每個子任務分別對 1000 個數(shù)進行求和， 最終匯總這 10 個子任務的結(jié)果。
Fork/Join 的運行流程圖如下：
圖片
工作竊取算法
Chaya：“碼哥。有任務要拆分，那必然會出現(xiàn)分配不均勻的情況？要如何實現(xiàn)負載均衡呢？”
這個問題問得好，Chaya 小姐姐。
我們設計一個工作竊取算法（Work-Stealing）來解決這個問題。每個工作線程維護一個雙端隊列（Deque）：
- 頭部：執(zhí)行自己拆分出的任務（LIFO）
- 尾部：竊取其他線程的任務（FIFO）
圖片
工作竊取算法的優(yōu)點：充分利用線程進行并行計算，減少了線程間的競爭。
工作竊取算法的缺點：在某些情況下還是存在競爭，比如雙端隊列里只有一個任務時。并 且該算法會消耗了更多的系統(tǒng)資源，比如創(chuàng)建多個線程和多個雙端隊列。
任務拆分流程
圖片
使用場景
場景 1：大規(guī)模數(shù)據(jù)處理（并行排序）
需求：對 10 億條數(shù)據(jù)排序，要求內(nèi)存可控且充分利用多核性能。
代碼實現(xiàn)：
public class ParallelMergeSort extends RecursiveAction {
privatefinalint[] array;
privatefinalint start;
privatefinalint end;
privatestaticfinalint THRESHOLD = 1_000_000; // 拆分閾值
@Override
protected void compute() {
if (end - start <= THRESHOLD) {
Arrays.sort(array, start, end); // 小任務直接排序
return;
}
int mid = (start + end) >>> 1;
invokeAll(
new ParallelMergeSort(array, start, mid),
new ParallelMergeSort(array, mid, end)
);
merge(array, start, mid, end); // 合并結(jié)果
}
// 生產(chǎn)級優(yōu)化：復用臨時數(shù)組減少內(nèi)存分配
private void merge(int[] array, int start, int mid, int end) {
int[] temp = ThreadLocalRandom.current().ints().toArray();
// ... 合并邏輯 ...
}
}
// 使用方式
ForkJoinPool pool = new ForkJoinPool(Runtime.getRuntime().availableProcessors());
int[] data = loadHugeData();
pool.invoke(new ParallelMergeSort(data, 0, data.length));性能優(yōu)化點：
- 合理設置
THRESHOLD（通過壓測確定最佳值） - 避免在遞歸中頻繁創(chuàng)建臨時數(shù)組
- 使用
ThreadLocalRandom保證線程安全
場景 2：金融計算（蒙特卡洛模擬）
需求：快速計算期權定價，要求高精度且低延遲。
代碼實現(xiàn)：
public class MonteCarloTask extends RecursiveTask<Double> {
privatefinalint iterations;
privatestaticfinalint THRESHOLD = 10_000;
@Override
protected Double compute() {
if (iterations <= THRESHOLD) {
return calculateSync(); // 同步計算
}
MonteCarloTask left = new MonteCarloTask(iterations / 2);
MonteCarloTask right = new MonteCarloTask(iterations / 2);
left.fork();
double rightResult = right.compute();
double leftResult = left.join(); // 注意順序：先計算再join
return (leftResult + rightResult) / 2;
}
private double calculateSync() {
double sum = 0;
for (int i = 0; i < iterations; i++) {
sum += randomSimulation();
}
return sum / iterations;
}
}
// 生產(chǎn)級調(diào)用（指定超時）
ForkJoinPool pool = new ForkJoinPool(4);
MonteCarloTask task = new MonteCarloTask(1_000_000);
pool.submit(task);
try {
double result = task.get(5, TimeUnit.SECONDS); // 嚴格超時控制
} catch (TimeoutException e) {
task.cancel(true);
// 降級策略...
}ForkJoinPool 生產(chǎn)級配置
自定義線程工廠
public class NamedForkJoinThreadFactory implements ForkJoinPool.ForkJoinWorkerThreadFactory {
privatefinal String namePrefix;
privatefinal AtomicInteger counter = new AtomicInteger(1);
@Override
public ForkJoinWorkerThread newThread(ForkJoinPool pool) {
ForkJoinWorkerThread thread = new ForkJoinWorkerThread(pool) {};
thread.setName(namePrefix + "-" + counter.getAndIncrement());
thread.setPriority(Thread.NORM_PRIORITY);
thread.setDaemon(false); // 生產(chǎn)環(huán)境必須為非守護線程
return thread;
}
}與其他并發(fā)框架對比
圖片
Fork/Join 適用場景：
- 遞歸可分治的問題（排序、遍歷、數(shù)學計算）
- 嚴格低延遲要求的計算任務
- 需要自動負載均衡的大規(guī)模數(shù)據(jù)處理
CountDownLatch
CountDownLatch 是一個同步工具類，它允許一個或多個線程一直等待，直到其他線程執(zhí)行完后再執(zhí)行。
例如，應用程序的主線程希望在負責啟動框架服務的線程已經(jīng)啟動所有框架服務之后執(zhí)行。
假如有這樣一個需求：處理 10 萬條數(shù)據(jù)，分片并行處理，全部完成后觸發(fā)匯總操作。
public class BatchProcessor {
privatestaticfinalint BATCH_SIZE = 1000;
privatefinal ExecutorService executor = Executors.newFixedThreadPool(8);
public void process(List<Data> allData) {
int total = allData.size();
CountDownLatch latch = new CountDownLatch(total / BATCH_SIZE);
for (int i = 0; i < total; i += BATCH_SIZE) {
List<Data> batch = allData.subList(i, Math.min(i+BATCH_SIZE, total));
executor.submit(() -> {
try {
processBatch(batch);
} finally {
latch.countDown(); // 確保計數(shù)減少
}
});
}
try {
if (!latch.await(5, TimeUnit.MINUTES)) {
thrownew TimeoutException("Batch processing timeout");
}
generateSummaryReport();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
shutdownNow();
}
}
private void processBatch(List<Data> batch) { /* ... */ }
}
CountDownLatch 的構造函數(shù)接收一個 int 類型的參數(shù)作為計數(shù)器，如果你想等待 N 個點完 成，這里就傳入 N。
當我們調(diào)用 CountDownLatch 的 countDown 方法時，N 就會減 1，CountDownLatch 的 await 方法 會阻塞當前線程，直到 N 變成零。
用在多個線程時，只需要把這個 CountDownLatch 的引用傳遞到線程里即可。
如果有某個線程處理得比較慢，我們不可能讓主線程一直等待，所以可以使 用另外一個帶指定時間的 await 方法——await（long time，TimeUnit unit），這個方法等待特定時 間后，就會不再阻塞當前線程。
實現(xiàn)原理
CountDownLatch 的核心實現(xiàn)原理是基于 AQS，AQS 全稱 AbstractQueuedSynchronizer，是 java.util.concurrent 中提供的一種高效且可擴展的同步機制；
它是一種提供了原子式管理同步狀態(tài)、阻塞和喚醒線程功能以及隊列模型的簡單框架。
除了 CountDownLatch 工具類，JDK 當中的 Semaphore、ReentrantLock 等工具類都是基于 AQS 來實現(xiàn)的。下面我們用 CountDownLatch 來分析一下 AQS 的實現(xiàn)。
圖片
CountDownLatch 的源碼實現(xiàn)，發(fā)現(xiàn)其實它的代碼實現(xiàn)非常簡單，算上注釋也才 300+ 行代碼，如果去掉注釋的話代碼不到 100 行，大部分方法實現(xiàn)都是調(diào)用的 Sync 這個靜態(tài)內(nèi)部類的實現(xiàn)，而 Sync 就是繼承自 AbstractQueuedSynchronizer。
CountDownLatch 的 UML 類圖如下：
圖片
核心代碼如下。
private staticfinalclass Sync extends AbstractQueuedSynchronizer {
Sync(int count) { setState(count); } // 初始化計數(shù)器
// 嘗試獲取共享鎖：當 state=0 時返回 1（成功），否則返回 -1（失敗）
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
// 嘗試釋放共享鎖：CAS 遞減 state，直到變?yōu)?0
protected boolean tryReleaseShared(int releases) {
for (;;) { // 自旋保證原子性
int c = getState();
if (c == 0) returnfalse; // 已經(jīng)釋放完畢
int nextc = c - 1;
if (compareAndSetState(c, nextc)) // CAS 更新
return nextc == 0; // 返回是否觸發(fā)喚醒
}
}
}Sync 重寫了 AQS 中的 tryAcquireShared 和 tryReleaseShared 兩個方法。
當調(diào)用 CountDownLatch 的 awit() 方法時，會調(diào)用內(nèi)部類 Sync 的 acquireSharedInterruptibly() 方法，在這個方法中會調(diào)用 tryAcquireShared 方法，這個方法就是 Sync 重寫的 AQS 中的方法；
調(diào)用 countDown() 方法原理基本類似。
await() 方法實現(xiàn)
在調(diào)用 await() 方法時，會直接調(diào)用 AQS 類的 acquireSharedInterruptibly 方法，在 acquireSharedInterruptibly 方法內(nèi)部會繼續(xù)調(diào)用 Sync 實現(xiàn)類中的 tryAcquireShared 方法，在 tryAcquireShared 方法中判斷 state 變量值是否為 0。
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1); // 進入 AQS 核心邏輯
}
// AQS 中的實現(xiàn)
public final void acquireSharedInterruptibly(int arg) {
if (Thread.interrupted()) throw new InterruptedException();
if (tryAcquireShared(arg) < 0) // 檢查 state 是否為 0
doAcquireSharedInterruptibly(arg); // 進入阻塞隊列
}doAcquireSharedInterruptibly 關鍵步驟。
圖片
關鍵點：
- 節(jié)點入隊：通過
addWaiter方法將線程封裝為SHARED模式節(jié)點加入隊列尾部 - 自旋檢查：循環(huán)判斷前驅(qū)節(jié)點是否是頭節(jié)點（公平性保證）
- 阻塞控制：調(diào)用
LockSupport.park()掛起線程，響應中斷。
countDown() 方法
當執(zhí)行 CountDownLatch 的 countDown() 方法，將計數(shù)器減一，也就是將 state 值減一，當減到 0 的時候，等待隊列中的線程被釋放。是調(diào)用 AQS 的 releaseShared() 方法來實現(xiàn)的。
public void countDown() {
sync.releaseShared(1); // 觸發(fā)釋放操作
}
// AQS 中的實現(xiàn)
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) { // CAS 遞減 state
doReleaseShared(); // 喚醒后續(xù)節(jié)點
return true;
}
return false;
}CyclicBarrier
CyclicBarrier 的字面意思是可循環(huán)使用（Cyclic）的屏障（Barrier）。
它要做的事情是，讓一 組線程到達一個屏障（也可以叫同步點）時被阻塞，直到最后一個線程到達屏障時，屏障才會 開門，所有被屏障攔截的線程才會繼續(xù)運行。
現(xiàn)實生活中我們經(jīng)常會遇到這樣的情景，在進行某個活動前需要等待人全部都齊了才開始。
例如吃飯時要等全家人都上座了才動筷子，旅游時要等全部人都到齊了才出發(fā)，比賽時要等運動員都上場后才開始。
CyclicBarrier 和 CountDownLatch 是不是很像，只是 CyclicBarrier 可以有不止一個柵欄，因為它的柵欄（Barrier）可以重復使用（Cyclic）。
圖片
CyclicBarrier 是 Java 并發(fā)包中的可重用同步屏障，其特性包括：
- 多階段協(xié)同：支持多次
await()的同步點 - 柵欄動作（Barrier Action）：當所有線程抵達屏障時觸發(fā)
- 自動重置：每次所有線程通過屏障后自動復位
- 中斷處理：可響應線程中斷并傳播異常
如何使用
構造方法
public CyclicBarrier(int parties)
public CyclicBarrier(int parties, Runnable barrierAction)解析：
- parties：傳入一個計數(shù)器值，用來配置可以阻塞多少個線程的。
- 第二個構造方法有一個 Runnable 參數(shù)，這個對象可以在計數(shù)器值減到 0 后，發(fā)起一次調(diào)用。
例如：下面代碼就會在計數(shù)器減到 0 后，打印出"回環(huán)屏障退出"。
CyclicBarrier cyclicBarrier = new CyclicBarrier(5, () -> System.out.println("回環(huán)屏障退出"));場景 ：多階段分布式計算匯總
需求：實時計算商品庫存分布（本地倉 + 區(qū)域倉 + 全國倉），需三階段統(tǒng)計數(shù)據(jù)匯總。
public class InventoryComputeService {
// 構建 3 線程等待 CyclicBarrier
privatefinalint PARTIES = 3;
privatefinal CyclicBarrier barrier = new CyclicBarrier(PARTIES, this::mergeData);
// 保存最后的結(jié)果
privatevolatile Map<String, Integer> result = new ConcurrentHashMap<>();
public void compute() {
// 異步執(zhí)行 3 個任務，執(zhí)行完成調(diào)用 barrier.await();，當所有任務完成后會執(zhí)行 mergeData
List<CompletableFuture<Void>> tasks = new ArrayList<>();
tasks.add(computeLocalStock());
tasks.add(computeRegionalStock());
tasks.add(computeNationalStock());
// 本次3 個任務任何一個計算出現(xiàn)異常的話，重置 barrier
CompletableFuture.allOf(tasks.toArray(new CompletableFuture[0]))
.exceptionally(ex -> {
barrier.reset(); // 異常處理
returnnull;
});
}
private CompletableFuture<Void> computeLocalStock() {
// 異步線程
return CompletableFuture.runAsync(() -> {
try {
// 模擬計算耗時
result.put("local", calculate(Region.LOCAL));
// 執(zhí)行完成，調(diào)用 await
barrier.await(5, TimeUnit.SECONDS); // 超時控制
} catch (Exception e) {
handleException(e);
}
});
}
// computeRegionalStock/computeNationalStock 同理...
private void mergeData() {
lock.lock();
try {
System.out.println("各區(qū)域最終庫存合并結(jié)果: " + result);
} finally {
lock.unlock();
result.clear(); // 清空狀態(tài)為下次計算準備
}
}
}代碼核心解釋
構造方法創(chuàng)建等待三個線程執(zhí)行完成的 CyclicBarrier，CyclicBarrier 與 CountDownLatch 最大的區(qū)別是 CountDownLatch 一次性的，CyclicBarrier 是可循環(huán)利用的。
private final CyclicBarrier barrier = new CyclicBarrier(PARTIES, this::mergeData);當三個線程都執(zhí)行完成，會調(diào)用 mergeData 方法統(tǒng)計結(jié)果。
圖片
實現(xiàn)原理
核心數(shù)據(jù)結(jié)構
public class CyclicBarrier {
privatefinal ReentrantLock lock = new ReentrantLock();
privatefinal Condition trip = lock.newCondition();
privatefinalint parties; // 需要同步的線程數(shù)
privatefinal Runnable barrierCommand; // 柵欄動作
private Generation generation = new Generation(); // 當前代
privatestaticclass Generation {
boolean broken = false; // 柵欄是否破裂
}
// 掛起線程數(shù)計數(shù)器（每次循環(huán)遞減）
privateint count;
}CyclicBarrier 狀態(tài)流轉(zhuǎn)
圖片
await 實現(xiàn)
public int await() throws InterruptedException, BrokenBarrierException {
try {
return dowait(false, 0L);
} catch (TimeoutException toe) {
thrownew Error(toe); // cannot happen
}
}
private int dowait(boolean timed, long nanos) throws InterruptedException, BrokenBarrierException, TimeoutException {
final ReentrantLock lock = this.lock;
lock.lock();
try {
// 回環(huán)屏障使用完畢或重置后，都會生成一個新的generation，這個對象可以用來讓線程退出回環(huán)屏障
final Generation g = generation;
// 每個進入的線程，都使計數(shù)器減1，當計數(shù)器歸零后進入下面的if判斷
int index = --count;
if (index == 0) { // tripped
boolean ranAction = false;
try {
// 如果實例化時傳入了Runnable對象，則在這里調(diào)用它的run()方法
final Runnable command = barrierCommand;
if (command != null)
command.run();
ranAction = true;
// 里面做了喚醒所有等待線程的操作，線程是在下面的自旋中掛起的
nextGeneration();
return0;
} finally {
if (!ranAction)
breakBarrier();
}
}
for (;;) {
// 此處省略的線程被interrupt的try catch
// 根據(jù)是否傳入等待時間來判斷調(diào)用哪一個方法
if (!timed) {
// condition的await()方法，這里會暫時釋放鎖
trip.await();
} elseif (nanos > 0L) {
nanos = trip.awaitNanos(nanos);
}
// 計數(shù)器歸零后，線程退出自旋
if (g != generation) {
return index;
}
}
} finally {
lock.unlock();
}
}上面代碼中的trip就是一個 Condition 對象，是CyclicBarrier的一個成員變量?？偨Y(jié)一下 doWait()方法，其實做的事情還是比較簡單的。
線程進入 doWait()， 先搶占到鎖 lock 鎖對象，并執(zhí)行計數(shù)器遞減 1 的操作。 遞減后的計數(shù)器值不為 0，則將自己掛起在 Condition 隊列中。
遞減后的計數(shù)器值為 0，則調(diào)用 signalAll()喚醒所有在條件隊列中的線程，并創(chuàng)建新的 generation 對象，讓線程可以退出回環(huán)屏障。
核心方法流程圖如下。
圖片
Semaphore
Semaphore，它是一個信號量，主要作用是用來控制并發(fā)中同一個時刻執(zhí)行的線程數(shù)量，可以用來做限流器，或者流程控制器。
在創(chuàng)建的時候會指定好它有多少個信號量，比如 Semaphre semaphore = new Semaphore(2)，就只有 2 個信號量。
核心功能是控制同時訪問特定資源的線程數(shù)量，具有以下特性：
- 許可管理：通過
acquire()/release()操作許可數(shù)量 - 公平性選擇：支持公平/非公平兩種模式
- 可中斷：支持帶超時的許可獲取
- 動態(tài)調(diào)整：運行時修改許可數(shù)量
這個信號量可以比作是車道，每一個時刻每條車道只能允許一輛汽車通過，你可以理解為高速收費站上的收費口，每個收費口任意一時刻只能允許一輛汽車通行。
畫個圖來講解一下：
圖片
如何使用
接口限流（突發(fā)流量控制）。
public class ApiRateLimiter {
// 生產(chǎn)級配置：許可數(shù) = QPS閾值 * 響應時間(秒)
privatestaticfinal Semaphore SEMAPHORE = new Semaphore(500);
privatestaticfinal Timer METRIC_TIMER = new Timer(true);
static {
// 監(jiān)控線程：每10秒打印許可使用率
METRIC_TIMER.schedule(new TimerTask() {
public void run() {
double usage = (SEMAPHORE.availablePermits() / 500.0) * 100;
log.info("API許可使用率: {0}%", 100 - usage);
}
}, 10_000, 10_000);
}
public Response handleRequest(Request request) {
if (!SEMAPHORE.tryAcquire(50, TimeUnit.MILLISECONDS)) { // 非阻塞獲取
thrownew BizException(429, "請求過于頻繁");
}
try {
return doBusinessLogic(request); // 核心業(yè)務邏輯
} finally {
SEMAPHORE.release(); // 確保釋放許可
}
}
}生產(chǎn)級要點：
- 使用
tryAcquire替代acquire避免線程阻塞 - 通過
finally保證許可釋放 - 集成監(jiān)控上報（Prometheus + Grafana）.
實現(xiàn)原理
Semaphore 有兩種模式，公平模式和非公平模式，分別對應兩個內(nèi)部類為 FairSync、NonfairSync，這兩個子類繼承了 Sync，都是基于之前講解過的 AQS 來實現(xiàn)的。
核心數(shù)據(jù)結(jié)構
畫個圖來說明一下內(nèi)部的結(jié)構如下:
圖片
Semaphore 的公平模式依賴于 FairSync 公平同步器來實現(xiàn)，非公平模式依賴于 NonfairSync 非公平同步器來實現(xiàn)。
其中 FairSync、NonfairSync 繼承自 Sync，而 Sync 又繼承自 AQS，這些同步器的底層都是依賴于 AQS 提供的機制來實現(xiàn)的。
public class Semaphore implements java.io.Serializable {
privatefinal Sync sync;
abstractstaticclass Sync extends AbstractQueuedSynchronizer {
Sync(int permits) { setState(permits); }
final int getPermits() { return getState(); }
// 非公平嘗試獲取許可
final int nonfairTryAcquireShared(int acquires) {
for (;;) {
int available = getState();
int remaining = available - acquires;
if (remaining < 0 || compareAndSetState(available, remaining))
return remaining;
}
}
// 釋放許可
protected final boolean tryReleaseShared(int releases) {
for (;;) {
int current = getState();
int next = current + releases;
if (next < current) thrownew Error("Maximum permit count exceeded");
if (compareAndSetState(current, next))
returntrue;
}
}
}
// 公平模式實現(xiàn)
staticfinalclass FairSync extends Sync {
protected int tryAcquireShared(int acquires) {
for (;;) {
if (hasQueuedPredecessors()) // 檢查是否有等待線程
return -1;
// ...與非公平模式相同
}
}
}
}所以掌握 AQS 很重要啊家人們，AQS 是模板方法模式的經(jīng)典運用。
這里的 Semaphore 實現(xiàn)的思路跟我們之前講過的 ReentrantLock 非常的相似，包括內(nèi)部類的結(jié)構都是一樣的，也是有公平和非公平兩種模式。
只是不同的是 Semaphore 是共享鎖，支持多個線程同時操作；然而 ReentrantLock 是互斥鎖，同一個時刻只允許一個線程操作。
公平模式 acquire
公平模式，Semaphore.acquire 方法源碼直接是調(diào)用 FairSync 的 acquireSharedInterruptibly，也就是進入了 AQS 的 acquireSharedInterruptibly 的模板方法里面了。
java.util.concurrent.Semaphore#acquire()源碼如下。
public void acquire() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}跳入 AQS 的 acquireSharedInterruptibly 方法。
java.util.concurrent.locks.AbstractQueuedSynchronizer#acquireSharedInterruptibly
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted() ||
// Semaphore.FairSync 子類實現(xiàn) tryAcquireShared
(tryAcquireShared(arg) < 0 &&
acquire(null, arg, true, true, false, 0L) < 0))
throw new InterruptedException();
}這個方法定義了一個模板流程：
- 先調(diào)用子類的
tryAcquireShared方法獲取共享鎖，也就是獲取信號量。 - 如果獲取信號量成功，即返回值大于等于 0，則直接返回。
- 如果獲取失敗，返回值小于 0，則調(diào)用 AQS 的
doAcquireSharedInterruptibly方法，進入 AQS 的等待隊列里面，等待別人釋放資源之后它再去獲取。
這里我們畫個圖理解一下：
圖片
Semaphore.FairSync 子類實現(xiàn) tryAcquireShared
protected int tryAcquireShared(int acquires) {
for (;;) {
// 這里作為公平模式，首先判斷一下AQS等待隊列里面
// 有沒有人在等待獲取信號量，如果有人排隊了，自己就不去獲取了
if (hasQueuedPredecessors())
return -1;
// 獲取剩余的信號量資源
int available = getState();
// 剩余資源減去我需要的資源，是否小于0
// 如果小于0則說明資源不夠了
// 如果大于等于0，說明資源是足夠我使用的
int remaining = available - acquires;
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}上面的源碼就是獲取信號量的核心流程了：
- 首先判斷一下 AQS 等待隊列里面是否有人在排隊，如果是，則自己不嘗試獲取資源了，乖乖的去排隊
- 如果沒有人在排隊，獲取一下當前剩余的信號量
available，然后減去自己需要的信號量acquires，得到減去后的結(jié)果remaining - 如果 remaining 小于 0，直接返回
remaining，說明資源不夠，獲取失敗了，這個時候就會進入 AQS 等待隊列等待。 - 如果
remaining大于等于 0，則執(zhí)行 CAS 操作compareAndSetState競爭資源，如果成功了，說明自己獲取信號量成功，如果失敗了同樣進入 AQS 等待隊列。
我們畫一下公平模式 FairSync 的 tryAcquireShared 流程圖，以及整個公平模式的 acquire 方法的流程圖：
圖片
公平模式 release
看完獲取，我們緊接著來看下釋放，這里 Semaphore 的 release 方法直接調(diào)用 Sync 的 releaseShared 方法：
public void release() {
sync.releaseShared(1);
}繼續(xù)來分析 releaseShared 方法，進入到 AQS 的 releaseShard 釋放資源的模板方法：
public final boolean releaseShared(int arg) {
// 1. 調(diào)用子類的tryReleaseShared釋放資源
if (tryReleaseShared(arg)) {
// 釋放資源成功，調(diào)用doReleaseShared喚醒等待隊列中等待資源的線程
doReleaseShared();
return true;
}
return false;
}這里的模板流程有：
- 調(diào)用子類的
tryReleaseShared去釋放資源，即釋放信號量 - 如果釋放成功了，則調(diào)用
doReleaseShared喚醒 AQS 中等待資源的線程，將資源傳播下去，如果釋放失敗，即返回小于等于 0，則直接返回。 - 所以，這里除了 AQS 的核心模板流程之外，具體釋放邏輯就是
Sync的tryReleaseShared方法的源碼了，我們繼續(xù)來查看：
protected final boolean tryReleaseShared(int releases) {
for (;;) {
int current = getState();
// 這里就是將釋放的信號量資源加回去而已
int next = current + releases;
if (next < current) // overflow
throw new Error("Maximum permit count exceeded");
// 嘗試CAS設置資源，成功直接返回，失敗則進入下一循環(huán)重試
if (compareAndSetState(current, next))
return true;
}
}釋放資源的流程圖如下：
圖片
Exchanger
Exchanger（交換者）是一個用于線程間協(xié)作的工具類。Exchanger 用于進行線程間的數(shù)據(jù)交 換。
它提供一個同步點，在這個同步點，兩個線程可以交換彼此的數(shù)據(jù)。這兩個線程通過 exchange 方法交換數(shù)據(jù)，如果第一個線程先執(zhí)行 exchange()方法，它會一直等待第二個線程也 執(zhí)行 exchange 方法，當兩個線程都到達同步點時，這兩個線程就可以交換數(shù)據(jù)，將本線程生產(chǎn) 出來的數(shù)據(jù)傳遞給對方。
圖片
使用場景
這個玩意的使用場景很少很少……大家對她有個了解即可，大可不必深入。
因為存在很多局限性。
- 僅限兩個線程
- 超過兩個線程使用同一
Exchanger會導致未定義行為。 - 替代方案：使用
CyclicBarrier或Phaser實現(xiàn)多線程同步。
- 阻塞風險
- 若一方線程未到達同步點，另一線程會永久阻塞。
- 解決方案：使用帶超時的
exchange(V x, long timeout, TimeUnit unit)。
- 性能瓶頸
- 頻繁交換大數(shù)據(jù)對象會導致內(nèi)存和 CPU 開銷。
- 優(yōu)化建議：交換輕量級對象（如引用或標識符），而非完整數(shù)據(jù)。
- 不適用于分布式系統(tǒng)
Exchanger僅限單 JVM 內(nèi)的線程通信。- 替代方案：消息隊列（如 Kafka）或 RPC 框架（如 gRPC）。
Exchanger在多種并發(fā)編程場景中都非常有用。例如，在遺傳算法中，可以使用Exchanger來實現(xiàn)個體之間的信息交換；在管道設計中，可以使用Exchanger來傳遞數(shù)據(jù)塊或任務；在游戲中，可以使用Exchanger來實現(xiàn)玩家之間的物品交易等。
如下代碼，用 Exchanger 實現(xiàn)兩個線程將交換彼此持有的字符串數(shù)據(jù)：
import java.util.concurrent.Exchanger;
publicclass ExchangerExample {
public static void main(String[] args) {
// 創(chuàng)建一個Exchanger對象
Exchanger<String> exchanger = new Exchanger<>();
// 創(chuàng)建一個線程，它將使用"Hello"與另一個線程交換數(shù)據(jù)
Thread producer = new Thread(() -> {
try {
String producedData = "Hello";
String consumerData = exchanger.exchange(producedData);
System.out.println("生產(chǎn)者線程交換后得到的數(shù)據(jù): " + consumerData);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}, "生產(chǎn)者線程");
// 創(chuàng)建一個線程，它將使用"World"與另一個線程交換數(shù)據(jù)
Thread consumer = new Thread(() -> {
try {
String consumerData = "World";
String producedData = exchanger.exchange(consumerData);
System.out.println("消費者線程交換后得到的數(shù)據(jù): " + producedData);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}, "消費者線程");
// 啟動線程
producer.start();
consumer.start();
}
}代碼中我們創(chuàng)建了一個 Exchanger 對象，并且定義了兩個線程：一個生產(chǎn)者線程和一個消費者線程。
生產(chǎn)者線程持有一個字符串 "Hello"，而消費者線程持有一個字符串 "World"。兩個線程都通過調(diào)用 exchanger.exchange() 方法來等待交換數(shù)據(jù)。
當兩個線程都到達交換點時（即都調(diào)用了 exchange() 方法），Exchanger 會確保它們安全地交換數(shù)據(jù)。交換完成后，每個線程都會得到對方原本持有的數(shù)據(jù)，并打印出來。
實現(xiàn)原理
分別從數(shù)據(jù)結(jié)構和 exchange 方法來實現(xiàn)流程來學習實現(xiàn)原理。
核心數(shù)據(jù)結(jié)構
Participant 線程本地存儲
public class Exchanger<V> {
// 每個線程持有一個Node
private final Participant participant;
static final class Participant extends ThreadLocal<Node> {
public Node initialValue() { return new Node(); }
}
}
圖片
關鍵作用：
- 每個線程通過
Participant持有獨立的Node對象 - 避免多線程競爭同一存儲位置
- 底層使用
ThreadLocal實現(xiàn)線程隔離
Node 交換節(jié)點設計
@sun.misc.Contended // 防止偽共享
static final class Node {
int index; // Arena下標
int bound; // 最近記錄的前導邊界
int collides; // CAS失敗計數(shù)
int hash; // 偽隨機自旋
Object item; // 攜帶的數(shù)據(jù)
volatile Object match; // 交換的數(shù)據(jù)
volatile Thread parked; // 掛起的線程
}內(nèi)存布局優(yōu)化：
- 使用
@Contended注解填充緩存行（64 字節(jié)） - 確保不同線程訪問的字段不在同一緩存行
- 示例內(nèi)存布局：
| 64字節(jié)緩存行 | Node.item | ...填充... |
| 64字節(jié)緩存行 | Node.match | ...填充... |每個線程的 Node 有一個 match 屬性用于存儲待交換的數(shù)據(jù)。
exchange 方法執(zhí)行流程
主流程源碼（精簡版）
public V exchange(V x) throws InterruptedException {
Object v;
Node[] a;
Node q = participant.get();
// Arena模式（
if ((a = arena) != null ||
(q = slotExchange(q, x, false, 0L)) == null)
return (V)v;
// ...省略超時處理
}
private final Object slotExchange(Node q, Object x, boolean timed, long nanos) {
// 核心交換邏輯（
for (;;) {
if (slot != null) { // 存在等待節(jié)點
Node node = (Node)slot;
if (U.compareAndSwapObject(this, SLOT, node, null)) {
Object v = node.item;
node.match = x; // 數(shù)據(jù)交換（
Thread t = node.parked;
if (t != null)
U.unpark(t); // 喚醒對方線程
return v;
}
} elseif (U.compareAndSwapObject(this, SLOT, null, q)) {
// 掛起當前線程
return timed ? awaitNanos(q, nanos) : await(q);
}
}
}
關鍵步驟解釋：
- CAS 設置槽位：
U.compareAndSwapObject(this, SLOT, null, q) - 數(shù)據(jù)交換：直接修改對方節(jié)點的
match字段 - 喚醒機制：通過
Unsafe.unpark()解除線程阻塞
