Fork/Join 框架与并行 Stream

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

public class ArraySumTask extends RecursiveTask<Long> {

    private static final int THRESHOLD = 5000;  // 拆分阈值
    private final int[] array;
    private final int start, end;

    public ArraySumTask(int[] array, int start, int end) {
        this.array = array;
        this.start = start;
        this.end   = end;
    }

    @Override
    protected Long compute() {
        // 基线条件：子任务足够小，直接计算
        if (end - start <= THRESHOLD) {
            long sum = 0;
            for (int i = start; i < end; i++) {
                sum += array[i];
            }
            return sum;
        }

        // 拆分：从中间一分为二
        int mid = (start + end) / 2;
        ArraySumTask left  = new ArraySumTask(array, start, mid);
        ArraySumTask right = new ArraySumTask(array, mid, end);

        // fork() 将子任务提交到 ForkJoinPool 的队列中异步执行
        left.fork();   // 异步执行左半部分

        // 当前线程直接计算右半部分（比再 fork 一个线程更高效）
        long rightResult = right.compute();  // 注意：不是 right.fork()!

        // join() 等待左半部分完成并获取结果
        long leftResult = left.join();

        return leftResult + rightResult;
    }
}

int[] data = new int[1_000_000];
// ... 填充数据 ...

// 方式一：使用公共 ForkJoinPool（parallelStream 也用这个）
ForkJoinPool commonPool = ForkJoinPool.commonPool();
long result = commonPool.invoke(new ArraySumTask(data, 0, data.length));

// 方式二：创建自定义 ForkJoinPool（生产推荐，隔离任务）
ForkJoinPool customPool = new ForkJoinPool(4);  // 4 个工作线程
try {
    result = customPool.invoke(new ArraySumTask(data, 0, data.length));
} finally {
    customPool.shutdown();
}

List<Integer> numbers = IntStream.rangeClosed(1, 10_000_000)
    .boxed()
    .collect(Collectors.toList());

// 串行：单线程
long serialSum = numbers.stream()
    .filter(n -> n % 2 == 0)
    .mapToLong(Integer::longValue)
    .sum();

// 并行：自动利用多核
long parallelSum = numbers.parallelStream()
    .filter(n -> n % 2 == 0)
    .mapToLong(Integer::longValue)
    .sum();

// reduce 写法
long reduced = numbers.parallelStream()
    .mapToLong(Integer::longValue)
    .reduce(0L, Long::sum);

// ❌ 错误示范：多个线程同时写同一个 ArrayList
List<String> results = new ArrayList<>();
list.parallelStream()
    .map(User::getName)
    .forEach(results::add);   // 💥 ArrayList 不是线程安全的！
// 结果：数据丢失、ArrayIndexOutOfBoundsException、甚至死循环

// ✅ 正确做法：用 collect 代替外部集合
List<String> results = list.parallelStream()
    .map(User::getName)
    .collect(Collectors.toList());  // Collectors 内部处理了并发安全

// ❌ sorted() 在并行流下需要先收集所有元素才能排序
// 并行优势被完全抵消，甚至更慢（多了一次并行→串行→并行的切换）
list.parallelStream()
    .sorted()       // 💥 有状态中间操作，破坏并行性
    .distinct()     // 💥 同样有状态，需要全局去重
    .limit(100)     // ⚠️ 在无序流上还行，有序流上会等待前序完成
    .collect(Collectors.toList());

// ✅ 如果不需要原始顺序，用 unordered() 释放约束
list.parallelStream()
    .unordered()    // 告诉框架"我不关心顺序"
    .distinct()
    .limit(100)
    .collect(Collectors.toList());

// ❌ Stream<Integer> 每个元素都是 Integer 对象，内存浪费 + GC 压力
int sum = list.parallelStream()
    .map(e -> e * 2)          // 返回 Stream<Integer>，装箱！
    .reduce(0, Integer::sum);   // 拆箱再求和

// ✅ 使用原始类型流（IntStream / LongStream / DoubleStream）
long sum = list.parallelStream()
    .mapToInt(Integer::intValue)  // 转 IntStream，无装箱
    .map(e -> e * 2)
    .asLongStream()
    .sum();

// ❌ 只有 100 个元素，拆分/调度/合并的开销远大于并行收益
List<Integer> small = Arrays.asList(1, 2, 3, ... , 100);
int sum = small.parallelStream().mapToInt(Integer::intValue).sum();
// 比 stream() 慢 5~10 倍！

// ✅ 小数据用普通 stream 即可
int sum = small.stream().mapToInt(Integer::intValue).sum();

// ❌ 外层和内层都用 parallelStream —— 争抢同一个 commonPool
outerList.parallelStream()
    .flatMap(item ->
        innerList.parallelStream()  // 💥 嵌套并行，线程互相等待
            .map(x -> process(item, x))
    )
    .collect(Collectors.toList());

// ✅ 只在一个层级使用 parallelStream
outerList.parallelStream()
    .flatMap(item ->
        innerList.stream()  // 内层用串行 stream
            .map(x -> process(item, x))
    )
    .collect(Collectors.toList());

// 你的任务是什么类型？

if (任务是 "集合数据的并行计算" && 数据量 > 10000 && 纯 CPU 操作) {
    → 用 parallelStream（简洁，够用）
}
else if (任务是 "递归分治" && 需要精细控制拆分阈值) {
    → 用 ForkJoinPool + RecursiveTask（自定义并行算法）
}
else if (任务是 "多个异步 IO 操作的编排") {
    → 用 CompletableFuture（异步回调 + thenCompose/thenCombine）
}
else {
    → 用普通 ThreadPoolExecutor（最简单，覆盖 80% 场景）
}

// 生产环境：不要依赖 commonPool，用自定义池隔离
ForkJoinPool customPool = new ForkJoinPool(8);

List<String> result;
try {
    // submit 一个 Callable，在里面使用 parallelStream
    // 此时 parallelStream 会使用 customPool 而不是 commonPool
    result = customPool.submit(() ->
        dataList.parallelStream()
            .filter(e -> e.length() > 5)
            .map(String::toUpperCase)
            .collect(Collectors.toList())
    ).get();  // get() 阻塞等待完成
} finally {
    customPool.shutdown();
}