Java 并发设计模式实战

选型思路：
  共享数据 → 优先不可变 / 线程封闭
  协作分工 → 生产者-消费者 / 工作窃取
  读多写少 → 读写锁 / Copy-on-Write
  延迟创建 → 双重检查锁定 / Holder 类

// JUC 落地：用 BlockingQueue 免手写同步
BlockingQueue<Order> queue = new ArrayBlockingQueue<>(1000);

// 生产者：提交订单
Runnable producer = () -> {
    try { queue.put(new Order("A-1")); }
    catch (InterruptedException e) { Thread.currentThread().interrupt(); }
};

// 消费者：处理订单
Runnable consumer = () -> {
    try {
        Order order = queue.take();   // 队列空则阻塞等待
        process(order);
    } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
};

生产关键点：
  有界队列 + 拒绝策略，防止生产者无限堆积
  消费者失败要有重试/死信队列，避免丢失
  用 offer/poll 配合超时，而不是裸 put/take 死等

ForkJoinPool 的工作窃取：
  1. 大任务 fork 成子任务，放入自己队列
  2. 线程空闲时从别人队列 steal 任务
  3. 任务合并用 join，形成递归分治

对比普通线程池：
  固定队列 → 空闲线程干等，任务堆积某队列
  工作窃取 → 空闲线程主动找活，利用率高

// ForkJoinPool 示例：并行求和
class SumTask extends RecursiveTask<Long> {
    private final long[] arr; private final int lo, hi;
    static final int THRESHOLD = 1000;

    @Override
    protected Long compute() {
        if (hi - lo <= THRESHOLD) {
            long sum = 0;
            for (int i = lo; i < hi; i++) sum += arr[i];
            return sum;
        }
        int mid = (lo + hi) >>> 1;
        SumTask left  = new SumTask(arr, lo, mid);
        SumTask right = new SumTask(arr, mid, hi);
        left.fork();            // 压入队列
        return right.compute() + left.join();  // 合并
    }
}

陷阱：
  1. 任务太小 → fork 开销超过计算，设阈值
  2. 共享可变状态 → 子任务互相踩踏，要保持子任务独立
  3. join 顺序 → 先 fork 后 join，避免栈深爆炸

并行流 parallelStream() 底层就是 ForkJoinPool.commonPool

// 不可变对象：final 字段 + 无 setter
public final class Point {
    private final int x;
    private final int y;
    public Point(int x, int y) { this.x = x; this.y = y; }
    public int x() { return x; }
    public int y() { return y; }
    // 更新返回新对象，而不是修改自身
    public Point move(int dx, int dy) {
        return new Point(x + dx, y + dy);
    }
}

实践组合：
  更新频繁的少量字段 → 可变 + 锁
  低频但被广泛共享 → 不可变，无锁
  大量只读引用 → 不可变 + Copy-on-Write 容器

ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();
Map<String, Config> configs = new HashMap<>();

// 读路径：多个读线程可并发
Config get(String key) {
    rwLock.readLock().lock();
    try { return configs.get(key); }
    finally { rwLock.readLock().unlock(); }
}

// 写路径：独占，写时读也阻塞
void put(String key, Config c) {
    rwLock.writeLock().lock();
    try { configs.put(key, c); }
    finally { rwLock.writeLock().unlock(); }
}

// StampedLock 乐观读：读时不加锁，validate 校验版本
StampedLock lock = new StampedLock();
long stamp = lock.tryOptimisticRead();
int v = value;                        // 乐观读
if (!lock.validate(stamp)) {          // 期间被写过 → 升级悲观读
    stamp = lock.readLock();
    try { v = value; }
    finally { lock.unlockRead(stamp); }
}

// 错误示范：不安全，可能读到半初始化对象
private static Singleton instance;
public static Singleton getInstance() {
    if (instance == null) {            // 第一次检查
        synchronized (Singleton.class) {
            if (instance == null) {    // 第二次检查
                instance = new Singleton();
            }
        }
    }
    return instance;
}

// 正确：volatile 禁止指令重排，保证初始化可见
private static volatile Singleton instance;

public static Singleton getInstance() {
    if (instance == null) {
        synchronized (Singleton.class) {
            if (instance == null) {
                instance = new Singleton();
            }
        }
    }
    return instance;
}

// 推荐：利用类加载机制保证线程安全，无需锁与 volatile
public class Singleton {
    private Singleton() {}
    private static class Holder {
        static final Singleton INSTANCE = new Singleton();
    }
    public static Singleton getInstance() {
        return Holder.INSTANCE;
    }
}

// 事件监听器列表：遍历时读者拿到稳定快照
CopyOnWriteArrayList<Listener> listeners = new CopyOnWriteArrayList<>();

listeners.add(listener);            // 写：复制数组，成本 O(n)
for (Listener l : listeners) {      // 读：无锁，拿的是快照
    l.onEvent(event);
}

// 用 CompletableFuture 做"异步承诺"
CompletableFuture<Price> priceFuture =
        CompletableFuture.supplyAsync(() -> queryPrice("BTC"))
                .orTimeout(3, TimeUnit.SECONDS);   // 超时即失败

priceFuture.thenAccept(p -> System.out.println("价格=" + p));