《Java 并发编程·线程与内存模型》12.1 工作窃取与 ForkJoinPool

worker 拿到父任务 → 拆成两半 → 一半提交回队列 → 自己算另一半 → 阻塞等子任务
                                     ↓
                          子任务在队列里等空闲线程
                                     ↓
                          所有线程都在等自己提交的任务

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

public final class SumTask extends RecursiveTask<Long> {
    private final int[] data;
    private final int lo;
    private final int hi;
    private final int threshold;

    public SumTask(int[] data, int lo, int hi, int threshold) {
        this.data = data;
        this.lo = lo;
        this.hi = hi;
        this.threshold = threshold;
    }

    @Override
    protected Long compute() {
        if (hi - lo <= threshold) {
            long sum = 0;
            for (int i = lo; i < hi; i++) {
                sum += data[i];
            }
            return sum;
        }
        int mid = (lo + hi) >>> 1;
        SumTask left = new SumTask(data, lo, mid, threshold);
        SumTask right = new SumTask(data, mid, hi, threshold);
        left.fork();                        // 左半挂到本线程队列尾部
        long rightSum = right.compute();    // 右半在本线程原地算
        long leftSum = left.join();         // 没被偷走就在这里算，被偷走就等
        return leftSum + rightSum;
    }
}

// 1. 默认：核数 - 1，最少 1
int p = ForkJoinPool.commonPool().getParallelism();

// 2. 系统属性覆盖（全 JVM 生效，启动时确定）
//    -Djava.util.concurrent.ForkJoinPool.common.parallelism=3

// 3. 自建池：显式指定
try (ForkJoinPool pool = new ForkJoinPool(4)) {
    long sum = pool.invoke(new SumTask(data, 0, data.length, 1 << 14));
}

availableProcessors = 10
commonPool.parallelism = 9
commonPool.poolSize (未使用前) = 0
N = 100000000, threshold = 16384
expected sum = 4950407612

sequential loop        runs=[66, 66, 67, 68] median=67ms
commonPool             parallelism= 9 poolSize= 9 runs=[13ms  13ms  14ms] median=13ms steals=189 ok=true
   叶子落在 10 个线程：cw-1=622, cw-2=1421, cw-3=1292, cw-4=570, cw-5=614, cw-6=1206, cw-7=365, cw-8=1239, cw-9=159, main=704
new ForkJoinPool(4)    parallelism= 4 poolSize= 7 runs=[16ms  17ms  17ms] median=17ms steals=44 ok=true
   叶子落在 5 个线程：fj-1-worker-1=5, fj-1-worker-3=2074, fj-1-worker-5=1976, fj-1-worker-6=2068, fj-1-worker-7=2069
new ForkJoinPool(10)   parallelism=10 poolSize=10 runs=[9ms  10ms  10ms] median=10ms steals=176 ok=true
commonPool.poolSize (使用后) = 9
commonPool.stealCount 累计 = 234

commonPool.parallelism = 3
commonPool             parallelism= 3 poolSize= 4 runs=[17ms  17ms  23ms] median=17ms steals=32 ok=true
   叶子落在 4 个线程：cw-2=2064, cw-3=2071, cw-4=2025, main=2032

package com.plume.conc.core;

import java.util.List;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

public final class StockAudit {
    public record Report(long totalStock, int outOfStock, int titles) {}

    private static final class AuditTask extends RecursiveTask<Report> {
        private static final int THRESHOLD = 512;
        private final List<Book> books;
        private final int lo;
        private final int hi;

        AuditTask(List<Book> books, int lo, int hi) {
            this.books = books;
            this.lo = lo;
            this.hi = hi;
        }

        @Override
        protected Report compute() {
            if (hi - lo <= THRESHOLD) {
                long stock = 0;
                int out = 0;
                for (int i = lo; i < hi; i++) {
                    Book book = books.get(i);
                    stock += book.stock();
                    if (book.stock() == 0) {
                        out++;
                    }
                }
                return new Report(stock, out, hi - lo);
            }
            int mid = (lo + hi) >>> 1;
            AuditTask left = new AuditTask(books, lo, mid);
            AuditTask right = new AuditTask(books, mid, hi);
            left.fork();
            Report r = right.compute();
            Report l = left.join();
            return new Report(l.totalStock() + r.totalStock(),
                              l.outOfStock() + r.outOfStock(),
                              l.titles() + r.titles());
        }
    }

    private final ForkJoinPool pool = new ForkJoinPool(4);

    public Report audit(List<Book> snapshot) {
        if (snapshot.isEmpty()) {
            return new Report(0, 0, 0);
        }
        return pool.invoke(new AuditTask(snapshot, 0, snapshot.size()));
    }

    public void close() {
        pool.shutdown();
    }
}