# Java 进阶并发框架

# 1. Java 并发理论基础

并发编程的目的是让程序运行得更快，但面临着上下文切换、死锁、以及受限于硬件和软件资源的制约。

核心挑战：

原子性：一个或多个操作要么全部执行且不被中断，要么都不执行。synchronized和Lock可以保证原子性。
可见性：一个线程修改了共享变量的值，其他线程能够立即看到。volatile、synchronized和Lock可以保证可见性。
有序性：程序执行的顺序按照代码的先后顺序执行。volatile和synchronized可以保证有序性(通过禁止指令重排序)。

Java内存模型(JMM)：定义了线程和主内存之间的抽象关系，是理解可见性和有序性的关键。

主内存(Main Memory)：所有共享变量都存储在主内存中。
工作内存(Working Memory)：每个线程有自己的工作内存，保存了该线程使用到的变量的主内存副本。
线程对变量的所有操作(读、写)都必须在工作内存中进行，不能直接读写主内存中的变量。
线程间变量值的传递均需要通过主内存来完成。

文字描述的图形展示 - Java内存模型交互：

初始状态：
主内存： sharedVariable = 0

线程A的工作内存： sharedVariable副本 = 0
线程B的工作内存： sharedVariable副本 = 0

步骤1： 线程A修改 sharedVariable = 1
线程A的工作内存： sharedVariable副本 = 1
主内存： sharedVariable = 0 (此时未刷新)

步骤2： 线程A将修改刷新到主内存
主内存： sharedVariable = 1

步骤3： 线程B从主内存读取最新的 sharedVariable
线程B的工作内存： sharedVariable副本 = 1

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

volatile关键字强制读写都直接作用于主内存，使得变量的修改对所有线程立即可见。

# 2. 关键字详解

# 2.1. volatile

volatile是轻量级的同步机制。它保证了可见性和禁止指令重排序，但不保证原子性。

设计思想：

内存屏障(Memory Barrier)：在写入volatile变量时，JVM会在写操作后插入一个StoreStore屏障和StoreLoad屏障，确保写入立即刷入主内存，并防止与后续操作重排序。
嗅探(Bus Sniffing)：当其他处理器的工作内存中的volatile变量副本因所在线程的操作失效时，会强制从主内存重新读取。

适用场景：状态标志位(如 boolean flag)、DCL单例模式(Double-Checked Locking)。

代码示例 - 状态标志：

public class VolatileExample {
    private volatile boolean flag = false;

    public void writer() {
        flag = true; // 写 volatile 变量
    }

    public void reader() {
        if (flag) { // 读 volatile 变量
            System.out.println("Flag is true");
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13

# 2.2. synchronized

synchronized是Java内置的重量级锁，可以修饰方法或代码块。它保证了同一时刻，只有一个线程可以执行某个方法或代码块，从而保证了原子性、可见性和有序性。

锁对象：

修饰实例方法：锁是当前实例对象(this)。
修饰静态方法：锁是当前类的Class对象。
修饰代码块：锁是synchronized括号里配置的对象。

文字描述的图形展示 - synchronized 方法执行流程：

线程1请求锁 ---> 对象锁状态：无锁 (01) ---> 线程1成功获取锁，锁状态：偏向锁/轻量级锁/重量级锁
      线程1执行同步块...
线程2请求锁 ---> 对象锁状态：已被线程1持有 ---> 线程2进入同步队列(Blocked状态)，等待
线程1执行完毕，释放锁 ---> 唤醒同步队列中的线程2 ---> 线程2获取锁，开始执行

1
2
3
4

代码示例：

public class SynchronizedExample {
    private int count = 0;
    private final Object lock = new Object();

    // 1. 修饰实例方法
    public synchronized void increment() {
        count++;
    }

    // 2. 修饰代码块
    public void incrementWithBlock() {
        synchronized (lock) {
            count++;
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

# 2.3. final

final关键字用于声明常量、不可变对象或不可覆盖的方法。

final变量：赋值后不可更改。如果是引用类型，则引用不可变，但对象内部状态可能可变(除非对象本身也是不可变的)。
final方法：不能被子类覆盖。
final类：不能被继承。

并发意义：final域能确保初始化安全性(Initialization Safety)。通过final域，JMM禁止把final域的写重排序到构造函数之外，从而能保证一个线程看到对象时，对象的final域已经被正确初始化，无需同步。

代码示例：

public class FinalExample {
    private final int immutableValue; // 必须在构造函数结束前初始化
    private final Map<String, String> immutableMap;

    public FinalExample(int value) {
        this.immutableValue = value;
        Map<String, String> map = new HashMap<>();
        map.put("key", "value");
        this.immutableMap = Collections.unmodifiableMap(map); // 创建不可变视图
    }
    // ... getters
}

1
2
3
4
5
6
7
8
9
10
11
12

# 3. JUC原子类：CAS, Unsafe和原子类详解

# 3.1. CAS (Compare-And-Swap)

CAS是一种乐观锁实现。操作包含三个操作数：内存位置(V)、预期原值(A)和新值(B)。当且仅当V的值等于A时，处理器才会用B更新V的值，否则不执行更新。整个操作是一个原子指令。

算法设计思想：无锁(Lock-Free)编程的核心。先读取当前值，进行计算，然后尝试用CAS更新。如果失败(说明其他线程已修改)，则重试(自旋)或采取其他策略。

文字描述的图形展示 - CAS 操作流程：

假设共享变量 value 在内存中初始为 0。

线程1： 读取 value = 0 (A)
线程1： 计算新值 B = A + 1 = 1
线程1： 执行 CAS(value, A=0, B=1) --> 成功！ value 变为 1。
        CAS流程：检查 value == 0? 是 --> 将 value 设置为 1。

线程2： 读取 value = 1 (A)
线程2： 计算新值 B = A + 1 = 2
线程1(同时)：再次读取 value = 1，计算 B=2，执行 CAS(value, 1, 2) 成功。
线程2： 执行 CAS(value, A=1, B=2) --> 失败！
        CAS流程：检查 value == 1? 否(现在 value 是 2) --> 设置失败。
线程2： 自旋重试，重新读取 value = 2，计算 B=3，执行 CAS(value, 2, 3) ...

1
2
3
4
5
6
7
8
9
10
11
12
13

# 3.2. Unsafe

sun.misc.Unsafe类提供了一系列执行低级别、不安全操作的方法，如直接操作内存、CAS、线程挂起/恢复等。它是许多并发工具类(如AtomicInteger、AQS)实现的基础。由于其不安全性，应用代码通常不应直接使用。

# 3.3. 原子类 (Atomic Classes)

JUC包提供了如AtomicInteger、AtomicLong、AtomicReference等原子类，它们利用CAS和volatile实现了单个变量的原子操作。

代码示例：

import java.util.concurrent.atomic.AtomicInteger;

public class AtomicExample {
    private AtomicInteger atomicCount = new AtomicInteger(0);

    public void safeIncrement() {
        // 相当于 count++，但是原子的
        atomicCount.incrementAndGet(); // 内部使用 CAS
    }

    public void updateWithCAS() {
        int oldValue, newValue;
        do {
            oldValue = atomicCount.get(); // 读取当前值
            newValue = oldValue + 10; // 计算新值
        } while (!atomicCount.compareAndSet(oldValue, newValue)); // CAS 更新，失败则循环重试
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

# 4. 线程基础与ThreadLocal

# 4.1. 线程基础

线程是调度的基本单位。Java中创建线程的方式：继承Thread类或实现Runnable/Callable接口。

线程状态：NEW, RUNNABLE, BLOCKED, WAITING, TIMED_WAITING, TERMINATED。
线程协作：Object.wait(), Object.notify()/notifyAll()。必须在synchronized块内使用。

# 4.2. ThreadLocal详解

ThreadLocal提供了线程局部变量。每个访问该变量的线程都有其独立的、初始化的变量副本。

设计思想：

空间换时间：为每个线程创建变量副本，避免了多线程访问共享变量时的同步开销。
数据隔离：每个线程只能访问自己的副本，实现了线程间的数据隔离。

实现原理：每个Thread对象内部都有一个ThreadLocalMap(一个定制化的WeakHashMap)。ThreadLocal本身作为Key，存储的变量副本作为Value。

ThreadLocal.get(): 获取当前线程的ThreadLocalMap，然后以ThreadLocal自身为Key获取Value。
ThreadLocal.set(T value): 同上，设置Key-Value对。

内存泄漏风险：ThreadLocalMap的Key是WeakReference<ThreadLocal>。如果ThreadLocal外部强引用被置为null，GC时Key会被回收，但Value是强引用，只要线程(如线程池中的核心线程)一直存活，Value就一直无法回收，导致内存泄漏。必须调用ThreadLocal.remove()清理。

文字描述的图形展示 - ThreadLocal 存储结构：

线程1 (Thread-1):                    线程2 (Thread-2):
ThreadLocalMap {                     ThreadLocalMap {
  Key: UserContextThreadLocal   -> Value: UserA对象         Key: UserContextThreadLocal   -> Value: UserB对象
  Key: TransactionThreadLocal   -> Value: TransactionX       Key: TransactionThreadLocal   -> Value: TransactionY
}                                 }

主线程 (Main):
ThreadLocalMap {
  Key: UserContextThreadLocal   -> Value: null (未设置)
}

1
2
3
4
5
6
7
8
9
10

如上图，同一个ThreadLocal对象(如UserContextThreadLocal)在不同线程的Map中作为Key，关联着不同的Value。

代码示例：

public class ThreadLocalExample {
    private static final ThreadLocal<SimpleDateFormat> dateFormatThreadLocal =
        ThreadLocal.withInitial(() -> new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"));

    public String formatDate(Date date) {
        // 每个线程获取自己独立的 SimpleDateFormat 实例，避免 SimpleDateFormat 非线程安全的问题
        return dateFormatThreadLocal.get().format(date);
    }

    public void cleanup() {
        // 重要：在线程结束时(如在线程池的Runnable的finally块中)清理，防止内存泄漏
        dateFormatThreadLocal.remove();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14

# 5. Java中所有的锁 & JUC锁

# 5.1. 锁的分类

乐观锁 vs 悲观锁：CAS vs synchronized/ReentrantLock。
公平锁 vs 非公平锁：按申请锁的顺序获取 vs 插队尝试获取。ReentrantLock可指定。
可重入锁(Reentrant Lock)：线程可以再次获取自己已经持有的锁。synchronized和ReentrantLock都是可重入的。
独占锁(写锁) vs 共享锁(读锁)：ReentrantReadWriteLock。

# 5.2. JUC锁核心类AQS详解

AbstractQueuedSynchronizer (AQS) 是构建锁和同步器的框架。JUC中的ReentrantLock、Semaphore、CountDownLatch等都基于AQS。

核心思想：

一个 volatile int 类型的 state 表示同步状态。
一个 FIFO 双向队列(CLH队列)来管理获取锁失败的线程。

文字描述的图形展示 - AQS 队列与 state：

初始： state = 0 (表示锁可用)，队列 head -> tail 都为空。

线程T1调用 lock()：
  尝试 CAS 将 state 从 0 设为 1 成功。T1持有锁。
  队列： head -> [空] <- tail

线程T2调用 lock()：
  尝试 CAS state(0->1) 失败(因为state=1)。
  将T2包装为Node，加入CLH队列尾部，并挂起(park)。
  队列： head -> [T2 Node] <- tail

线程T1调用 unlock()：
  将 state 从 1 设为 0。
  唤醒(unpark)队列头节点(即T2)。
  队列： head -> [T2 Node(被唤醒)] <- tail

线程T2被唤醒后：
  再次尝试获取锁(CAS state 0->1)成功。
  T2成为新的持有者，原头节点出队。
  队列： head -> [空] <- tail

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

# 5.3. LockSupport

提供线程阻塞和唤醒的基础原语park()和unpark(thread)。比Object.wait()/notify()更灵活(无需先获取锁，且许可证机制更精准)。

代码示例：

Thread thread = new Thread(() -> {
    System.out.println("Thread will be parked.");
    LockSupport.park(); // 阻塞，等待许可证
    System.out.println("Thread is unparked.");
});
thread.start();
Thread.sleep(1000);
LockSupport.unpark(thread); // 发放许可证，唤醒线程

1
2
3
4
5
6
7
8

# 5.4. ReentrantLock详解

基于AQS实现的可重入独占锁。比synchronized功能更丰富：可中断、可超时、可设置公平/非公平、支持多个条件变量(Condition)。

代码示例：

public class ReentrantLockExample {
    private final ReentrantLock lock = new ReentrantLock(true); // 公平锁
    private int count = 0;

    public void increment() {
        lock.lock(); // 获取锁，类比 synchronized(lock) {
        try {
            count++;
        } finally {
            lock.unlock(); // 必须finally中释放锁，类比 }
        }
    }

    public void tryLockExample() throws InterruptedException {
        if (lock.tryLock(1, TimeUnit.SECONDS)) { // 尝试获取锁，最多等待1秒
            try {
                // 操作共享资源
            } finally {
                lock.unlock();
            }
        } else {
            System.out.println("Get lock failed, do something else.");
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

# 5.5. ReentrantReadWriteLock详解

读写锁。读锁(共享锁)之间不互斥，写锁(独占锁)与任何锁都互斥。适用于读多写少的场景。

文字描述的图形展示 - 读写锁规则：

场景分析：
1. 线程A获取了读锁。
   --> 其他线程(B, C)可以获取读锁。 (读读不互斥)
   --> 其他线程(D)尝试获取写锁会被阻塞。 (读写互斥)

2. 线程A获取了写锁。
   --> 其他所有线程(B读锁，C写锁)都会被阻塞。 (写写互斥，写读互斥)

state的高16位表示读锁持有数，低16位表示写锁重入次数。

1
2
3
4
5
6
7
8
9

代码示例：

public class CacheWithReadWriteLock {
    private final Map<String, Object> cache = new HashMap<>();
    private final ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();
    private final ReentrantReadWriteLock.ReadLock readLock = rwLock.readLock();
    private final ReentrantReadWriteLock.WriteLock writeLock = rwLock.writeLock();

    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();
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

# 6. JUC工具类详解

# 6.1. CountDownLatch

一个或多个线程等待其他一组线程完成操作。一次性使用，计数无法重置。

设计思想：基于AQS，state初始化为计数N。线程调用countDown()时，state减1。调用await()的线程会阻塞，直到state变为0。

文字描述的图形展示 - CountDownLatch 工作流程：

主线程：创建 CountDownLatch(N=3)，启动3个工作线程，然后调用 latch.await() 进入等待。

工作线程1：完成任务，调用 latch.countDown()， N=2。
工作线程2：完成任务，调用 latch.countDown()， N=1。
工作线程3：完成任务，调用 latch.countDown()， N=0。

主线程：感知到 N==0，从 await() 返回，继续执行。

1
2
3
4
5
6
7

代码示例：

public class CountDownLatchDemo {
    public static void main(String[] args) throws InterruptedException {
        int workerCount = 3;
        CountDownLatch startSignal = new CountDownLatch(1); // 统一开始信号
        CountDownLatch doneSignal = new CountDownLatch(workerCount); // 完成信号

        for (int i = 0; i < workerCount; i++) {
            new Thread(() -> {
                try {
                    startSignal.await(); // 等待主线程发令
                    System.out.println(Thread.currentThread().getName() + " working...");
                    Thread.sleep((long) (Math.random() * 1000));
                    doneSignal.countDown();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
        }

        Thread.sleep(1000); // 主线程准备时间
        System.out.println("Ready... Go!");
        startSignal.countDown(); // 发令，所有工作线程开始运行

        doneSignal.await(); // 等待所有工作线程完成
        System.out.println("All workers finished.");
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

# 6.2. CyclicBarrier

一组线程相互等待，到达一个公共屏障点(Barrier)时被阻塞，直到最后一个线程到达屏障，所有被阻塞的线程才能继续执行。可循环使用(Reset)。

设计思想：内部使用ReentrantLock和Condition实现。每代(Generation)有一个计数。线程调用await()时计数减1并等待。当计数减为0时，触发预定义的Runnable屏障动作(如果存在)，然后唤醒所有等待线程，并重置计数开启下一代。

文字描述的图形展示 - CyclicBarrier 工作流程：

线程T1, T2, T3 同时开始执行，CyclicBarrier(3)。

T1: 到达屏障点，调用 await()， 计数减为2，T1等待。
T2: 到达屏障点，调用 await()， 计数减为1，T2等待。
T3: 到达屏障点，调用 await()， 计数减为0。
    触发屏障动作(如果有)。
    唤醒T1和T2。
    重置计数为3，开启新一代。
T1, T2, T3: 同时继续执行各自后续任务。

1
2
3
4
5
6
7
8
9

代码示例：

public class CyclicBarrierDemo {
    public static void main(String[] args) {
        int threadCount = 3;
        CyclicBarrier barrier = new CyclicBarrier(threadCount,
                () -> System.out.println("All threads reached barrier, continue together!"));

        for (int i = 0; i < threadCount; i++) {
            final int threadId = i;
            new Thread(() -> {
                try {
                    System.out.println("Thread " + threadId + " is preparing...");
                    Thread.sleep((long) (Math.random() * 2000));
                    System.out.println("Thread " + threadId + " reached barrier.");
                    barrier.await(); // 等待其他线程
                    System.out.println("Thread " + threadId + " continues its work.");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

# 6.3. Semaphore

信号量，用于控制同时访问特定资源的线程数量。可以用于做流量控制或资源池。

设计思想：基于AQS的共享模式。state表示许可证(permits)数量。acquire()获取许可证(state减1)，没有许可证时阻塞；release()释放许可证(state加1)。

文字描述的图形展示 - Semaphore 控制资源访问：

资源池(如数据库连接)有3个许可证 (Semaphore permits=3)。

线程T1, T2, T3, T4, T5 请求访问。

T1: acquire()成功，获得许可1，剩余permits=2，开始访问。
T2: acquire()成功，获得许可2，剩余permits=1，开始访问。
T3: acquire()成功，获得许可3，剩余permits=0，开始访问。

T4: acquire()失败，因为permits=0，进入等待队列阻塞。
T5: acquire()失败，同样阻塞。

线程T2: 访问结束，release()， permits变为1。
    唤醒等待队列中的一个线程(如T4)。
T4: 获取许可， permits变为0，开始访问。

1
2
3
4
5
6
7
8
9
10
11
12
13
14

代码示例：

public class SemaphoreDemo {
    // 模拟一个只有5个许可的信号量，控制并发线程数
    private static final Semaphore SEMAPHORE = new Semaphore(5);

    public static void main(String[] args) {
        for (int i = 0; i < 10; i++) {
            new Thread(() -> {
                try {
                    SEMAPHORE.acquire(); // 获取许可
                    System.out.println(Thread.currentThread().getName() + " got permit, working...");
                    Thread.sleep(2000);
                    System.out.println(Thread.currentThread().getName() + " release permit.");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    SEMAPHORE.release(); // 释放许可
                }
            }).start();
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

# 6.4. Phaser

更灵活、功能更强的可重用同步屏障。可以动态注册/注销参与线程，并分阶段(Phase)同步。

设计思想：每个阶段(Phase)都有一个编号，从0开始。线程通过arriveAndAwaitAdvance()到达并等待其他线程。当所有已注册的线程都到达后，阶段号递增，所有线程进入下一阶段。支持分层(Tiering)减少竞争。

代码示例(简化)：

public class PhaserDemo {
    public static void main(String[] args) {
        Phaser phaser = new Phaser(3); // 注册3个线程
        for (int i = 0; i < 3; i++) {
            final int taskId = i;
            new Thread(() -> {
                System.out.println("Task-" + taskId + " phase-0 start.");
                phaser.arriveAndAwaitAdvance(); // 到达阶段0，等待其他线程

                System.out.println("Task-" + taskId + " phase-1 start.");
                phaser.arriveAndAwaitAdvance(); // 到达阶段1，等待

                System.out.println("Task-" + taskId + " finished.");
                phaser.arriveAndDeregister(); // 到达并注销
            }).start();
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

# 6.5. Exchanger

用于两个线程间交换数据。每个线程在调用exchange(T myData)时，会阻塞等待另一个线程也调用此方法，然后交换数据并同时返回。

代码示例：

public class ExchangerDemo {
    public static void main(String[] args) {
        Exchanger<String> exchanger = new Exchanger<>();
        new Thread(() -> {
            String dataA = "Data from Thread-A";
            try {
                System.out.println("Thread-A sending: " + dataA);
                String received = exchanger.exchange(dataA);
                System.out.println("Thread-A received: " + received);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();

        new Thread(() -> {
            String dataB = "Data from Thread-B";
            try {
                Thread.sleep(1000); // 模拟处理耗时
                System.out.println("Thread-B sending: " + dataB);
                String received = exchanger.exchange(dataB);
                System.out.println("Thread-B received: " + received);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

# 7. JUC集合详解

# 7.1. BlockingQueue详解

阻塞队列，支持当队列满时插入操作被阻塞，队列空时获取操作被阻塞。是生产者-消费者模型的核心数据结构。

常用实现：

ArrayBlockingQueue：有界、数组实现、可选公平性。
LinkedBlockingQueue：可选有界(默认Integer.MAX_VALUE)、链表实现、两把锁(putLock, takeLock)提高吞吐。
PriorityBlockingQueue：无界、支持优先级排序、基于堆。
SynchronousQueue：不存储元素，每个插入操作必须等待一个对应的移除操作，反之亦然。直接传递。

文字描述的图形展示 - ArrayBlockingQueue (capacity=3) 工作流程：

初始： head=0, tail=0, count=0
队列： [空, 空, 空]

生产者P1: put("A")成功。 tail=1, count=1
队列： ["A", 空, 空]
         head
         tail

生产者P2: put("B")成功。 tail=2, count=2
队列： ["A", "B", 空]
         head       tail

消费者C1: take()得到"A"。 head=1, count=1
队列： [空, "B", 空]
              head
              tail

生产者P3: put("C")成功。 tail=(2+1)%3=0, count=2
队列： ["C", "B", 空]
              head  tail

生产者P4: put("D") 队列满(count==capacity)，P4被阻塞。
消费者C2: take()得到"B"。 head=2, count=1
      唤醒被阻塞的生产者P4。
队列： ["C", 空, 空]
                   head
         tail

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

代码示例 - 生产者消费者：

public class BlockingQueueDemo {
    private static final BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(10);

    static class Producer implements Runnable {
        @Override
        public void run() {
            try {
                int value = 0;
                while (true) {
                    queue.put(value);
                    System.out.println("Produced: " + value);
                    value++;
                    Thread.sleep(100);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    static class Consumer implements Runnable {
        @Override
        public void run() {
            try {
                while (true) {
                    Integer value = queue.take();
                    System.out.println("Consumed: " + value);
                    Thread.sleep(200);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

# 7.2. ConcurrentHashMap详解

高并发环境下线程安全的HashMap。JDK 7采用分段锁(Segment)，JDK 8改为 synchronized + CAS + 红黑树。

JDK 8 设计思想：

数组+链表+红黑树：同HashMap。
CAS初始化/扩容：使用volatile和CAS保证原子性操作。
锁粒度细化：只锁住链表或红黑树的头节点(桶)，而不是整个表。
并发扩容：多个线程可以协助进行数据迁移。

文字描述的图形展示 - ConcurrentHashMap (JDK8) 结构：

table: (volatile Node<K,V>[])
索引 0: -> Node1(key1,hash1) -> Node2(key2,hash2) -> null (链表)
索引 1: -> TreeNode...(红黑树) ...
索引 2: -> ForwardingNode (扩容时占位符，指向新表nextTable)
索引 3: -> NodeX... -> null

线程T1 put(key1):
  计算hash，定位到桶索引0。
  如果桶为空，CAS插入新Node。
  如果桶不为空，synchronized锁住桶的头节点(Node1)，遍历链表插入/更新。

线程T2 get(key2):
  计算hash，定位到桶索引0。
  由于get操作不涉及修改，且value用volatile修饰，可以直接遍历链表读取，无需加锁。

1
2
3
4
5
6
7
8
9
10
11
12
13
14

代码示例：

public class ConcurrentHashMapDemo {
    private static final ConcurrentHashMap<String, Integer> scores = new ConcurrentHashMap<>();

    public static void main(String[] args) throws InterruptedException {
        // 线程安全的聚合操作
        scores.put("Alice", 0);
        Runnable task = () -> {
            for (int i = 0; i < 1000; i++) {
                scores.compute("Alice", (k, v) -> v + 1); // 原子性的更新函数
                // 或者用 merge: scores.merge("Alice", 1, Integer::sum);
            }
        };
        Thread t1 = new Thread(task);
        Thread t2 = new Thread(task);
        t1.start(); t2.start();
        t1.join(); t2.join();
        System.out.println("Final score: " + scores.get("Alice")); // 输出 2000
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

# 7.3. ConcurrentLinkedQueue

基于无锁(Lock-Free)算法实现的线程安全非阻塞FIFO队列。使用CAS操作(head、tail节点)实现入队和出队。

设计思想：

不变式(Invariants)：所有存活的节点都能通过head节点的next链最终到达。
松弛更新(Slack)：tail和head节点可能滞后于真实的尾节点和头节点，以降低CAS竞争频率。

代码示例：

public class ConcurrentLinkedQueueDemo {
    public static void main(String[] args) {
        ConcurrentLinkedQueue<String> queue = new ConcurrentLinkedQueue<>();
        queue.offer("Task1");
        queue.offer("Task2");
        String task = queue.poll(); // 非阻塞，取不到返回null
        if (task != null) {
            System.out.println("Processed: " + task);
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11

# 7.4. CopyOnWriteArrayList

写时复制(Copy-On-Write) 列表。适用于读多写少的并发场景。

设计思想：

读取：直接读取，无需加锁，因为内部数组(array)是volatile的。
写入(add, set, remove)：加锁(ReentrantLock)，复制原数组，在新数组上执行修改，然后将volatile的array引用指向新数组。

优点：读取性能极高，完全无锁。缺点：内存占用大(每次写都复制)，数据一致性是最终一致性，读取时可能读到旧数据。

文字描述的图形展示 - CopyOnWriteArrayList 写操作：

初始： array -> ["A", "B", "C"] (volatile引用)

线程T1 (读)： String e = list.get(1); // 直接读取 array[1] -> "B"

线程T2 (写：添加"D")：
  1. 获取锁。
  2. 复制原数组：newArray = ["A", "B", "C", 空]
  3. 在新数组上操作：newArray[3] = "D"
  4. 将 volatile 的 array 引用指向新数组：array = newArray
  5. 释放锁。

之后所有读操作都看到新数组：["A", "B", "C", "D"]
线程T1在写之前读取，所以看到的是旧数组。这就是弱一致性。

1
2
3
4
5
6
7
8
9
10
11
12
13

代码示例：

public class CopyOnWriteArrayListDemo {
    // 非常适合做监听器列表
    private static final CopyOnWriteArrayList<EventListener> listeners = new CopyOnWriteArrayList<>();

    public static void addListener(EventListener listener) {
        listeners.add(listener); // 写时复制
    }

    public static void fireEvent(Event event) {
        // 遍历时，即使有新的监听器加入，也不会影响当前这次事件的触发(遍历的是旧的快照)
        for (EventListener listener : listeners) {
            listener.onEvent(event);
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

# 8. JUC线程池详解

# 8.1. ThreadPoolExecutor

线程池的核心实现类。旨在通过复用线程来减少创建和销毁线程的开销，并提供任务队列和拒绝策略来管理过载任务。

核心参数：

corePoolSize：核心线程数，即使空闲也会保留(除非allowCoreThreadTimeOut=true)。
maximumPoolSize：最大线程数。
workQueue：任务队列，保存等待执行的任务(如LinkedBlockingQueue, ArrayBlockingQueue, SynchronousQueue)。
keepAliveTime：非核心线程空闲时的存活时间。
threadFactory：创建新线程的工厂。
handler：饱和(拒绝)策略。AbortPolicy(默认，抛异常)、CallerRunsPolicy(调用者运行)、DiscardOldestPolicy(丢弃最老)、DiscardPolicy(丢弃)。

文字描述的图形展示 - ThreadPoolExecutor 任务处理流程：

新任务提交(execute(Runnable command)):

      |-- 当前运行线程数 < corePoolSize?
      |      是 --> 创建新核心线程执行任务。
      |      否 --> 任务放入 workQueue？
      |              是 --> 等待。
      |              否 (队列已满) --> 当前线程数 < maximumPoolSize？
      |                                  是 --> 创建新非核心线程执行任务。
      |                                  否 --> 执行拒绝策略 (handler.rejectedExecution)。

状态流转：
      RUNNING (接受新任务，处理队列任务)
          |-- shutdown() -> SHUTDOWN (不接受新任务，处理队列任务)
          |-- shutdownNow() -> STOP (不接受新任务，不处理队列任务，中断进行中任务)
          |-- (任务队列空，线程池为空) -> TIDYING
          |-- terminated()钩子执行完毕 -> TERMINATED

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

代码示例：

public class ThreadPoolExecutorDemo {
    public static void main(String[] args) {
        int corePoolSize = 2;
        int maxPoolSize = 4;
        long keepAliveTime = 10L;
        BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(2);
        RejectedExecutionHandler handler = new ThreadPoolExecutor.CallerRunsPolicy();

        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                corePoolSize, maxPoolSize, keepAliveTime, TimeUnit.SECONDS,
                workQueue, Executors.defaultThreadFactory(), handler);

        for (int i = 1; i <= 10; i++) {
            final int taskId = i;
            executor.execute(() -> {
                System.out.println(Thread.currentThread().getName() + " executing task " + taskId);
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

        executor.shutdown(); // 平滑关闭
        try {
            executor.awaitTermination(1, TimeUnit.MINUTES);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("All tasks finished.");
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

# 8.2. ScheduledThreadPoolExecutor

用于在给定延迟后运行任务，或者定期执行任务。基于ThreadPoolExecutor，使用DelayedWorkQueue(一个无界的优先级阻塞队列，按任务的触发时间排序)作为任务队列。

核心方法：

schedule(Runnable command, long delay, TimeUnit unit): 延迟执行一次。
scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit): 固定速率执行(关注开始时间间隔)。
scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit): 固定延迟执行(关注结束到开始的间隔)。

代码示例：

public class ScheduledThreadPoolDemo {
    public static void main(String[] args) {
        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);

        // 延迟5秒后执行一次
        scheduler.schedule(() -> System.out.println("Delayed task"), 5, TimeUnit.SECONDS);

        // 延迟1秒后开始，每3秒执行一次(固定速率)
        scheduler.scheduleAtFixedRate(() -> System.out.println("Fixed Rate task"), 1, 3, TimeUnit.SECONDS);

        // 延迟2秒后开始，每次执行结束后等2秒再执行下一次(固定延迟)
        scheduler.scheduleWithFixedDelay(() -> {
            System.out.println("Fixed Delay task start");
            try {
                Thread.sleep(1000); // 模拟任务执行耗时
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("Fixed Delay task end");
        }, 2, 2, TimeUnit.SECONDS);

        // scheduler.shutdown(); // 实际应用记得关闭
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

# 8.3. FutureTask详解

FutureTask实现了RunnableFuture接口，而RunnableFuture继承了Runnable和Future。因此，它可以被提交给Executor执行，也可以用来获取异步计算的结果。

设计思想：

内部维护任务状态(state)，包括NEW, COMPLETING, NORMAL, EXCEPTIONAL, CANCELLED, INTERRUPTING, INTERRUPTED。
调用get()获取结果的线程，如果任务未完成，会通过LockSupport.park()进入等待队列。
任务执行完成后(无论是正常结束还是异常)，会唤醒所有等待的线程。

代码示例：

public class FutureTaskDemo {
    public static void main(String[] args) throws Exception {
        Callable<Integer> callableTask = () -> {
            Thread.sleep(2000);
            return 42;
        };

        FutureTask<Integer> futureTask = new FutureTask<>(callableTask);
        new Thread(futureTask).start(); // 可以用线程执行

        System.out.println("Do something else while task is running...");
        // 阻塞等待结果
        Integer result = futureTask.get(); // 如果任务未完成，会阻塞在这里
        System.out.println("Task result: " + result);

        // 也可以提交给线程池
        ExecutorService executor = Executors.newSingleThreadExecutor();
        Future<Integer> future = executor.submit(callableTask);
        System.out.println("Future from executor: " + future.get());
        executor.shutdown();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

# 8.4. Fork-Join框架详解

用于并行执行可以递归拆分成子任务的大任务。采用工作窃取(Work-Stealing)算法。

核心组件：

ForkJoinPool：特殊的线程池，用于运行ForkJoinTask。
ForkJoinTask：表示可以被ForkJoinPool执行的任务。其子类RecursiveTask(有返回值)和RecursiveAction(无返回值)。

工作窃取算法思想：每个工作线程都维护一个双端队列(Deque)来存放自己的任务。

线程执行自己队列中的任务(从队头LIFO或FIFO获取，取决于实现)。
当自己的队列为空时，会从其他线程的队列队尾窃取一个任务执行(FIFO)。

优点：减少了线程间的竞争，充分利用了CPU资源。

文字描述的图形展示 - 工作窃取(Work-Stealing)：

ForkJoinPool 有3个工作线程 (T1, T2, T3)，每个有自己的双端队列。

初始： T1队列: [大任务A](可拆分)   T2队列: [空]   T3队列: [空]

步骤1： T1从自己队头取出任务A，执行，并将其拆分为A1, A2, A3。将A2, A3压入自己队头。
        T1队列: [A2, A3] (队头在右)
        T1继续执行A1。

步骤2： T2空闲，自己的队列空，于是随机选择T1，从其队列的**队尾**窃取一个任务(窃取A2)。
        T2队列: [A2]
        T1队列: [A3]
        T1继续执行A1，T2执行A2。

步骤3： T3空闲，窃取T1的队尾任务A3。
        T3队列: [A3]
        T1队列: [空]
        T1, T2, T3并行执行子任务。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

代码示例 - 计算1到n的和：

public class ForkJoinSumCalculator extends RecursiveTask<Long> {
    private static final long THRESHOLD = 10_000; // 拆分阈值
    private final long[] numbers;
    private final int start;
    private final int end;

    public ForkJoinSumCalculator(long[] numbers) {
        this(numbers, 0, numbers.length);
    }
    private ForkJoinSumCalculator(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) {
            // 如果任务足够小，直接计算
            return computeSequentially();
        }
        // 拆分任务
        ForkJoinSumCalculator leftTask = new ForkJoinSumCalculator(numbers, start, start + length / 2);
        leftTask.fork(); // 异步执行左子任务(将其推入当前线程的队列)
        ForkJoinSumCalculator rightTask = new ForkJoinSumCalculator(numbers, start + length / 2, end);
        Long rightResult = rightTask.compute(); // 同步执行右子任务
        Long leftResult = leftTask.join(); // 获取左子任务结果(阻塞等待)
        return leftResult + rightResult; // 合并结果
    }

    private long computeSequentially() {
        long sum = 0;
        for (int i = start; i < end; i++) {
            sum += numbers[i];
        }
        return sum;
    }

    public static void main(String[] args) {
        long[] numbers = LongStream.rangeClosed(1, 1_000_000).toArray();
        ForkJoinTask<Long> task = new ForkJoinSumCalculator(numbers);
        Long result = new ForkJoinPool().invoke(task); // 提交任务给ForkJoinPool
        System.out.println("Sum: " + result);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

本教程涵盖了Java并发编程的核心框架和组件，从内存模型、线程基础、锁机制，到JUC提供的丰富工具集和线程池。理解这些内容需要结合实践，建议在学习过程中多编写和调试代码，以加深理解。并发编程复杂且微妙，务必重视线程安全和性能的权衡。

← Java进阶集合框架 Java进阶IO框架 →