Java 并发编程与 JUC 包深度指南

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 演示 ReentrantLock 的可重入、可中断、超时获取特性
 */
public class ReentrantLockDemo {
    // 公平锁：按请求顺序获取，吞吐量略低
    private final Lock lock = new ReentrantLock(true);
    private int count = 0;

    public void increment() {
        lock.lock(); // 获取锁
        try {
            // 可重入：同一线程可再次获取该锁而不会死锁
            count++;
            System.out.println(Thread.currentThread().getName() + " 计数: " + count);
        } finally {
            lock.unlock(); // 必须在 finally 中释放，避免异常导致死锁
        }
    }

    // 演示 tryLock 超时获取
    public boolean tryIncrementWithTimeout() throws InterruptedException {
        if (lock.tryLock(2, java.util.concurrent.TimeUnit.SECONDS)) {
            try {
                count++;
                return true;
            } finally {
                lock.unlock();
            }
        }
        System.out.println(Thread.currentThread().getName() + " 获取锁超时");
        return false;
    }

    public static void main(String[] args) {
        ReentrantLockDemo demo = new ReentrantLockDemo();
        Runnable task = () -> {
            for (int i = 0; i < 5; i++) {
                demo.increment();
            }
        };

        Thread t1 = new Thread(task, "线程-A");
        Thread t2 = new Thread(task, "线程-B");
        t1.start();
        t2.start();
    }
}

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 使用 Condition 实现生产者-消费者模式，支持多条件精准唤醒
 */
public class ConditionProducerConsumer {
    private final Lock lock = new ReentrantLock();
    private final Condition notFull = lock.newCondition();  // 队列未满条件
    private final Condition notEmpty = lock.newCondition(); // 队列非空条件
    private final int[] buffer = new int[10];
    private int count = 0;
    private int putIndex = 0;
    private int takeIndex = 0;

    public void produce(int value) throws InterruptedException {
        lock.lock();
        try {
            // 当缓冲区满时，生产者等待
            while (count == buffer.length) {
                notFull.await();
            }
            buffer[putIndex] = value;
            putIndex = (putIndex + 1) % buffer.length;
            count++;
            System.out.println(Thread.currentThread().getName() + " 生产: " + value + "，当前库存: " + count);
            // 只唤醒等待消费的线程
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }

    public int consume() throws InterruptedException {
        lock.lock();
        try {
            // 当缓冲区空时，消费者等待
            while (count == 0) {
                notEmpty.await();
            }
            int value = buffer[takeIndex];
            takeIndex = (takeIndex + 1) % buffer.length;
            count--;
            System.out.println(Thread.currentThread().getName() + " 消费: " + value + "，当前库存: " + count);
            // 只唤醒等待生产的线程
            notFull.signal();
            return value;
        } finally {
            lock.unlock();
        }
    }

    public static void main(String[] args) {
        ConditionProducerConsumer pc = new ConditionProducerConsumer();
        Thread producer = new Thread(() -> {
            for (int i = 0; i < 20; i++) {
                try {
                    pc.produce(i);
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }, "生产者");

        Thread consumer = new Thread(() -> {
            for (int i = 0; i < 20; i++) {
                try {
                    pc.consume();
                    Thread.sleep(150);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }, "消费者");

        producer.start();
        consumer.start();
    }
}

import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.HashMap;
import java.util.Map;

/**
 * 演示 ReadWriteLock：读读共享，读写互斥，写写互斥
 * 适用于读多写少的缓存场景
 */
public class ReadWriteLockCache {
    private final ReadWriteLock rwl = new ReentrantReadWriteLock();
    private final Map<String, String> cache = new HashMap<>();

    public String get(String key) {
        // 先加读锁查询缓存
        rwl.readLock().lock();
        try {
            String value = cache.get(key);
            if (value != null) {
                return value;
            }
        } finally {
            rwl.readLock().unlock();
        }

        // 缓存未命中，加写锁加载数据
        rwl.writeLock().lock();
        try {
            // 双重检查：可能其他线程已写入
            String value = cache.get(key);
            if (value == null) {
                value = loadFromDatabase(key);
                cache.put(key, value);
            }
            return value;
        } finally {
            rwl.writeLock().unlock();
        }
    }

    // 模拟从数据库加载
    private String loadFromDatabase(String key) {
        System.out.println("从数据库加载: " + key);
        return "ValueOf" + key;
    }

    public void put(String key, String value) {
        rwl.writeLock().lock();
        try {
            cache.put(key, value);
        } finally {
            rwl.writeLock().unlock();
        }
    }

    public static void main(String[] args) {
        ReadWriteLockCache cache = new ReadWriteLockCache();
        // 启动多个读线程
        for (int i = 0; i < 5; i++) {
            new Thread(() -> System.out.println("读取: " + cache.get("key1")), "读线程" + i).start();
        }
        // 启动写线程
        new Thread(() -> cache.put("key1", "新值"), "写线程").start();
    }
}

import java.util.concurrent.locks.StampedLock;

/**
 * StampedLock 支持三种模式：读锁、写锁、乐观读
 * 乐观读不会阻塞写操作，适合读极多、写极少的场景
 */
public class StampedLockDemo {
    private final StampedLock lock = new StampedLock();
    private double x = 0.0;
    private double y = 0.0;

    // 乐观读：先读数据，再验证stamp是否被写过
    public double distanceFromOrigin() {
        long stamp = lock.tryOptimisticRead(); // 获取乐观读戳
        double currentX = x;
        double currentY = y;
        // 验证stamp：若期间有写操作，stamp会变化
        if (!lock.validate(stamp)) {
            // 被污染了，升级为悲观读锁
            stamp = lock.readLock();
            try {
                currentX = x;
                currentY = y;
            } finally {
                lock.unlockRead(stamp);
            }
        }
        return Math.sqrt(currentX * currentX + currentY * currentY);
    }

    public void move(double deltaX, double deltaY) {
        long stamp = lock.writeLock();
        try {
            x += deltaX;
            y += deltaY;
        } finally {
            lock.unlockWrite(stamp);
        }
    }

    public static void main(String[] args) {
        StampedLockDemo point = new StampedLockDemo();
        point.move(3, 4);
        System.out.println("距离原点: " + point.distanceFromOrigin());
    }
}

import java.util.concurrent.CountDownLatch;

/**
 * CountDownLatch：一个或多个线程等待其他线程完成操作后再执行
 * 倒计时器不可重置，用完后需重新创建实例
 */
public class CountDownLatchDemo {
    public static void main(String[] args) throws InterruptedException {
        int workerCount = 3;
        CountDownLatch latch = new CountDownLatch(workerCount);

        for (int i = 0; i < workerCount; i++) {
            final int id = i;
            new Thread(() -> {
                System.out.println("工作者 " + id + " 开始工作...");
                try {
                    Thread.sleep((id + 1) * 1000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                System.out.println("工作者 " + id + " 完成工作");
                latch.countDown(); // 计数减一
            }).start();
        }

        System.out.println("主线程等待所有工作者完成...");
        latch.await(); // 阻塞，直到计数归零
        System.out.println("所有工作者已完成，主线程继续执行");
    }
}

import java.util.concurrent.CyclicBarrier;

/**
 * CyclicBarrier：让一组线程到达屏障后互相等待，直到所有线程都到达后才继续执行
 * 与 CountDownLatch 不同：CyclicBarrier 可重置（reset），支持重复使用
 */
public class CyclicBarrierDemo {
    public static void main(String[] args) {
        int party = 3;
        // 所有线程到达后执行一次 Runnable（屏障操作）
        CyclicBarrier barrier = new CyclicBarrier(party, () -> {
            System.out.println("--- 所有线程已到达屏障，本轮汇合完成 ---");
        });

        Runnable task = () -> {
            for (int round = 1; round <= 2; round++) {
                try {
                    System.out.println(Thread.currentThread().getName() + " 正在执行第 " + round + " 轮任务...");
                    Thread.sleep((int) (Math.random() * 2000));
                    System.out.println(Thread.currentThread().getName() + " 到达屏障");
                    barrier.await(); // 到达屏障并等待
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        };

        for (int i = 0; i < party; i++) {
            new Thread(task, "队员-" + i).start();
        }
    }
}

import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

/**
 * Semaphore：信号量，控制同时访问某个资源的线程数量
 * 常用于限流，如数据库连接池控制
 */
public class SemaphoreDemo {
    private final Semaphore semaphore = new Semaphore(3); // 最多允许3个线程同时访问

    public void accessResource(int threadId) {
        try {
            // 获取许可，若当前无可用许可则阻塞
            semaphore.acquire();
            System.out.println("线程 " + threadId + " 获取许可，开始访问资源");
            TimeUnit.SECONDS.sleep(2);
            System.out.println("线程 " + threadId + " 释放许可，访问结束");
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            semaphore.release(); // 释放许可，供其他线程使用
        }
    }

    public static void main(String[] args) {
        SemaphoreDemo demo = new SemaphoreDemo();
        for (int i = 0; i < 8; i++) {
            final int id = i;
            new Thread(() -> demo.accessResource(id), "Thread-" + i).start();
        }
    }
}

import java.util.concurrent.Exchanger;

/**
 * Exchanger：两个线程在Exchange点交换数据
 * 典型应用：遗传算法、管道数据校对
 */
public class ExchangerDemo {
    public static void main(String[] args) {
        Exchanger<String> exchanger = new Exchanger<>();

        Thread threadA = new Thread(() -> {
            try {
                String dataA = "来自线程A的数据";
                System.out.println("线程A准备交换数据: " + dataA);
                // 到达交换点后阻塞，直到线程B也到达
                String received = exchanger.exchange(dataA);
                System.out.println("线程A收到: " + received);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }, "线程A");

        Thread threadB = new Thread(() -> {
            try {
                Thread.sleep(2000); // 模拟延迟
                String dataB = "来自线程B的数据";
                System.out.println("线程B准备交换数据: " + dataB);
                String received = exchanger.exchange(dataB);
                System.out.println("线程B收到: " + received);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }, "线程B");

        threadA.start();
        threadB.start();
    }
}

import java.util.concurrent.Phaser;

/**
 * Phaser：支持动态注册/注销参与者的分阶段同步器
 * 比 CountDownLatch 和 CyclicBarrier 更灵活，支持多阶段
 */
public class PhaserDemo {
    public static void main(String[] args) {
        Phaser phaser = new Phaser(1); // 注册主线程

        for (int i = 0; i < 3; i++) {
            phaser.register(); // 动态注册参与者
            new Thread(new Worker(phaser, i)).start();
        }

        System.out.println("主线程等待所有参与者完成阶段0...");
        phaser.arriveAndAwaitAdvance(); // 主线程到达并等待
        System.out.println("阶段0完成，进入阶段1...");
        phaser.arriveAndAwaitAdvance();
        System.out.println("阶段1完成");
        phaser.arriveAndDeregister(); // 主线程注销
    }

    static class Worker implements Runnable {
        private final Phaser phaser;
        private final int id;

        Worker(Phaser phaser, int id) {
            this.phaser = phaser;
            this.id = id;
        }

        @Override
        public void run() {
            System.out.println("工作者 " + id + " 到达阶段0");
            phaser.arriveAndAwaitAdvance(); // 到达阶段0并等待

            // 阶段1工作
            System.out.println("工作者 " + id + " 执行阶段1任务...");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }

            System.out.println("工作者 " + id + " 到达阶段1");
            phaser.arriveAndDeregister(); // 到达阶段1并注销
        }
    }
}

import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

/**
 * AtomicInteger 基于 CAS（Compare-And-Swap）实现无锁线程安全
 * 底层调用 Unsafe.compareAndSwapInt，依赖 CPU 原子指令
 */
public class AtomicDemo {
    private final AtomicInteger counter = new AtomicInteger(0);
    private final AtomicReference<String> name = new AtomicReference<>("初始值");

    public void increment() {
        // getAndIncrement 等价于 i++ 的原子版本
        int newValue = counter.getAndIncrement();
        System.out.println(Thread.currentThread().getName() + " 递增后: " + newValue);
    }

    // 乐观锁模式：CAS 更新，失败则重试
    public void casUpdateName(String expected, String newName) {
        boolean updated = name.compareAndSet(expected, newName);
        System.out.println("CAS更新" + (updated ? "成功" : "失败") + "，当前值: " + name.get());
    }

    public static void main(String[] args) {
        AtomicDemo demo = new AtomicDemo();
        for (int i = 0; i < 5; i++) {
            new Thread(demo::increment, "线程-" + i).start();
        }

        demo.casUpdateName("初始值", "新值A");
        demo.casUpdateName("初始值", "新值B"); // 预期已变，CAS失败
    }
}

import java.util.concurrent.atomic.LongAdder;

/**
 * LongAdder 在高并发下性能优于 AtomicLong
 * 原理：分散热点，内部维护 Cell[] 数组，不同线程操作不同 Cell，最后汇总
 * Serial（低竞争）场景 AtomicLong 更优，因 LongAdder 有额外汇总开销
 */
public class LongAdderDemo {
    private final LongAdder adder = new LongAdder();

    public void increment() {
        adder.increment(); // 底层无锁，分散竞争
    }

    public long getSum() {
        return adder.sum(); // 获取当前总和
    }

    public static void main(String[] args) throws InterruptedException {
        LongAdderDemo demo = new LongAdderDemo();
        Thread[] threads = new Thread[10];
        for (int i = 0; i < threads.length; i++) {
            threads[i] = new Thread(() -> {
                for (int j = 0; j < 10000; j++) {
                    demo.increment();
                }
            });
            threads[i].start();
        }
        for (Thread t : threads) {
            t.join();
        }
        System.out.println("最终累加结果: " + demo.getSum());
    }
}

import java.util.concurrent.atomic.AtomicStampedReference;

/**
 * ABA 问题：线程A读取值为A，线程B改为B又改回A，线程A的CAS仍能成功，但值已变化过
 * AtomicStampedReference 用版本号（stamp）解决，每次修改 stamp 自增
 */
public class AbaProblemDemo {
    public static void main(String[] args) throws InterruptedException {
        // 初始值 100，版本号 0
        AtomicStampedReference<Integer> value = new AtomicStampedReference<>(100, 0);

        int[] stampHolder = new int[1];
        Integer current = value.get(stampHolder);
        int initialStamp = stampHolder[0];
        System.out.println("初始值: " + current + "，stamp: " + initialStamp);

        // 线程B：制造 ABA 问题（100→101→100）
        Thread threadB = new Thread(() -> {
            int[] holder = new int[1];
            value.get(holder);
            int stamp = holder[0];
            // 第一次 CAS：100→101
            value.compareAndSet(100, 101, stamp, stamp + 1);
            // 获取最新 stamp
            value.get(holder);
            stamp = holder[0];
            // 第二次 CAS：101→100
            value.compareAndSet(101, 100, stamp, stamp + 1);
            System.out.println("线程B完成 ABA 操纵，当前stamp: " + value.get(holder));
        });

        threadB.start();
        threadB.join();

        // 线程A：用旧 stamp 尝试 CAS，会失败
        boolean result = value.compareAndSet(100, 200, initialStamp, initialStamp + 1);
        System.out.println("线程A用旧stamp CAS结果: " + result); // false，因为 stamp 已变

        // 用当前最新 stamp 重试
        int[] finalHolder = new int[1];
        value.get(finalHolder);
        result = value.compareAndSet(100, 200, finalHolder[0], finalHolder[0] + 1);
        System.out.println("线程A用新stamp CAS结果: " + result); // true
    }
}

import sun.misc.Unsafe;
import java.lang.reflect.Field;

/**
 * 使用 Unsafe 理解底层 CAS 实现
 * 生产环境不应直接使用 Unsafe，应使用封装好的原子类
 */
public class UnsafeCasDemo {
    private volatile long value = 0;
    private static final Unsafe UNSAFE;
    private static final long VALUE_OFFSET;

    static {
        try {
            // 通过反射获取 Unsafe 实例
            Field field = Unsafe.class.getDeclaredField("theUnsafe");
            field.setAccessible(true);
            UNSAFE = (Unsafe) field.get(null);
            // 获取 value 字段的内存偏移量
            VALUE_OFFSET = UNSAFE.objectFieldOffset(UnsafeCasDemo.class.getDeclaredField("value"));
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public boolean compareAndSwap(long expected, long newValue) {
        // 底层调用 CPU cmpxchg 指令
        return UNSAFE.compareAndSwapLong(this, VALUE_OFFSET, expected, newValue);
    }

    public static void main(String[] args) {
        UnsafeCasDemo demo = new UnsafeCasDemo();
        boolean result = demo.compareAndSwap(0, 1);
        System.out.println("首次 CAS 结果: " + result + "，当前值: " + demo.value);
        result = demo.compareAndSwap(0, 2);
        System.out.println("二次 CAS 结果: " + result + "，当前值: " + demo.value);
    }
}

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

/**
 * CompletableFuture 入门：创建、完成、异常处理
 */
public class CompletableFutureBasic {
    public static void main(String[] args) throws Exception {
        // 1. runAsync：无返回值异步任务
        CompletableFuture<Void> future1 = CompletableFuture.runAsync(() -> {
            System.out.println("runAsync 执行线程: " + Thread.currentThread().getName());
        });

        // 2. supplyAsync：有返回值异步任务
        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            return "Hello";
        });
        System.out.println("supplyAsync 结果: " + future2.get());

        // 3. 手动完成
        CompletableFuture<String> future3 = new CompletableFuture<>();
        new Thread(() -> future3.complete("手动完成")).start();
        System.out.println("手动完成结果: " + future3.get());
    }
}

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

/**
 * 演示 thenCompose、thenCombine、thenAccept、thenApply 的区别与组合
 */
public class CompletableFutureChain {
    public static void main(String[] args) throws Exception {
        // thenApply：转换结果类型 T -> U
        CompletableFuture<Integer> lengthFuture = CompletableFuture
                .supplyAsync(() -> {
                    System.out.println("步骤1：获取字符串");
                    return "CompletableFuture";
                })
                .thenApply(s -> {
                    System.out.println("步骤2：计算长度");
                    return s.length();
                });
        System.out.println("thenApply 结果: " + lengthFuture.get());

        // thenCompose：将结果扁平化为另一个 CompletableFuture（避免 Future<Future<T>>）
        CompletableFuture<String> composed = CompletableFuture
                .supplyAsync(() -> "user:123")
                .thenCompose(userId -> fetchUserName(userId)); // 返回 CompletableFuture<String>
        System.out.println("thenCompose 结果: " + composed.get());

        // thenCombine：合并两个独立任务的结果
        CompletableFuture<Double> combined = fetchPrice("Apple").thenCombine(
                fetchExchangeRate(),
                (price, rate) -> price * rate // (T, U) -> V
        );
        System.out.println("thenCombine 结果（美元价格）: " + combined.get());

        // allOf：等待所有任务完成（无返回值，需自行从各 Future 取结果）
        CompletableFuture<String> taskA = asyncTask("A", 1);
        CompletableFuture<String> taskB = asyncTask("B", 2);
        CompletableFuture<String> taskC = asyncTask("C", 1);
        CompletableFuture<Void> allDone = CompletableFuture.allOf(taskA, taskB, taskC);
        allDone.thenRun(() -> {
            try {
                System.out.println("allOf 全部完成: " + taskA.get() + ", " + taskB.get() + ", " + taskC.get());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).get();

        // anyOf：任一任务完成即返回
        CompletableFuture<Object> anyDone = CompletableFuture.anyOf(
                asyncTask("快任务", 1),
                asyncTask("慢任务", 3)
        );
        System.out.println("anyOf 最先完成: " + anyDone.get());
    }

    static CompletableFuture<String> fetchUserName(String userId) {
        return CompletableFuture.supplyAsync(() -> "NameOf" + userId);
    }

    static CompletableFuture<Double> fetchPrice(String product) {
        return CompletableFuture.supplyAsync(() -> 100.0);
    }

    static CompletableFuture<Double> fetchExchangeRate() {
        return CompletableFuture.supplyAsync(() -> 7.2);
    }

    static CompletableFuture<String> asyncTask(String name, int seconds) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(seconds);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            return name + " 完成";
        });
    }
}

import java.util.concurrent.CompletableFuture;

/**
 * CompletableFuture 异常处理：exceptionally 与 handle 的区别
 * exceptionally：发生异常时提供替代结果，正常情况不执行
 * handle：无论是否异常都执行，类似 try-catch-finally 的结合
 */
public class CompletableFutureException {
    public static void main(String[] args) {
        // 1. exceptionally：仅异常时执行，返回替代值
        CompletableFuture<Integer> withExceptionally = CompletableFuture
                .supplyAsync(() -> 10 / 0) // 触发 ArithmeticException
                .exceptionally(ex -> {
                    System.out.println("捕获异常: " + ex.getMessage());
                    return -1; // 提供降级值
                });
        System.out.println("exceptionally 结果: " + withExceptionally.join());

        // 2. handle：无论成败都处理，接收正常结果或异常
        CompletableFuture<Integer> withHandle = CompletableFuture
                .supplyAsync(() -> 10 / 5)
                .handle((result, ex) -> {
                    if (ex != null) {
                        System.out.println("handle 捕获异常: " + ex.getMessage());
                        return -1;
                    }
                    return result * 2; // 正常情况继续处理
                });
        System.out.println("handle 结果: " + withHandle.join());

        // 3. 结合自定义线程池与超时控制
        java.util.concurrent.Executor executor = java.util.concurrent.Executors.newFixedThreadPool(4);
        CompletableFuture<String> future = CompletableFuture
                .supplyAsync(() -> {
                    try {
                        Thread.sleep(5000);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                    return "太慢了";
                }, executor)
                .orTimeout(2, java.util.concurrent.TimeUnit.SECONDS) // JDK 9+ 超时控制
                .exceptionally(ex -> "超时或异常: " + ex.getClass().getSimpleName());
        System.out.println("超时测试: " + future.join());

        ((java.util.concurrent.ExecutorService) executor).shutdown();
    }
}

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

/**
 * ForkJoinPool + RecursiveTask：大任务递归拆分为小任务，多线程并行计算
 * "工作窃取"机制：线程完成自己队列任务后，从其他线程队列尾部窃取任务执行
 * 适用于：递归可分解的计算密集型任务（大数据求和、排序、MapReduce）
 */
public class ForkJoinSum extends RecursiveTask<Long> {
    private static final int THRESHOLD = 10000; // 拆分阈值
    private final long[] numbers;
    private final int start;
    private final int end;

    public ForkJoinSum(long[] numbers, int start, int end) {
        this.numbers = numbers;
        this.start = start;
        this.end = end;
    }

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

        // 拆分任务
        int middle = start + length / 2;
        ForkJoinSum leftTask = new ForkJoinSum(numbers, start, middle);
        ForkJoinSum rightTask = new ForkJoinSum(numbers, middle, end);

        // 异步执行左子任务
        leftTask.fork();
        // 当前线程直接执行右子任务（减少调度开销）
        long rightResult = rightTask.compute();
        // 等待左子任务结果并合并
        long leftResult = leftTask.join();

        return leftResult + rightResult;
    }

    public static void main(String[] args) {
        long[] numbers = new long[100_000_000];
        for (int i = 0; i < numbers.length; i++) {
            numbers[i] = i + 1;
        }

        ForkJoinPool pool = new ForkJoinPool();
        long startTime = System.currentTimeMillis();
        long result = pool.invoke(new ForkJoinSum(numbers, 0, numbers.length));
        long endTime = System.currentTimeMillis();
        System.out.println("ForkJoin 求和结果: " + result);
        System.out.println("耗时: " + (endTime - startTime) + " ms");
        pool.shutdown();
    }
}