JUC 并发工具

一、为什么需要 JUC？

Java 早期的并发控制只有 synchronized + wait() + notify()。这种方案虽然简单，但存在严重局限：

无法尝试获取锁：线程要么获取到锁，要么无限阻塞，无法"尝试一下，不行就走"
无法中断等待：等待锁的线程无法被中断，只能一直等
无法实现公平锁：synchronized 内部锁竞争是非公平的，可能导致线程饥饿
无法实现读写分离：读操作不会修改数据，本应可以并发，但 synchronized 强制串行
生产者-消费者困难：wait/notify 使用不当容易导致死锁或虚假唤醒
无法实现复杂同步模式：如"等待多个任务完成"、"控制同时访问的线程数"

JUC（java.util.concurrent）从 JDK 5 开始引入，由 Doug Lea 设计，提供了比内置锁更丰富、更灵活、更高性能的并发工具。

核心价值对比：

需求	synchronized 方案	JUC 方案
尝试获取锁	不支持	`tryLock()`
可中断的锁等待	不支持	`lockInterruptibly()`
公平锁	不支持	`new ReentrantLock(true)`
读写分离	不支持	`ReentrantReadWriteLock`
等待多个任务完成	`wait/notify` 复杂实现	`CountDownLatch`
控制并发数	`Semaphore` 手动实现	`Semaphore`
生产者-消费者	`wait/notify` 易出错	`BlockingQueue`
多条件等待	一个锁一个条件	一个锁多个 `Condition`

二、Lock 接口与 ReentrantLock

2.1 Lock 接口

Lock 是 JUC 提供的显式锁接口，比 synchronized 更灵活。

java

public interface Lock {
    void lock();                          // 获取锁，阻塞直到成功
    void lockInterruptibly() throws InterruptedException;  // 可中断的获取锁
    boolean tryLock();                    // 尝试获取锁，立即返回
    boolean tryLock(long time, TimeUnit unit) throws InterruptedException;  // 超时尝试
    void unlock();                        // 释放锁
    Condition newCondition();             // 创建条件变量
}

2.2 ReentrantLock —— 可重入锁

可重入：同一线程可以多次获取同一把锁，不会自己把自己锁死。内部维护一个计数器，每次 lock 计数器 +1，每次 unlock 计数器 -1，计数器归零时锁被释放。

java

class ReentrantLockDemo {
    private final ReentrantLock lock = new ReentrantLock();
    private int count = 0;

    public void increment() {
        lock.lock();
        try {
            count++;
            // 可重入：同一线程调用另一个加锁方法
            doSomething();
        } finally {
            lock.unlock();  // 必须在 finally 中释放
        }
    }

    public void doSomething() {
        lock.lock();  // 同一线程再次获取同一把锁，计数器 +1
        try {
            System.out.println("可重入锁：同一线程可以再次获取");
        } finally {
            lock.unlock();
        }
    }
}

2.3 公平锁 vs 非公平锁

java

// 非公平锁（默认）：新来的线程可能"插队"抢到锁，吞吐量高，但可能线程饥饿
ReentrantLock unfairLock = new ReentrantLock();        // 或 new ReentrantLock(false)

// 公平锁：按等待时间顺序获取锁，FIFO，公平但性能较低
ReentrantLock fairLock = new ReentrantLock(true);

维度	公平锁	非公平锁
获取顺序	先到先得（FIFO）	可插队
线程饥饿	不会	可能导致
吞吐量	较低	较高（减少上下文切换）
适用场景	需要公平性的场景	追求高吞吐量

2.4 tryLock —— 尝试获取锁

java

public class TryLockDemo {
    private final Lock lock = new ReentrantLock();

    public boolean transfer(Account from, Account to, int amount) {
        // 尝试获取两把锁，超时则放弃
        if (from.lock.tryLock()) {
            try {
                if (to.lock.tryLock(1, TimeUnit.SECONDS)) {
                    try {
                        if (from.getBalance() >= amount) {
                            from.debit(amount);
                            to.credit(amount);
                            return true;
                        }
                    } finally {
                        to.lock.unlock();
                    }
                }
            } finally {
                from.lock.unlock();
            }
        }
        return false;
    }
}

2.5 lockInterruptibly —— 可中断的锁等待

java

public class InterruptibleLockDemo {
    private final Lock lock = new ReentrantLock();

    public void doWork() {
        try {
            lock.lockInterruptibly();  // 等待锁时可被中断
            try {
                // 长时间操作
                Thread.sleep(10000);
            } finally {
                lock.unlock();
            }
        } catch (InterruptedException e) {
            // 线程被中断，不再等待锁
            Thread.currentThread().interrupt();
            System.out.println("锁等待被中断");
        }
    }
}

2.6 Condition —— 条件变量

synchronized 中一个锁只有一个等待队列（通过 wait/notify），Lock 可以通过 Condition 创建多个等待队列，实现更精细的线程调度。

java

public class BoundedBuffer {
    private final Lock lock = new ReentrantLock();
    private final Condition notFull = lock.newCondition();   // 队列不满条件
    private final Condition notEmpty = lock.newCondition();  // 队列不空条件
    private final Object[] items = new Object[100];
    private int putIndex, takeIndex, count;

    public void put(Object item) throws InterruptedException {
        lock.lock();
        try {
            while (count == items.length) {
                notFull.await();  // 队列满，等待 notFull 信号
            }
            items[putIndex] = item;
            if (++putIndex == items.length) putIndex = 0;
            count++;
            notEmpty.signal();  // 通知消费者：队列不空了
        } finally {
            lock.unlock();
        }
    }

    public Object take() throws InterruptedException {
        lock.lock();
        try {
            while (count == 0) {
                notEmpty.await();  // 队列空，等待 notEmpty 信号
            }
            Object item = items[takeIndex];
            if (++takeIndex == items.length) takeIndex = 0;
            count--;
            notFull.signal();  // 通知生产者：队列不满了
            return item;
        } finally {
            lock.unlock();
        }
    }
}

2.7 synchronized vs ReentrantLock 对比

维度	synchronized	ReentrantLock
实现层面	JVM 关键字，C++ 实现	Java API，纯 Java 实现
锁释放	自动（代码块退出或异常）	手动（必须 finally 中 unlock）
可中断	不可中断	支持 `lockInterruptibly()`
公平锁	不支持	支持 `new ReentrantLock(true)`
尝试获取	不支持	支持 `tryLock()`
条件变量	一个锁一个条件	一个锁多个 `Condition`
性能（JDK 6+）	优化后相当	优化后相当
何时使用	一般场景，简单可靠	需要高级特性时

三、ReadWriteLock —— 读写锁

3.1 为什么需要读写锁？

大多数并发场景中，读操作远多于写操作。synchronized 和 ReentrantLock 强制所有操作串行，读-读之间本不需要互斥。读写锁通过读写分离提升并发度：

读锁共享：多个线程可以同时持有读锁，只要没有写锁
写锁独占：写锁是独占的，读写互斥、写写互斥

java

public class ReadWriteCache {
    private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
    private final Lock readLock = rwLock.readLock();
    private final Lock writeLock = rwLock.writeLock();
    private final Map<String, Object> cache = new HashMap<>();

    // 读操作：使用读锁，多个线程可并发读
    public Object get(String key) {
        readLock.lock();
        try {
            return cache.get(key);
        } finally {
            readLock.unlock();
        }
    }

    // 写操作：使用写锁，独占访问
    public void put(String key, Object value) {
        writeLock.lock();
        try {
            cache.put(key, value);
        } finally {
            writeLock.unlock();
        }
    }
}

3.2 锁降级

写锁可以降级为读锁（先获取写锁，再获取读锁，然后释放写锁），但读锁不能升级为写锁。

java

public class LockDowngradeDemo {
    private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
    private final Lock readLock = rwLock.readLock();
    private final Lock writeLock = rwLock.writeLock();
    private volatile boolean cacheValid = false;
    private Object data;

    public Object process() {
        readLock.lock();
        if (cacheValid) {
            try {
                return data;
            } finally {
                readLock.unlock();
            }
        }
        readLock.unlock();

        // 缓存失效，需要更新
        writeLock.lock();
        try {
            if (!cacheValid) {  // 双重检查
                data = computeNewData();  // 耗时的计算
                cacheValid = true;
            }
            // 锁降级：获取读锁后再释放写锁
            readLock.lock();
        } finally {
            writeLock.unlock();
        }

        try {
            return data;  // 持有读锁返回数据
        } finally {
            readLock.unlock();
        }
    }

    private Object computeNewData() {
        return new Object();
    }
}

四、AQS —— JUC 的基石

4.1 AQS 是什么

AQS（AbstractQueuedSynchronizer）是 JUC 包最核心的框架，ReentrantLock、Semaphore、CountDownLatch、ReentrantReadWriteLock、CyclicBarrier 等绝大多数同步工具都是基于 AQS 实现的。

4.2 核心思想

AQS 维护两个核心组件：

volatile int state：同步状态，子类通过操作 state 实现不同的同步语义
FIFO 等待队列（CLH 队列变体）：存放获取锁失败的线程

AQS 核心结构：
┌─────────────────────────────────────────────────┐
│  volatile int state     ← 同步状态              │
│  volatile Node head     ← CLH 队列头            │
│  volatile Node tail     ← CLH 队列尾            │
├─────────────────────────────────────────────────┤
│  模板方法（子类继承实现）：                      │
│    tryAcquire(int)       → 独占模式获取锁        │
│    tryRelease(int)       → 独占模式释放锁        │
│    tryAcquireShared(int) → 共享模式获取锁        │
│    tryReleaseShared(int) → 共享模式释放锁        │
│    isHeldExclusively()   → 是否独占              │
└─────────────────────────────────────────────────┘

4.3 CLH 队列

CLH 队列是一个变体的 CLH 锁队列，每个线程封装为一个 Node 节点：

java

// Node 结构（简化）
static final class Node {
    volatile Node prev;        // 前驱节点
    volatile Node next;        // 后继节点
    volatile Thread thread;    // 等待的线程
    volatile int waitStatus;   // 等待状态：SIGNAL(-1), CANCELLED(1), CONDITION(-2), PROPAGATE(-3)
    Node nextWaiter;           // 条件队列中的下一个节点
}

4.4 获取锁的流程（acquire）

线程调用 acquire(1)
    │
    ├─ tryAcquire(1) 成功？
    │   ├─ 是 → 获取锁，线程继续执行
    │   └─ 否 → 进入 acquireQueued
    │            │
    │            ├─ addWaiter → 将当前线程包装成 Node 加入 CLH 队列尾部
    │            │
    │            └─ acquireQueued → 自旋 + 阻塞
    │                 │
    │                 └─ 前驱是 head 且 tryAcquire 成功？
    │                      ├─ 是 → 当前节点成为新的 head，获取锁
    │                      └─ 否 → shouldParkAfterFailedAcquire
    │                               └─ 前驱节点 SIGNAL 状态？
    │                                    ├─ 是 → parkAndCheckInterrupt（阻塞）
    │                                    └─ 否 → 将前驱设为 SIGNAL，继续自旋

4.5 不同同步器的 state 语义

同步器	state 含义	获取/释放
`ReentrantLock`	0=未锁定，>0=重入次数	独占模式
`Semaphore`	剩余许可数	共享模式
`CountDownLatch`	还需要等待的计数	共享模式
`ReentrantReadWriteLock`	高 16 位=读锁次数，低 16 位=写锁重入次数	读写混合

java

// 自定义一个简单的互斥锁（基于 AQS）
class SimpleLock {
    private final Sync sync = new Sync();

    private static class Sync extends AbstractQueuedSynchronizer {
        @Override
        protected boolean tryAcquire(int arg) {
            if (compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }

        @Override
        protected boolean tryRelease(int arg) {
            if (getState() == 0) throw new IllegalMonitorStateException();
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }

        @Override
        protected boolean isHeldExclusively() {
            return getState() == 1;
        }
    }

    public void lock()   { sync.acquire(1); }
    public void unlock() { sync.release(1); }
    public boolean isLocked() { return sync.isHeldExclusively(); }
}

五、CountDownLatch —— 倒数门闩

5.1 概念

CountDownLatch 是一个一次性的同步辅助工具：一个线程（或多个线程）等待其他线程完成一组操作后，才能继续执行。像一个倒计时器，计数归零时门闩打开。

初始化：new CountDownLatch(3)  → 计数器 = 3
线程A: latch.await()  → 阻塞等待计数器归零
线程B: latch.countDown()  → 计数器 = 2
线程C: latch.countDown()  → 计数器 = 1
线程D: latch.countDown()  → 计数器 = 0  → 线程A被唤醒，继续执行

5.2 典型场景

java

// 场景1：等待多个子任务完成后，主线程汇总结果
public class TaskCoordinator {
    public void execute() throws InterruptedException {
        int taskCount = 5;
        CountDownLatch latch = new CountDownLatch(taskCount);
        ExecutorService executor = Executors.newFixedThreadPool(taskCount);

        for (int i = 0; i < taskCount; i++) {
            final int taskId = i;
            executor.submit(() -> {
                try {
                    doTask(taskId);
                } finally {
                    latch.countDown();  // 每个任务完成，计数-1
                }
            });
        }

        latch.await();  // 主线程等待所有任务完成
        System.out.println("所有任务完成，汇总结果");
        executor.shutdown();
    }

    private void doTask(int id) {
        System.out.println("Task " + id + " 开始执行");
        // 模拟耗时操作
        try { Thread.sleep(ThreadLocalRandom.current().nextInt(1000)); } catch (InterruptedException e) {}
        System.out.println("Task " + id + " 完成");
    }
}

java

// 场景2：模拟并发——所有线程同时起跑
public class RacingDemo {
    public void race() throws InterruptedException {
        int runnerCount = 8;
        CountDownLatch readyLatch = new CountDownLatch(runnerCount);
        CountDownLatch startLatch = new CountDownLatch(1);

        for (int i = 0; i < runnerCount; i++) {
            final int runnerId = i;
            new Thread(() -> {
                try {
                    System.out.println("选手 " + runnerId + " 准备就绪");
                    readyLatch.countDown();  // 通知准备完毕
                    startLatch.await();       // 等待发令枪
                    System.out.println("选手 " + runnerId + " 起跑！");
                } catch (InterruptedException e) {}
            }).start();
        }

        readyLatch.await();  // 等待所有选手准备完毕
        System.out.println("====== 各就各位，预备... ======");
        Thread.sleep(1000);   // 模拟发令枪准备
        startLatch.countDown();  // 发令枪！所有选手同时起跑
        System.out.println("====== 砰！ ======");
    }
}

5.3 CountDownLatch vs join

维度	CountDownLatch	Thread.join()
灵活性	线程结束时 countDown，不阻塞	必须等待线程终止
线程池	支持（线程池中线程不终止）	不支持（线程池线程不会终止）
计数	可多次 countDown	只能等线程结束
重用	不可重用（计数器不能重置）	N/A

六、CyclicBarrier —— 循环栅栏

6.1 概念

CyclicBarrier 让一组线程互相等待，直到所有线程都到达屏障点，然后一起继续执行。与 CountDownLatch 的关键区别：CyclicBarrier 可以重复使用，计数器归零后自动重置。

初始化：new CyclicBarrier(3)  → 阈值 = 3
线程A: barrier.await()  → 到达屏障，等待
线程B: barrier.await()  → 到达屏障，等待
线程C: barrier.await()  → 到达屏障，阈值到达 → 所有线程同时释放，继续执行
                        → 计数器自动重置为 3，可重复使用

6.2 典型场景

java

// 场景：多线程分批计算，每批计算完成后汇总
public class CyclicBarrierDemo {
    public void compute() {
        int threadCount = 4;
        CyclicBarrier barrier = new CyclicBarrier(threadCount, () -> {
            // 当所有线程到达屏障时，由最后一个到达的线程执行此回调
            System.out.println("===== 本批计算完成，开始汇总 =====");
        });

        for (int i = 0; i < threadCount; i++) {
            final int threadId = i;
            new Thread(() -> {
                for (int batch = 0; batch < 3; batch++) {
                    try {
                        System.out.println("线程 " + threadId + " 执行第 " + batch + " 批计算");
                        Thread.sleep(ThreadLocalRandom.current().nextInt(1000));
                        barrier.await();  // 等待其他线程完成本批
                    } catch (InterruptedException | BrokenBarrierException e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
    }
}

6.3 CountDownLatch vs CyclicBarrier

维度	CountDownLatch	CyclicBarrier
可重用	不可重用（一次性）	可重用（计数器自动重置）
角色	一个/多个线程等待其他线程	多个线程互相等待
回调	无	支持 barrierAction（最后一个到达的线程执行）
典型场景	主线程等待子任务完成	多线程分阶段协作
异常处理	无	线程中断/超时会触发 BrokenBarrierException

七、Semaphore —— 信号量

7.1 概念

Semaphore 控制同时访问某个资源的线程数量。就像一个停车场，只有 N 个车位，满了就得等。

java

// 初始化 3 个许可（最多 3 个线程同时访问资源）
Semaphore semaphore = new Semaphore(3);

// 公平模式：按等待顺序获取许可
Semaphore fairSemaphore = new Semaphore(3, true);

7.2 典型场景

java

// 数据库连接池：限制同时连接数
public class ConnectionPool {
    private final Semaphore semaphore;
    private final List<Connection> connections = new ArrayList<>();

    public ConnectionPool(int size) {
        semaphore = new Semaphore(size);
        for (int i = 0; i < size; i++) {
            connections.add(createConnection());
        }
    }

    public Connection getConnection() throws InterruptedException {
        semaphore.acquire();  // 获取许可（阻塞直到有可用连接）
        return acquireConnection();
    }

    public void releaseConnection(Connection conn) {
        releaseConnection(conn);
        semaphore.release();  // 归还许可
    }

    // 尝试获取，超时放弃
    public Connection getConnection(long timeout, TimeUnit unit)
            throws InterruptedException, TimeoutException {
        if (semaphore.tryAcquire(timeout, unit)) {
            return acquireConnection();
        }
        throw new TimeoutException("获取连接超时");
    }

    private synchronized Connection acquireConnection() {
        return connections.remove(connections.size() - 1);
    }

    private synchronized void releaseConnection(Connection conn) {
        connections.add(conn);
    }

    private Connection createConnection() {
        // 创建实际数据库连接
        return null;
    }
}

java

// 限流器：控制 API 调用速率
public class RateLimiter {
    private final Semaphore semaphore;

    public RateLimiter(int permitsPerSecond) {
        this.semaphore = new Semaphore(permitsPerSecond);
        // 每秒补充许可（简单实现）
        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
        scheduler.scheduleAtFixedRate(() -> {
            semaphore.release(permitsPerSecond - semaphore.availablePermits());
        }, 1, 1, TimeUnit.SECONDS);
    }

    public boolean tryAcquire() {
        return semaphore.tryAcquire();
    }
}

八、Exchanger —— 交换器

Exchanger 用于两个线程之间交换数据。当两个线程都到达 exchange() 点时，彼此交换数据。

java

public class ExchangerDemo {
    public void demo() {
        Exchanger<String> exchanger = new Exchanger<>();

        new Thread(() -> {
            try {
                String data = "Thread-A 的数据";
                System.out.println("Thread-A 准备交换: " + data);
                String received = exchanger.exchange(data);
                System.out.println("Thread-A 收到: " + received);
            } catch (InterruptedException e) {}
        }, "Thread-A").start();

        new Thread(() -> {
            try {
                Thread.sleep(1000);  // 模拟处理延迟
                String data = "Thread-B 的数据";
                System.out.println("Thread-B 准备交换: " + data);
                String received = exchanger.exchange(data);
                System.out.println("Thread-B 收到: " + received);
            } catch (InterruptedException e) {}
        }, "Thread-B").start();
    }
}
// 输出：
// Thread-A 准备交换: Thread-A 的数据
// Thread-B 准备交换: Thread-B 的数据
// Thread-B 收到: Thread-A 的数据
// Thread-A 收到: Thread-B 的数据

九、BlockingQueue —— 阻塞队列

9.1 概念

BlockingQueue 是线程安全的队列，支持阻塞插入和阻塞移除。它是生产者-消费者模式的最佳实现。

java

// 入队操作
add(e)      // 队列满时抛异常
offer(e)    // 队列满时返回 false
put(e)      // 队列满时阻塞等待
offer(e, time, unit)  // 队列满时超时等待

// 出队操作
remove()    // 队列空时抛异常
poll()      // 队列空时返回 null
take()      // 队列空时阻塞等待
poll(time, unit)  // 队列空时超时等待

9.2 常用实现

实现	底层结构	特性	适用场景
`ArrayBlockingQueue`	数组	有界，FIFO，一把锁	固定容量场景
`LinkedBlockingQueue`	链表	可选有界（默认 Integer.MAX_VALUE），两把锁（put/take 分离）	高吞吐量
`PriorityBlockingQueue`	二叉堆	无界，按优先级出队	优先级调度
`DelayQueue`	PriorityQueue	延迟出队，元素需实现 Delayed 接口	定时任务
`SynchronousQueue`	无容量	不存储元素，put 和 take 必须配对	直接交付（CachedThreadPool）
`LinkedTransferQueue`	链表	无界，支持 transfer（阻塞直到被消费）	高性能交付

9.3 生产者-消费者模式

java

public class ProducerConsumerDemo {
    private static final int QUEUE_CAPACITY = 10;
    private static final BlockingQueue<String> queue = new ArrayBlockingQueue<>(QUEUE_CAPACITY);

    static class Producer implements Runnable {
        @Override
        public void run() {
            try {
                for (int i = 0; i < 20; i++) {
                    String item = "Product-" + i;
                    queue.put(item);  // 队列满时阻塞
                    System.out.println("生产: " + item + " | 队列大小: " + queue.size());
                    Thread.sleep(ThreadLocalRandom.current().nextInt(500));
                }
                queue.put("STOP");  // 毒丸：终止信号
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    static class Consumer implements Runnable {
        @Override
        public void run() {
            try {
                while (true) {
                    String item = queue.take();  // 队列空时阻塞
                    if ("STOP".equals(item)) {
                        System.out.println("消费者收到终止信号，退出");
                        break;
                    }
                    System.out.println("消费: " + item + " | 队列大小: " + queue.size());
                    Thread.sleep(ThreadLocalRandom.current().nextInt(1000));
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    public static void main(String[] args) {
        new Thread(new Producer()).start();
        new Thread(new Consumer()).start();
    }
}

十、ConcurrentHashMap 深度解析

10.1 为什么需要 ConcurrentHashMap？

HashMap 线程不安全，多线程 put 可能导致死循环（JDK 7）或数据丢失。HashTable 使用 synchronized 锁住整个表，并发度极低。ConcurrentHashMap 通过分段锁（JDK 7）和 CAS + synchronized（JDK 8）实现高并发。

10.2 JDK 7：分段锁（Segment）

ConcurrentHashMap (JDK 7)
┌──────────────────────────────────────────────────┐
│  Segment[0]  →  HashEntry[] → 链表               │
│  Segment[1]  →  HashEntry[] → 链表               │
│  Segment[2]  →  HashEntry[] → 链表               │
│  ...                                             │
│  Segment[15] →  HashEntry[] → 链表               │
└──────────────────────────────────────────────────┘
默认 concurrentLevel = 16，即 16 个 Segment
每个 Segment 继承 ReentrantLock，独立加锁

核心思想：将整个哈希表分成多个 Segment，每个 Segment 独立加锁。不同 Segment 之间可以并发操作。并发度 = Segment 数量（默认 16）。

JDK 7 缺点： Segment 数量固定，不能动态扩容；哈希冲突严重时链表过长，查询 O(n)；跨 Segment 操作（如 size()）需要获取所有 Segment 的锁。

10.3 JDK 8：CAS + synchronized

JDK 8 完全重写 ConcurrentHashMap，放弃分段锁，改用 CAS + synchronized：

ConcurrentHashMap (JDK 8)
┌──────────────────────────────────────────────────┐
│  Node[] table     ← 数组（同 HashMap）            │
│  ├── [0] → Node → Node → TreeNode（红黑树）      │
│  ├── [1] → null                                  │
│  ├── [2] → Node（头节点）                        │
│  ├── [3] → ForwardingNode（扩容中）               │
│  └── ...                                         │
└──────────────────────────────────────────────────┘

核心机制：

CAS 初始化 table：多个线程并发初始化时，只有一个 CAS 成功
CAS 写入空桶：如果桶为空，通过 casTabAt 直接写入
synchronized 锁住桶头节点：如果桶非空，锁住头节点，粒度到单个桶
红黑树：链表长度 >= 8 且 table.length >= 64 时树化
并发扩容：多线程协助扩容，每个线程分配一段迁移区间

java

// JDK 8 putVal 核心逻辑（简化）
final V putVal(K key, V value, boolean onlyIfAbsent) {
    // 1. 计算 hash
    int hash = spread(key.hashCode());
    // 2. 循环尝试
    for (Node<K,V>[] tab = table;;) {
        Node<K,V> f;
        // 3. 桶为空：CAS 写入
        if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
            if (casTabAt(tab, i, null, new Node<>(hash, key, value)))
                break;  // CAS 成功，写入完成
        }
        // 4. 正在扩容：协助迁移
        else if ((fh = f.hash) == MOVED)
            tab = helpTransfer(tab, f);
        else {
            // 5. 桶非空：synchronized 锁住头节点
            synchronized (f) {
                if (tabAt(tab, i) == f) {
                    // 遍历链表或红黑树，插入或更新
                }
            }
        }
    }
}

10.4 JDK 7 vs JDK 8 ConcurrentHashMap

维度	JDK 7	JDK 8
数据结构	Segment 数组 + HashEntry 链表	Node 数组 + 链表/红黑树
锁粒度	Segment 级别（默认 16 个）	桶级别（每个桶独立）
并发度	固定（Segment 数量）	动态（桶数量）
查找复杂度	O(n) 链表	O(n) 链表，O(log n) 红黑树
扩容	Segment 内部扩容	多线程协助扩容
size()	多次尝试，失败则锁所有 Segment	通过 baseCount + CounterCell 累加

10.5 ConcurrentHashMap 的 size() 实现

java

// JDK 8 通过 baseCount + CounterCell 数组实现高并发计数
// 相当于一个额外的 LongAdder
// 避免了 JDK 7 中 size() 需要锁所有 Segment 的问题
// 每个线程在自己的 CounterCell 中累加，减少竞争

十一、CopyOnWriteArrayList —— 写时复制

11.1 概念

CopyOnWriteArrayList 是线程安全的 ArrayList，读操作不加锁，写操作加锁并复制整个数组。适用于读多写少的场景。

java

// 写操作：ReentrantLock + Arrays.copyOf
public boolean add(E e) {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        Object[] elements = getArray();
        int len = elements.length;
        // 复制一份新数组，长度+1，追加新元素
        Object[] newElements = Arrays.copyOf(elements, len + 1);
        newElements[len] = e;
        setArray(newElements);
        return true;
    } finally {
        lock.unlock();
    }
}

// 读操作：无锁，直接读取当前数组
public E get(int index) {
    return get(getArray(), index);
}

11.2 适用场景 vs 不适用场景

适用场景	不适用场景
读多写少（读 >> 写）	写操作频繁
数据量小	数据量大（复制开销大）
不要求实时一致性	要求实时一致性（读的是快照，可能不是最新数据）
迭代时不需要锁	迭代时修改频繁

11.3 核心问题

迭代器是**快照（snapshot）**式的：迭代器创建时复制一份数组快照，后续对 CopyOnWriteArrayList 的修改不影响当前迭代器。这避免了 ConcurrentModificationException，但代价是迭代器可能读到过期数据。

java

CopyOnWriteArrayList<String> list = new CopyOnWriteArrayList<>();
list.add("A");
list.add("B");

Iterator<String> it = list.iterator();
list.add("C");  // 写操作，复制新数组

// 迭代器仍然遍历旧的快照：A, B（不包含 C）
while (it.hasNext()) {
    System.out.println(it.next());  // 输出: A, B
}

十二、ForkJoinPool —— 分而治之

12.1 概念

ForkJoinPool 是 JDK 7 引入的线程池，专为分治任务设计。核心思想：将大任务递归拆分为小任务，并行执行，最后合并结果。

                    [大任务]
                   /       \
            [子任务1]     [子任务2]
            /     \       /     \
        [小任务] [小任务] [小任务] [小任务]
          ↓        ↓        ↓        ↓
        [结果1] [结果2] [结果3] [结果4]
          \        \       /        /
            [合并结果]   [合并结果]
                   \       /
                  [最终结果]

12.2 工作窃取（Work-Stealing）

每个工作线程有自己的双端队列（deque）。线程从自己的队列头部取任务，当自己队列为空时，从其他线程的队列尾部偷取任务。这避免了线程空闲，提高了 CPU 利用率。

线程A: [Task1] [Task2] [Task3] ← 自己从头部取
线程B: [Task4] [Task5]          ← 当前线程

线程C 空闲 → 从线程A的尾部偷 Task3
线程D 空闲 → 从线程B的尾部偷 Task5

12.3 RecursiveTask 示例

java

// 计算 1 到 N 的累加和（分治版）
public class ForkJoinSum extends RecursiveTask<Long> {
    private static final int THRESHOLD = 10_000;  // 阈值：小于此值不再拆分
    private final long start;
    private final long end;

    public ForkJoinSum(long start, long end) {
        this.start = start;
        this.end = end;
    }

    @Override
    protected Long compute() {
        long length = end - start + 1;
        // 任务足够小，直接计算
        if (length <= THRESHOLD) {
            long sum = 0;
            for (long i = start; i <= end; i++) {
                sum += i;
            }
            return sum;
        }
        // 任务太大，一分为二
        long mid = (start + end) / 2;
        ForkJoinSum left = new ForkJoinSum(start, mid);
        ForkJoinSum right = new ForkJoinSum(mid + 1, end);

        left.fork();   // 异步执行左半部分
        right.fork();  // 异步执行右半部分

        return left.join() + right.join();  // 等待结果并合并
    }

    public static void main(String[] args) {
        ForkJoinPool pool = new ForkJoinPool();
        ForkJoinSum task = new ForkJoinSum(1, 1_000_000_000L);
        long result = pool.invoke(task);
        System.out.println("Sum: " + result);  // 500000000500000000
    }
}

12.4 ForkJoinPool 与 Parallel Stream 的关系

Java 8 并行流的底层就是 ForkJoinPool：

java

// 以下两段代码底层使用相同的 ForkJoinPool.commonPool()
list.parallelStream().mapToInt(Integer::intValue).sum();

// 等价于手动使用 ForkJoinPool
ForkJoinPool.commonPool().submit(() ->
    list.parallelStream().mapToInt(Integer::intValue).sum()
);

十三、JUC 工具速查表

工具	核心作用	类比	是否可重用
`ReentrantLock`	显式锁，替代 synchronized	独占锁	是
`ReadWriteLock`	读写分离	共享读 + 独占写	是
`CountDownLatch`	等待一组操作完成	倒计时门闩	否
`CyclicBarrier`	线程互相等待	栅栏/发令枪	是
`Semaphore`	控制并发访问数	停车场车位	是
`Exchanger`	两线程交换数据	交换器	是
`BlockingQueue`	阻塞队列	生产者-消费者管道	是
`ConcurrentHashMap`	高并发 Map	线程安全 HashMap	是
`CopyOnWriteArrayList`	读无锁 List	快照式 List	是
`ForkJoinPool`	分治任务线程池	工作窃取	是

十四、面试要点

Q1: synchronized 和 ReentrantLock 的区别？什么时候用哪个？

维度	synchronized	ReentrantLock
实现	JVM 关键字	Java API
锁释放	自动	手动（finally unlock）
可中断	不支持	支持 lockInterruptibly
公平锁	不支持	支持
尝试获取	不支持	支持 tryLock
条件变量	一个	多个 Condition

选择原则：一般场景用 synchronized（简单、自动释放、不易出错），需要高级特性（可中断、超时、公平锁、多条件）时用 ReentrantLock。

Q2: AQS 是什么？它是如何实现锁的？

AQS 是 JUC 的核心框架，通过维护一个 volatile int state（同步状态）和一个 FIFO 的 CLH 等待队列实现锁。子类只需实现 tryAcquire/tryRelease 定义获取/释放逻辑，AQS 负责排队、阻塞、唤醒等通用逻辑。ReentrantLock、Semaphore、CountDownLatch 等都是基于 AQS 实现的。

Q3: CountDownLatch 和 CyclicBarrier 的区别？

CountDownLatch 是一次性的，一个或多个线程等待其他线程的 countDown 事件，计数器归零后不能重置。CyclicBarrier 是可重复使用的，一组线程互相等待直到所有线程到达屏障点，计数器归零后自动重置。典型场景：CountDownLatch 用于主线程等待子任务完成，CyclicBarrier 用于多线程分阶段协作。

Q4: ConcurrentHashMap JDK 7 和 JDK 8 的实现区别？

JDK 7 使用分段锁（Segment），每个 Segment 继承 ReentrantLock，Segment 数量固定（默认 16），并发度受限。JDK 8 放弃分段锁，改用 CAS + synchronized 锁住单个桶，桶级别粒度更细，并发度更高。JDK 8 还引入了红黑树（链表长度 >= 8 时树化），查询效率从 O(n) 提升到 O(log n)，并支持多线程协助扩容。

Q5: ConcurrentHashMap 的 get() 方法需要加锁吗？

不需要。因为 Node 的 val 和 next 字段都是 volatile 修饰的，保证了多线程间的可见性。JDK 8 中 get() 全程无锁，通过 tabAt 方法（内部使用 Unsafe.getObjectVolatile）读取 volatile 数组，确保读到最新值。

Q6: CopyOnWriteArrayList 的优缺点？适用场景？

优点：读操作无锁，性能极高；迭代器是快照，不会抛 ConcurrentModificationException。缺点：写操作需要复制整个数组，内存开销大（写时双倍内存），写操作加锁（ReentrantLock），不适合高频写入。适用场景：读多写少（如配置信息、白名单/黑名单），数据量较小。

Q7: BlockingQueue 中 put() 和 offer() 的区别？

put() 在队列满时阻塞等待，直到队列有空间。offer() 在队列满时立即返回 false，不会阻塞。offer(e, time, unit) 在队列满时阻塞等待指定时间，超时后返回 false。选择取决于业务需求：生产者不能丢弃数据时用 put()，可以丢弃或重试时用 offer()。

Q8: ForkJoinPool 的工作窃取算法是什么？

每个工作线程维护一个双端队列。线程从自己的队列头部取任务（LIFO），当自己队列为空时，从其他线程的队列尾部偷取任务（FIFO）。这种设计减少了线程间的竞争：自己取任务和偷取任务操作的是队列的不同端。工作窃取使得所有线程保持忙碌，最大化 CPU 利用率。

Q9: Semaphore 和 ReentrantLock 有什么区别？

ReentrantLock 是二值信号量（只有 0 和 1），一次只允许一个线程访问临界区。Semaphore 允许多个线程同时访问资源（可设置多个许可），适合流量控制、连接池等场景。ReentrantLock 是可重入的，Semaphore 不是可重入的（同一线程获取两次许可会消耗两个许可）。

Q10: 为什么 ConcurrentHashMap 的 key 和 value 不允许为 null？

HashMap 允许 null key 和 null value，但 ConcurrentHashMap 不允许。原因：在并发环境下，如果 get() 返回 null，无法区分是"key 不存在"还是"value 为 null"。这会导致歧义，因为 containsKey 和 get 之间可能有其他线程修改。HashTable 也是因为同样的原因不允许 null。如果确实需要表示"不存在"，使用特殊值或 Optional。ConcurrentHashMap 的歧义分析：HashMap 中可以通过 containsKey 二次确认，但 ConcurrentHashMap 中 containsKey 和 get 之间可能被其他线程修改，即使用 containsKey 也无法准确判断，因此直接禁止 null 是最安全的做法。

JUC 并发工具 ​

一、为什么需要 JUC？ ​

二、Lock 接口与 ReentrantLock ​

2.1 Lock 接口 ​

2.2 ReentrantLock —— 可重入锁 ​

2.3 公平锁 vs 非公平锁 ​

2.4 tryLock —— 尝试获取锁 ​

2.5 lockInterruptibly —— 可中断的锁等待 ​

2.6 Condition —— 条件变量 ​

2.7 synchronized vs ReentrantLock 对比 ​

三、ReadWriteLock —— 读写锁 ​

3.1 为什么需要读写锁？ ​

3.2 锁降级 ​

四、AQS —— JUC 的基石 ​

4.1 AQS 是什么 ​

4.2 核心思想 ​

4.3 CLH 队列 ​

4.4 获取锁的流程（acquire） ​

4.5 不同同步器的 state 语义 ​

五、CountDownLatch —— 倒数门闩 ​

5.1 概念 ​

5.2 典型场景 ​

5.3 CountDownLatch vs join ​

六、CyclicBarrier —— 循环栅栏 ​

6.1 概念 ​

6.2 典型场景 ​

6.3 CountDownLatch vs CyclicBarrier ​

七、Semaphore —— 信号量 ​

7.1 概念 ​

7.2 典型场景 ​

八、Exchanger —— 交换器 ​

九、BlockingQueue —— 阻塞队列 ​

9.1 概念 ​

9.2 常用实现 ​

9.3 生产者-消费者模式 ​

十、ConcurrentHashMap 深度解析 ​

10.1 为什么需要 ConcurrentHashMap？ ​

10.2 JDK 7：分段锁（Segment） ​

10.3 JDK 8：CAS + synchronized ​

10.4 JDK 7 vs JDK 8 ConcurrentHashMap ​

10.5 ConcurrentHashMap 的 size() 实现 ​

十一、CopyOnWriteArrayList —— 写时复制 ​

11.1 概念 ​

11.2 适用场景 vs 不适用场景 ​

11.3 核心问题 ​

十二、ForkJoinPool —— 分而治之 ​

12.1 概念 ​

12.2 工作窃取（Work-Stealing） ​

12.3 RecursiveTask 示例 ​

12.4 ForkJoinPool 与 Parallel Stream 的关系 ​

十三、JUC 工具速查表 ​

十四、面试要点 ​

JUC 并发工具

一、为什么需要 JUC？

二、Lock 接口与 ReentrantLock

2.1 Lock 接口

2.2 ReentrantLock —— 可重入锁

2.3 公平锁 vs 非公平锁

2.4 tryLock —— 尝试获取锁

2.5 lockInterruptibly —— 可中断的锁等待

2.6 Condition —— 条件变量

2.7 synchronized vs ReentrantLock 对比

三、ReadWriteLock —— 读写锁

3.1 为什么需要读写锁？

3.2 锁降级

四、AQS —— JUC 的基石

4.1 AQS 是什么

4.2 核心思想

4.3 CLH 队列

4.4 获取锁的流程（acquire）

4.5 不同同步器的 state 语义

五、CountDownLatch —— 倒数门闩

5.1 概念

5.2 典型场景

5.3 CountDownLatch vs join

六、CyclicBarrier —— 循环栅栏

6.1 概念

6.2 典型场景

6.3 CountDownLatch vs CyclicBarrier

七、Semaphore —— 信号量

7.1 概念

7.2 典型场景

八、Exchanger —— 交换器

九、BlockingQueue —— 阻塞队列

9.1 概念

9.2 常用实现

9.3 生产者-消费者模式

十、ConcurrentHashMap 深度解析

10.1 为什么需要 ConcurrentHashMap？

10.2 JDK 7：分段锁（Segment）

10.3 JDK 8：CAS + synchronized

10.4 JDK 7 vs JDK 8 ConcurrentHashMap

10.5 ConcurrentHashMap 的 size() 实现

十一、CopyOnWriteArrayList —— 写时复制

11.1 概念

11.2 适用场景 vs 不适用场景

11.3 核心问题

十二、ForkJoinPool —— 分而治之

12.1 概念

12.2 工作窃取（Work-Stealing）

12.3 RecursiveTask 示例

12.4 ForkJoinPool 与 Parallel Stream 的关系

十三、JUC 工具速查表

十四、面试要点