Java并发编程(JUC)

前置知识要求

在学习 JUC 之前，你需要掌握以下基础知识：

基础多线程是学习JUC的前提，务必确保基础概念清晰后再开始JUC的学习旅程。

什么是 JUC？

java.util.concurrent 简称 JUC，是 Java 5 引入的并发工具包。它提供了一套强大的并发编程工具，大大简化了多线程开发。

🎯 为什么需要并发编程？

• 提升性能：充分利用多核CPU的处理能力
• 改善响应：避免耗时任务阻塞用户界面
• 提高吞吐：同时处理多个请求或任务

📦 JUC 核心组件概览

// JUC 包的主要结构
java.util.concurrent           // 核心并发工具
├── Executor                    // 执行器框架
├── ExecutorService             // 线程池接口
├── ThreadPoolExecutor          // 线程池实现
├── Callable/Volatile           // 异步任务和原子操作
├── CountDownLatch/CyclicBarrier// 同步工具
├── Semaphore                   // 信号量
├── Lock                        // 显式锁
├── Concurrent collections      // 并发集合
└── Atomic classes              // 原子类

JUC 让并发编程从"手工作坊"时代进入了"工业化生产"时代，大大降低了并发编程的复杂度。

JUC 高级特性概览

在掌握多线程基础后，JUC 提供了更强大的高级特性来解决复杂的并发问题。

🚀 JUC 核心优势

📦 JUC 高级特性图谱

JUC 高级特性体系
├── 线程池体系 (ThreadPoolExecutor)
├── 锁机制 (ReentrantLock, ReadWriteLock)
├── 同步工具 (Semaphore, CountDownLatch, CyclicBarrier)
├── 定时任务 (ScheduledExecutorService)
├── 并行计算 (Fork/Join 框架)
├── 并发集合 (ConcurrentHashMap, BlockingQueue)
├── 原子类 (AtomicInteger, AtomicReference)
├── 无锁编程 (CAS 算法)
└── 异步编程 (CompletableFuture)

JUC 让并发编程从"能用"提升到"好用"和"高性能"，是高级Java开发的必备技能。

线程池基础（Executor）

import java.util.concurrent.*;

public class PoolDemo {
    public static void main(String[] args) {
        ExecutorService pool = new ThreadPoolExecutor(
            2, 4, 60, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(100),
            Executors.defaultThreadFactory(),
            new ThreadPoolExecutor.AbortPolicy()
        );

        for (int i = 0; i < 10; i++) {
            int id = i;
            pool.submit(() -> {
                System.out.println("任务" + id + " - " + Thread.currentThread().getName());
            });
        }

        pool.shutdown();
    }
}

常用线程池工厂

ExecutorService fixed = Executors.newFixedThreadPool(4);
ExecutorService cached = Executors.newCachedThreadPool();
ExecutorService single = Executors.newSingleThreadExecutor();
ScheduledExecutorService scheduled = Executors.newScheduledThreadPool(2);

scheduled.scheduleAtFixedRate(() -> {
    System.out.println("tick");
}, 0, 1, TimeUnit.SECONDS);

Callable与Future：带返回值的异步任务

前面我们学习了线程池提交 Runnable 任务，但如果任务需要返回结果呢？

import java.util.concurrent.*;

public class CallableDemo {
    public static void main(String[] args) throws Exception {
        // 1. 定义 Callable 任务
        Callable<Integer> task = () -> {
            int sum = 0;
            for (int i = 1; i <= 100; i++) {
                sum += i;
            }
            return sum; // 返回计算结果
        };

        // 2. 创建线程池并提交任务
        ExecutorService pool = Executors.newFixedThreadPool(2);
        Future<Integer> future = pool.submit(task);

        // 3. 获取结果（会阻塞直到任务完成）
        Integer result = future.get();
        System.out.println("计算结果: " + result);

        // 4. 关闭线程池
        pool.shutdown();
    }
}

💡 实际应用场景：并发计算

// 并发计算示例
ExecutorService pool = Executors.newFixedThreadPool(10);
List<Future<Integer>> futures = new ArrayList<>();

// 提交 10 个计算任务
for (int i = 0; i < 10; i++) {
    final int start = i * 100 + 1;
    Callable<Integer> task = () -> {
        int sum = 0;
        for (int j = start; j < start + 100; j++) {
            sum += j;
        }
        return sum;
    };
    futures.add(pool.submit(task));
}

// 汇总所有结果
int total = 0;
for (Future<Integer> f : futures) {
    total += f.get(); // 阻塞等待每个任务完成
}

System.out.println("总和: " + total); // 500500
pool.shutdown();

Callable + Future 是异步编程的基础。配合线程池可以实现高效的并发计算。 在 Java 8 中，还可以使用更强大的 CompletableFuture 进行链式调用和组合操作（详见本章后续内容）。

ThreadPoolExecutor 深度配置与监控 (1)

从基础线程池使用到生产级配置优化，JUC 提供了完整的线程池管理和监控能力。

🏗️ ThreadPoolExecutor 核心参数详解

⚙️ 生产级线程池配置示例

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

public class ProductionThreadPoolConfig {

    // 自定义线程工厂
    static class CustomThreadFactory implements ThreadFactory {
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;

        CustomThreadFactory(String namePrefix) {
            this.namePrefix = namePrefix;
        }

        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r, namePrefix + "-" + threadNumber.getAndIncrement());
            t.setDaemon(false); // 非守护线程
            t.setPriority(Thread.NORM_PRIORITY);
            // 设置未捕获异常处理器
            t.setUncaughtExceptionHandler((thread, ex) -> {
                System.err.printf("Thread %s threw exception: %s%n",
                    thread.getName(), ex.getMessage());
            });
            return t;
        }
    }

    // CPU密集型线程池配置
    public static ThreadPoolExecutor cpuIntensivePool() {
        int coreCount = Runtime.getRuntime().availableProcessors();
        return new ThreadPoolExecutor(
            coreCount,           // 核心线程数 = CPU核数
            coreCount,           // 最大线程数 = CPU核数（无扩容）
            0L, TimeUnit.MILLISECONDS,  // 核心线程无超时
            new LinkedBlockingQueue<>(1000), // 有界队列
            new CustomThreadFactory("CPU-Pool"),
            new ThreadPoolExecutor.AbortPolicy() // 队列满则拒绝
        );
    }

    // IO密集型线程池配置
    public static ThreadPoolExecutor ioIntensivePool() {
        int coreCount = Runtime.getRuntime().availableProcessors();
        return new ThreadPoolExecutor(
            coreCount * 2,       // 核心线程数 = CPU核数 * 2
            coreCount * 4,       // 最大线程数 = CPU核数 * 4
            60L, TimeUnit.SECONDS, // 非核心线程60秒超时
            new LinkedBlockingQueue<>(2000), // 更大的队列
            new CustomThreadFactory("IO-Pool"),
            new ThreadPoolExecutor.CallerRunsPolicy() // 队列满时调用者执行
        );
    }
}

合理的线程池配置需要根据任务类型（CPU密集型/IO密集型）和系统资源进行调优。

线程池监控与调优 (2)

生产环境中需要对线程池进行实时监控和动态调优，确保系统稳定性和性能。

📊 线程池核心监控指标

🚀 线程池监控系统实现

public class ThreadPoolMonitor {
    private final ThreadPoolExecutor executor;
    private final ScheduledExecutorService monitor;

    public ThreadPoolMonitor(ThreadPoolExecutor executor) {
        this.executor = executor;
        this.monitor = Executors.newSingleThreadScheduledExecutor();
        startMonitoring();
    }

    private void startMonitoring() {
        monitor.scheduleAtFixedRate(() -> {
            try {
                // 收集监控数据
                int activeCount = executor.getActiveCount();
                int poolSize = executor.getPoolSize();
                int corePoolSize = executor.getCorePoolSize();
                int maximumPoolSize = executor.getMaximumPoolSize();
                long completedTaskCount = executor.getCompletedTaskCount();
                long taskCount = executor.getTaskCount();
                int queueSize = executor.getQueue().size();

                // 计算性能指标
                double completionRate = taskCount > 0 ?
                    (double) completedTaskCount / taskCount : 0.0;
                double queueUtilization = maximumPoolSize > 0 ?
                    (double) queueSize / (queueSize + poolSize) : 0.0;

                // 输出监控信息
                System.out.printf("=== 线程池监控 ===%n");
                System.out.printf("活跃线程: %d/%d (核心: %d, 最大: %d)%n",
                    activeCount, poolSize, corePoolSize, maximumPoolSize);
                System.out.printf("任务完成: %d/%d (%.2f%%)%n",
                    completedTaskCount, taskCount, completionRate * 100);
                System.out.printf("队列大小: %d (利用率: %.2f%%)%n",
                    queueSize, queueUtilization * 100);

                // 告警检查
                checkAlerts(activeCount, poolSize, queueSize, completionRate);

            } catch (Exception e) {
                System.err.println("监控异常: " + e.getMessage());
            }
        }, 0, 5, TimeUnit.SECONDS);
    }
}

实时监控和动态调优是生产环境线程池管理的关键技能。

Lock 与可重入锁

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

public class LockDemo {
    private final ReentrantLock lock = new ReentrantLock(true);
    public void work() throws InterruptedException {
        if (lock.tryLock(100, TimeUnit.MILLISECONDS)) {
            try {
                // 临界区
            } finally {
                lock.unlock();
            }
        } else {
            System.out.println("获取锁超时");
        }
    }
}

Condition 与生产者/消费者

import java.util.LinkedList;
import java.util.concurrent.locks.*;

class PCQueue<T> {
    private final LinkedList<T> q = new LinkedList<>();
    private final int cap;
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition notEmpty = lock.newCondition();
    private final Condition notFull = lock.newCondition();

    PCQueue(int cap) { this.cap = cap; }

    public void put(T x) throws InterruptedException {
        lock.lock();
        try {
            while (q.size() == cap) notFull.await();
            q.add(x);
            notEmpty.signal();
        } finally { lock.unlock(); }
    }

    public T take() throws InterruptedException {
        lock.lock();
        try {
            while (q.isEmpty()) notEmpty.await();
            T x = q.removeFirst();
            notFull.signal();
            return x;
        } finally { lock.unlock(); }
    }
}

import java.util.concurrent.*;

BlockingQueue<Integer> q = new ArrayBlockingQueue<>(10);

Runnable producer = () -> {
    try {
        for (int i = 0; i < 100; i++) q.put(i);
    } catch (InterruptedException ignored) {}
};

Runnable consumer = () -> {
    try {
        while (true) {
            Integer x = q.take();
            // 处理 x
        }
    } catch (InterruptedException ignored) {}
};

高级锁机制与死锁排查

掌握 JUC 锁机制的高级特性，学会识别和解决死锁问题。

🔒 ReentrantReadWriteLock：读写分离锁

🚨 死锁识别与排查

public class DeadlockDetection {

    // 死锁的四个必要条件
    public static void analyzeDeadlockConditions() {
        System.out.println("死锁的四个必要条件：");
        System.out.println("1. 互斥条件：资源不能被多个线程同时使用");
        System.out.println("2. 占有并等待：线程持有资源的同时等待其他资源");
        System.out.println("3. 不可抢占：资源不能被强制性地抢占");
        System.out.println("4. 循环等待：存在线程资源的环形等待链");
    }

    // 死锁预防策略
    public static class DeadlockPrevention {

        // 1. 锁排序：所有线程按固定顺序获取锁
        private static final Object lock1 = new Object();
        private static final Object lock2 = new Object();

        public static void lockOrderingExample() {
            // 所有线程都按 lock1 -> lock2 的顺序获取锁
            synchronized (lock1) {
                System.out.println("获取 lock1");
                synchronized (lock2) {
                    System.out.println("获取 lock2 - 不会死锁");
                }
            }
        }

        // 2. 超时机制：使用 tryLock 避免无限等待
        public static void timeoutExample() throws InterruptedException {
            ReentrantLock lock1 = new ReentrantLock();
            ReentrantLock lock2 = new ReentrantLock();

            // 先尝试获取 lock1，再尝试获取 lock2
            if (lock1.tryLock(1, TimeUnit.SECONDS)) {
                try {
                    if (lock2.tryLock(1, TimeUnit.SECONDS)) {
                        try {
                            System.out.println("成功获取两个锁");
                        } finally {
                            lock2.unlock();
                        }
                    } else {
                        System.out.println("获取 lock2 超时，释放 lock1");
                    }
                } finally {
                    lock1.unlock();
                }
            }
        }
    }
}

掌握死锁预防和检测技巧是编写可靠并发程序的关键。

Semaphore 信号量

import java.util.concurrent.Semaphore;

public class SemDemo {
    static final Semaphore sem = new Semaphore(3, true);
    static void serve() {
        try {
            sem.acquire();
            // 处理请求
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            sem.release();
        }
    }
}

同步工具类：协调线程执行

JUC 提供了多种同步工具类，用于协调多个线程的执行顺序。

🚦 CountDownLatch：倒计时门闩

public class CountDownLatchExample {
    public static void main(String[] args) throws InterruptedException {
        // 创建一个 CountDownLatch，计数为 3
        CountDownLatch latch = new CountDownLatch(3);

        // 启动 3 个工作线程
        for (int i = 0; i < 3; i++) {
            int workerId = i;
            new Thread(() -> {
                try {
                    System.out.println("工作线程 " + workerId + " 开始工作");
                    Thread.sleep(2000); // 模拟工作
                    System.out.println("工作线程 " + workerId + " 完成工作");
                    latch.countDown(); // 计数减 1
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }).start();
        }

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

🔄 CyclicBarrier：循环栅栏

public class CyclicBarrierExample {
    public static void main(String[] args) {
        // 创建 CyclicBarrier，等待 3 个线程，到达后执行指定动作
        CyclicBarrier barrier = new CyclicBarrier(3, () -> {
            System.out.println("=== 所有线程都到达屏障点，执行汇总操作 ===");
        });

        for (int i = 0; i < 3; i++) {
            int threadId = i;
            new Thread(() -> {
                try {
                    System.out.println("线程 " + threadId + " 第一个阶段");
                    Thread.sleep(1000);

                    System.out.println("线程 " + threadId + " 到达屏障点，等待其他线程");
                    barrier.await(); // 等待其他线程

                    System.out.println("线程 " + threadId + " 第二个阶段");

                } catch (Exception e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
}

定时任务：Timer vs ScheduledExecutor

import java.util.Timer;
import java.util.TimerTask;

Timer timer = new Timer();
timer.schedule(new TimerTask() {
    public void run() { System.out.println("once"); }
}, 1000);

import java.util.concurrent.*;

ScheduledExecutorService ses = Executors.newScheduledThreadPool(2);
ses.scheduleAtFixedRate(() -> {
    System.out.println("fixed-rate");
}, 0, 1, TimeUnit.SECONDS);

ses.scheduleWithFixedDelay(() -> {
    System.out.println("fixed-delay");
}, 0, 1, TimeUnit.SECONDS);

Fork/Join 框架

import java.util.concurrent.*;

class SumTask extends RecursiveTask<Long> {
    private final long[] arr; private final int start, end; private static final int THRESHOLD = 10_000;
    SumTask(long[] arr, int start, int end) { this.arr = arr; this.start = start; this.end = end; }
    protected Long compute() {
        int len = end - start;
        if (len <= THRESHOLD) {
            long sum = 0; for (int i = start; i < end; i++) sum += arr[i]; return sum;
        }
        int mid = start + len / 2;
        SumTask left = new SumTask(arr, start, mid);
        SumTask right = new SumTask(arr, mid, end);
        left.fork();
        long r = right.compute();
        long l = left.join();
        return l + r;
    }
}

ForkJoinPool pool = ForkJoinPool.commonPool();
long[] data = new long[5_000_000];
long result = pool.invoke(new SumTask(data, 0, data.length));

Fork/Join 套路

import java.util.concurrent.*;

class DivideTask extends RecursiveTask<Long> {
    private final long[] arr;
    private final int start, end;
    private static final int THRESHOLD = 10_000;

    DivideTask(long[] arr, int start, int end) {
        this.arr = arr; this.start = start; this.end = end;
    }

    protected Long compute() {
        int len = end - start;
        if (len <= THRESHOLD) {
            long sum = 0;
            for (int i = start; i < end; i++) sum += arr[i];
            return sum;
        }
        int mid = start + len / 2;
        DivideTask left = new DivideTask(arr, start, mid);
        DivideTask right = new DivideTask(arr, mid, end);
        left.fork();
        long r = right.compute();
        long l = left.join();
        return l + r;
    }
}

public class UseForkJoin {
    public static void main(String[] args) {
        long[] data = new long[5_000_000];
        ForkJoinPool pool = ForkJoinPool.commonPool();
        long result = pool.invoke(new DivideTask(data, 0, data.length));
        System.out.println(result);
    }
}

同步集合 vs 并发集合

import java.util.*;

List<String> syncList = Collections.synchronizedList(new ArrayList<>());
syncList.add("a");

// 迭代需要外部同步
synchronized (syncList) {
    for (String s : syncList) { /* 处理 */ }
}

Map<String, Integer> syncMap = Collections.synchronizedMap(new HashMap<>());

// ConcurrentHashMap：线程安全的 HashMap
ConcurrentHashMap concurrentMap = new ConcurrentHashMap<>();
concurrentMap.putIfAbsent("key", 100);

// CopyOnWriteArrayList：写时复制的 List，适合读多写少
CopyOnWriteArrayList list = new CopyOnWriteArrayList<>();

// 阻塞队列
ArrayBlockingQueue arrayQueue = new ArrayBlockingQueue<>(100);
LinkedBlockingQueue linkedQueue = new LinkedBlockingQueue<>();
PriorityBlockingQueue priorityQueue = new PriorityBlockingQueue<>();

在高并发场景优先选择并发集合，降低锁竞争与迭代风险。

无锁编程与原子类深入

CAS（Compare-And-Swap）是无锁编程的核心，Java 原子类基于 CAS 实现高性能的线程安全操作。

🔄 CAS 原理详解

⚡ 原子类实战应用

public class AtomicExample {
    // 基本类型原子类
    private AtomicInteger atomicInt = new AtomicInteger(0);
    private AtomicLong atomicLong = new AtomicLong(0L);
    private AtomicBoolean atomicBool = new AtomicBoolean(false);

    // 引用类型原子类
    private AtomicReference atomicRef = new AtomicReference<>();

    public void demonstrateAtomicOperations() {
        // 原子递增
        int result1 = atomicInt.incrementAndGet(); // 先加1再返回
        int result2 = atomicInt.getAndIncrement(); // 先返回再加1

        // 原子比较并设置（CAS）
        boolean success = atomicInt.compareAndSet(5, 100); // 如果当前值是5，则设置为100

        // 原子操作链
        atomicInt.updateAndGet(x -> x * 2); // 原子地执行函数操作
        atomicInt.accumulateAndGet(10, (x, y) -> x + y); // 原子地累加
    }

    // CAS 操作示例：只有当前值为期望值时才更新
    public boolean incrementIfEven() {
        int current, next;
        do {
            current = atomicInt.get();
            if (current % 2 != 0) {
                return false; // 如果是奇数，不执行更新
            }
            next = current + 1;
        } while (!atomicInt.compareAndSet(current, next));

        return true;
    }
}

原子类提供了高性能的线程安全操作，特别适合计数器、状态标志等场景。

CompletableFuture 深度编程

CompletableFuture 是 Java 8 引入的革命性异步编程工具，支持链式调用和组合操作。

🎯 为什么需要 CompletableFuture？

⚡ 链式调用示例

public class CompletableFutureChaining {
    public static void main(String[] args) throws Exception {
        CompletableFuture future = CompletableFuture
            .supplyAsync(() -> "用户数据")

            // 转换操作
            .thenApply(data -> {
                System.out.println("1. 处理: " + data);
                return data + " -> 增加权限";
            })

            // 消费操作
            .thenAccept(processedData -> {
                System.out.println("2. 消费: " + processedData);
            })

            // 后续操作
            .thenRun(() -> {
                System.out.println("3. 清理资源");
            });

        // 等待完成
        future.get();
    }
}

链式调用让异步编程变得直观，可以清晰地表达业务流程。

JUC 最佳实践和常见陷阱

✅ 最佳实践

• 优先使用线程池：避免手动创建和管理线程
• 合理设置线程池参数：根据任务类型（CPU密集型/IO密集型）调整
• 使用合适的并发集合：根据场景选择 ConcurrentHashMap、CopyOnWriteArrayList 等
• 避免锁竞争：使用分段锁、读写锁或无锁数据结构
• 正确处理异常：在异步任务中妥善处理异常
• 资源清理：确保 ExecutorService 等资源被正确关闭

❌ 常见陷阱

// 陷阱1：线程池未关闭导致资源泄漏
public class BadPractice1 {
    public void badMethod() {
        ExecutorService executor = Executors.newFixedThreadPool(10);
        executor.submit(() -> System.out.println("任务执行"));
        // 忘记调用 executor.shutdown()
    }
}

// 陷阱2：错误地使用 synchronized 替代 JUC 工具
public class BadPractice2 {
    private final List list = new ArrayList<>();

    public void addItem(String item) {
        synchronized (list) { // 不如使用 CopyOnWriteArrayList
            list.add(item);
        }
    }
}

// 陷阱3：阻塞操作忘记处理 InterruptedException
public class BadPractice3 {
    public void badSleep() {
        try {
            Thread.sleep(1000); // 可能被中断但未处理
        } catch (InterruptedException e) {
            // 错误：忽略中断信号
        }
    }
}

🔍 性能调优建议

// 性能调优示例
public class PerformanceOptimization {

    // CPU 密集型任务：线程数 = CPU 核心数
    public ExecutorService cpuIntensivePool() {
        int cores = Runtime.getRuntime().availableProcessors();
        return new ThreadPoolExecutor(
            cores, cores, 0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<>()
        );
    }

    // IO 密集型任务：线程数可以适当增加
    public ExecutorService ioIntensivePool() {
        int cores = Runtime.getRuntime().availableProcessors();
        return new ThreadPoolExecutor(
            cores * 2, cores * 4, 60L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(1000)
        );
    }
}

掌握 JUC 不仅能提升程序性能，还能写出更健壮、可维护的并发代码。实践是最好的老师，多写多练才能精通。