Java并发

引言：从硬件到Java的抽象 在上一篇文章中，我们学习了CPU缓存、MESI协议、False Sharing等硬件层面的并发机制。但问题来了： 作为Java开发者，我们需要关心这些底层细节吗？ 看这个例子： public class VisibilityProblem { private boolean flag = false; private int data = 0; // 线程1 public void writer() { data = 42; // 1 flag = true; // 2 } // 线程2 public void reader() { if (flag) { // 3 int result = data; // 4 System.out.println(result); // 可能输出0！ } } } 三个疑问： 为什么线程2可能看不到flag = true？（可见性问题） 为什么即使看到flag = true，也可能读到data = 0？（有序性问题） 这个行为在不同CPU上是否一致？（平台差异） Java的解决方案：Java内存模型（JMM） JMM就像一份**“合同”**： 对JVM实现者：规定必须遵守的行为规范 对Java程序员：提供统一的并发语义保证 屏蔽平台差异：无论x86、ARM还是RISC-V，行为一致 本篇文章，我们将深入理解JMM的设计理念和工作机制。 一、为什么需要Java内存模型？ 1.1 硬件层面的挑战 不同CPU架构的内存模型差异巨大： CPU架构 内存模型 Store-Load重排序 说明 x86/x64 TSO (Total Store Order) 否 较强的内存模型，大部分情况有序 ARM Weak 是 弱内存模型，重排序激进 RISC-V RVWMO 是 类似ARM，弱内存模型 PowerPC Weak 是 弱内存模型 示例：同样的Java代码，在x86上可能正常，在ARM上可能出错！ ...

引言：一个不准的计数器 public class Counter { private int count = 0; public void increment() { count++; // 看似简单的一行代码 } public int getCount() { return count; } } 测试：100个线程各执行1000次increment() Counter counter = new Counter(); for (int i = 0; i < 100; i++) { new Thread(() -> { for (int j = 0; j < 1000; j++) { counter.increment(); } }).start(); } // 期望：100,000 // 实际：95,732（每次不同！） 为什么？ 一、count++不是原子操作 1.1 字节码分析 count++; // 一行代码 实际执行： 1. LOAD count // 读取count的值到寄存器 2. ADD 1 // 寄存器值+1 3. STORE count // 写回count 1.2 时序问题 时间 线程A 线程B count t1 LOAD count=0 0 t2 LOAD count=0 0 t3 ADD (寄存器=1) 0 t4 ADD (寄存器=1) 0 t5 STORE count=1 1 t6 STORE count=1 1 期望结果：2 实际结果：1 结论：非原子操作 + 多线程 = 数据竞争 ...

引言：一个令人困惑的性能问题 某电商系统在双十一期间做性能压测，工程师发现一个奇怪的现象： // 订单统计类 public class OrderStats { private volatile long successCount = 0; // 成功订单数 private volatile long failCount = 0; // 失败订单数 public void recordSuccess() { successCount++; } public void recordFail() { failCount++; } } 在32核机器上，当16个线程同时调用recordSuccess()时，性能只有单线程的2倍，远低于理论值16倍！ 更诡异的是：当把两个字段分开到不同的类中，性能提升了8倍！ // 拆分后 public class SuccessStats { private volatile long successCount = 0; // 单独一个类 } public class FailStats { private volatile long failCount = 0; // 单独一个类 } 为什么会这样？这就是**False Sharing（伪共享）**问题，根源在于CPU缓存的工作机制。 要理解这个问题，我们需要从CPU缓存的硬件架构说起。 一、为什么需要CPU缓存？ 1.1 计算机存储层次结构 计算机的存储系统是一个金字塔结构： 速度快 容量小 价格贵 ↓ ┌───────────────┐ │ 寄存器 │ ~1ns (几十个) ├───────────────┤ │ L1 Cache │ ~2ns (32KB-128KB) ├───────────────┤ │ L2 Cache │ ~10ns (256KB-1MB) ├───────────────┤ │ L3 Cache │ ~40ns (8MB-64MB) ├───────────────┤ │ 主内存 │ ~100ns (8GB-128GB) ├───────────────┤ │ SSD硬盘 │ ~100μs (256GB-2TB) ├───────────────┤ │ 机械硬盘 │ ~10ms (1TB-10TB) └───────────────┘ ↑ 速度慢 容量大 价格便宜 1.2 性能差距有多大？ 如果把CPU访问L1缓存的时间比作1秒，那么： ...

一、为什么需要JMH？ 1.1 传统性能测试的问题 // ❌ 错误的性能测试 public class BadPerformanceTest { public static void main(String[] args) { long start = System.currentTimeMillis(); for (int i = 0; i < 1_000_000; i++) { // 测试代码 } long end = System.currentTimeMillis(); System.out.println("耗时：" + (end - start) + "ms"); } } 问题： ❌ JVM预热不足：JIT未优化 ❌ GC影响：测试期间GC ❌ 死代码消除：编译器优化掉无用代码 ❌ 缓存影响：CPU缓存命中率 ❌ 采样不足：单次测试不准确 1.2 JMH的优势 // ✅ 使用JMH进行科学测试 @BenchmarkMode(Mode.AverageTime) @OutputTimeUnit(TimeUnit.NANOSECONDS) @State(Scope.Thread) public class JMHTest { @Benchmark public int test() { return 1 + 1; } } JMH特性： ...

引言：生产者-消费者问题 经典场景：生产者生产数据→放入队列→消费者从队列取数据。 问题： 队列满时，生产者如何等待？ 队列空时，消费者如何等待？ 如何通知对方继续工作？ 这就是 wait/notify 机制要解决的问题。 一、wait/notify基础 1.1 三个核心方法 public class Object { // 释放锁，进入等待队列 public final void wait() throws InterruptedException {} // 释放锁，等待指定时间 public final void wait(long timeout) throws InterruptedException {} // 唤醒一个等待线程 public final void notify() {} // 唤醒所有等待线程 public final void notifyAll() {} } 关键约束：必须在 synchronized 块中调用！ 1.2 工作原理 对象监视器（Monitor）结构 ┌─────────────────────────────────────┐ │ Object Monitor │ │ │ │ Entry Set（入口队列） │ │ ┌───────┐ ┌───────┐ │ │ │线程2 │ │线程3 │ ← BLOCKED │ │ └───────┘ └───────┘ │ │ ↓ │ │ ┌────────────┐ │ │ │ Owner │ ← RUNNABLE │ │ │ (线程1) │ │ │ └────────────┘ │ │ ↓ wait() │ │ Wait Set（等待队列） │ │ ┌───────┐ ┌───────┐ │ │ │线程4 │ │线程5 │ ← WAITING │ │ └───────┘ └───────┘ │ │ ↑ notify() │ └─────────────────────────────────────┘ 二、生产者-消费者实现 2.1 标准实现 class SharedQueue { private Queue<Integer> queue = new LinkedList<>(); private int capacity = 5; // 生产 public synchronized void produce(int value) throws InterruptedException { while (queue.size() == capacity) { // ← 用while，不是if System.out.println("队列满，生产者等待"); wait(); // 释放锁，进入等待 } queue.offer(value); System.out.println("生产: " + value); notifyAll(); // 唤醒消费者 } // 消费 public synchronized int consume() throws InterruptedException { while (queue.isEmpty()) { // ← 用while，不是if System.out.println("队列空，消费者等待"); wait(); } int value = queue.poll(); System.out.println("消费: " + value); notifyAll(); // 唤醒生产者 return value; } } 2.2 为什么用while不用if？ // ❌ 错误：使用if if (queue.isEmpty()) { wait(); // 被唤醒后直接往下执行 } int value = queue.poll(); // ← 可能queue仍然为空！ // ✅ 正确：使用while while (queue.isEmpty()) { wait(); // 被唤醒后重新检查条件 } int value = queue.poll(); // 确保queue不为空 原因：虚假唤醒（Spurious Wakeup） ...

一、JDK自带工具 1.1 jps：查看Java进程 # 列出所有Java进程 jps -l # 输出示例： # 12345 com.example.MyApplication # 67890 org.apache.catalina.startup.Bootstrap # 参数说明： # -l：显示完整类名 # -v：显示JVM参数 # -m：显示main方法参数 1.2 jstack：线程堆栈分析 # 1. 导出线程堆栈 jstack <pid> > threads.txt # 2. 检测死锁 jstack <pid> | grep -A 20 "Found one Java-level deadlock" # 3. 查看线程数 jstack <pid> | grep "java.lang.Thread.State" | wc -l # 4. 查看BLOCKED线程 jstack <pid> | grep "java.lang.Thread.State: BLOCKED" -A 5 实战案例： # 场景1：CPU 100%排查 # 1. 找到Java进程 jps -l # 2. 找到CPU占用高的线程 top -Hp <pid> # Linux # 输出：线程ID 12345占用CPU 90% # 3. 转换线程ID为16进制 printf "%x\n" 12345 # 输出：0x3039 # 4. 查找对应线程堆栈 jstack <pid> | grep -A 50 "0x3039" # 场景2：死锁排查 jstack <pid> | grep -A 50 "Found one Java-level deadlock" 1.3 jconsole：可视化监控 # 启动jconsole jconsole # 或连接远程JVM jconsole <hostname>:<port> 核心功能： ...

引言：停止线程的困境 假设你正在开发一个下载工具，用户点击"取消"按钮，应该如何停止下载线程？ // 某个下载任务正在执行 Thread downloadThread = new Thread(() -> { while (true) { downloadData(); // 持续下载 } }); downloadThread.start(); // 用户点击"取消"按钮 // ❌ 如何停止这个线程？ downloadThread.stop(); // 已废弃，不能用！ 为什么Thread.stop()被废弃？如何正确停止线程？ 一、为什么不能用stop()？ 1.1 stop()的问题 public class StopProblemDemo { private int balance = 1000; // 银行账户余额 public synchronized void transfer(int amount) { balance -= amount; // 步骤1：扣款 // 如果在这里被stop()... balance += amount; // 步骤2：到账 } public static void main(String[] args) throws Exception { StopProblemDemo account = new StopProblemDemo(); Thread thread = new Thread(() -> { account.transfer(500); }); thread.start(); Thread.sleep(1); // 让线程执行一半 thread.stop(); // ← 强制停止 System.out.println("余额: " + account.balance); // 500？1000？ } } 问题： ...

一、线程栈相关参数 1.1 -Xss：线程栈大小 # 默认值（不同平台不同） # Linux/macOS：1MB # Windows：根据虚拟内存 # 设置线程栈大小 -Xss512k # 设置为512KB -Xss1m # 设置为1MB（推荐） -Xss2m # 设置为2MB 作用： 控制每个线程的栈内存大小 栈用于存储局部变量、方法调用链 调优建议： // ❌ 栈太小：StackOverflowError -Xss128k // 递归深度受限 // ❌ 栈太大：可创建线程数少 -Xss10m // 内存浪费，线程数受限 // ✅ 推荐配置 -Xss1m // 平衡性能和内存 计算可创建线程数： 最大线程数 = (最大进程内存 - JVM堆内存 - JVM非堆内存) / 线程栈大小 例如： - 最大进程内存：4GB - JVM堆内存：2GB - JVM非堆内存：512MB - 线程栈大小：1MB 最大线程数 = (4GB - 2GB - 512MB) / 1MB ≈ 1536个线程 二、锁优化相关参数 2.1 偏向锁 # 启用偏向锁（JDK 8默认启用） -XX:+UseBiasedLocking # 禁用偏向锁 -XX:-UseBiasedLocking # 偏向锁延迟启动时间（默认4秒） -XX:BiasedLockingStartupDelay=0 # 立即启用 偏向锁原理： ...

引言：一道高频面试题 面试官：请说一下Java中创建线程有哪几种方式？ 候选人A：继承Thread类、实现Runnable接口…还有吗？ 面试官：还有吗？Callable算吗？线程池算吗？它们有什么区别？ 大部分候选人能说出前两种，但很少有人能说清楚： 为什么推荐实现Runnable而不是继承Thread？ Callable和Runnable的本质区别是什么？ 线程池是怎么创建线程的？ start()和run()到底有什么区别？ 今天我们从第一性原理出发，彻底搞懂这些问题。 一、方式1：继承Thread类 1.1 基本用法 public class MyThread extends Thread { @Override public void run() { System.out.println("线程执行: " + Thread.currentThread().getName()); } public static void main(String[] args) { MyThread thread = new MyThread(); thread.start(); // 启动线程 } } 输出： 线程执行: Thread-0 1.2 原理分析 // Thread类的核心结构 public class Thread implements Runnable { private Runnable target; // 任务对象 public Thread() { // 空构造 } public Thread(Runnable target) { this.target = target; // 传入任务 } @Override public void run() { if (target != null) { target.run(); // 执行任务 } // 子类可以重写这个方法 } public synchronized void start() { // 1. 检查状态 if (threadStatus != 0) throw new IllegalThreadStateException(); // 2. 加入线程组 group.add(this); // 3. 调用native方法启动线程 boolean started = false; try { start0(); // ← native方法，创建操作系统线程 started = true; } finally { // ... } } private native void start0(); // JNI调用，创建OS线程 } 关键点： ...

一、为什么要监控线程池？ 1.1 常见线程池问题 // 线程池配置不当，导致的问题： // 1. 线程数过小：任务堆积 ThreadPoolExecutor pool = new ThreadPoolExecutor( 2, 2, // ❌ 核心线程数太少 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(1000) ); // 结果：任务大量排队，响应慢 // 2. 队列无界：内存溢出 ThreadPoolExecutor pool = new ThreadPoolExecutor( 10, 20, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>() // ❌ 无界队列 ); // 结果：任务无限堆积，OOM // 3. 拒绝策略不当：任务丢失 ThreadPoolExecutor pool = new ThreadPoolExecutor( 10, 20, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadPoolExecutor.DiscardPolicy() // ❌ 静默丢弃 ); // 结果：任务丢失，业务异常 监控目的： ✅ 发现性能瓶颈 ✅ 预防资源耗尽 ✅ 优化参数配置 ✅ 及时告警 二、核心监控指标 2.1 线程池状态指标 ThreadPoolExecutor executor = new ThreadPoolExecutor( 10, 20, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(1000) ); // 1. 核心线程数 int corePoolSize = executor.getCorePoolSize(); // 2. 最大线程数 int maximumPoolSize = executor.getMaximumPoolSize(); // 3. 当前线程数 int poolSize = executor.getPoolSize(); // 4. 活跃线程数（正在执行任务的线程数） int activeCount = executor.getActiveCount(); // 5. 历史最大线程数 int largestPoolSize = executor.getLargestPoolSize(); // 6. 任务总数 long taskCount = executor.getTaskCount(); // 7. 已完成任务数 long completedTaskCount = executor.getCompletedTaskCount(); // 8. 队列中任务数 int queueSize = executor.getQueue().size(); // 9. 队列剩余容量 int remainingCapacity = executor.getQueue().remainingCapacity(); System.out.println("核心线程数：" + corePoolSize); System.out.println("最大线程数：" + maximumPoolSize); System.out.println("当前线程数：" + poolSize); System.out.println("活跃线程数：" + activeCount); System.out.println("队列中任务数：" + queueSize); System.out.println("已完成任务数：" + completedTaskCount); 2.2 关键指标说明 指标 说明 正常范围 异常信号 活跃线程数/当前线程数 线程利用率 60%-80% >90%：线程不足 队列中任务数 任务积压情况 <50% >80%：任务堆积 任务完成速率 处理能力 稳定 持续下降：性能问题 拒绝任务数 容量溢出 0 >0：需要扩容 三、线程池监控实战 3.1 自定义监控线程池 public class MonitoredThreadPoolExecutor extends ThreadPoolExecutor { private final AtomicLong totalExecutionTime = new AtomicLong(0); private final AtomicLong rejectedTaskCount = new AtomicLong(0); public MonitoredThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) { super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, new MonitoredRejectedExecutionHandler()); } @Override protected void beforeExecute(Thread t, Runnable r) { super.beforeExecute(t, r); // 任务执行前的逻辑 } @Override protected void afterExecute(Runnable r, Throwable t) { super.afterExecute(r, t); // 任务执行后的逻辑 if (t != null) { System.err.println("任务执行异常：" + t.getMessage()); } // 统计执行时间（需要在Runnable中记录） } @Override protected void terminated() { super.terminated(); System.out.println("线程池已关闭"); printStatistics(); } // 自定义拒绝策略：记录拒绝次数 private class MonitoredRejectedExecutionHandler implements RejectedExecutionHandler { @Override public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { rejectedTaskCount.incrementAndGet(); System.err.println("任务被拒绝！拒绝次数：" + rejectedTaskCount.get()); // 告警逻辑 if (rejectedTaskCount.get() % 100 == 0) { System.err.println("⚠️ 告警：已拒绝 " + rejectedTaskCount.get() + " 个任务！"); } // 降级逻辑：调用者线程执行 r.run(); } } // 打印统计信息 public void printStatistics() { System.out.println("========== 线程池统计 =========="); System.out.println("核心线程数：" + getCorePoolSize()); System.out.println("最大线程数：" + getMaximumPoolSize()); System.out.println("当前线程数：" + getPoolSize()); System.out.println("活跃线程数：" + getActiveCount()); System.out.println("历史最大线程数：" + getLargestPoolSize()); System.out.println("任务总数：" + getTaskCount()); System.out.println("已完成任务数：" + getCompletedTaskCount()); System.out.println("队列中任务数：" + getQueue().size()); System.out.println("拒绝任务数：" + rejectedTaskCount.get()); System.out.println("================================"); } } 3.2 定时监控 public class ThreadPoolMonitor { public static void startMonitoring(ThreadPoolExecutor executor) { ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1); scheduler.scheduleAtFixedRate(() -> { System.out.println("========== 线程池监控 =========="); System.out.println("当前时间：" + LocalDateTime.now()); System.out.println("核心线程数：" + executor.getCorePoolSize()); System.out.println("最大线程数：" + executor.getMaximumPoolSize()); System.out.println("当前线程数：" + executor.getPoolSize()); System.out.println("活跃线程数：" + executor.getActiveCount()); System.out.println("队列大小：" + executor.getQueue().size()); System.out.println("已完成任务：" + executor.getCompletedTaskCount()); // 计算线程利用率 double threadUtilization = (double) executor.getActiveCount() / executor.getPoolSize() * 100; System.out.printf("线程利用率：%.2f%%\n", threadUtilization); // 计算队列使用率 BlockingQueue<Runnable> queue = executor.getQueue(); int queueCapacity = queue.size() + queue.remainingCapacity(); double queueUtilization = (double) queue.size() / queueCapacity * 100; System.out.printf("队列使用率：%.2f%%\n", queueUtilization); // 告警逻辑 if (threadUtilization > 90) { System.err.println("⚠️ 告警：线程利用率过高！"); } if (queueUtilization > 80) { System.err.println("⚠️ 告警：队列积压严重！"); } System.out.println("================================\n"); }, 0, 5, TimeUnit.SECONDS); // 每5秒监控一次 } public static void main(String[] args) { ThreadPoolExecutor executor = new ThreadPoolExecutor( 10, 20, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100) ); // 启动监控 startMonitoring(executor); // 提交任务... } } 四、线程池参数调优 4.1 核心线程数调优 // CPU密集型任务 int cpuCount = Runtime.getRuntime().availableProcessors(); int corePoolSize = cpuCount + 1; // N + 1 // I/O密集型任务 int corePoolSize = cpuCount * 2; // 2N // 混合型任务（推荐公式） // corePoolSize = N * (1 + WT/ST) // N = CPU核心数 // WT = 等待时间（I/O时间） // ST = 计算时间（CPU时间） // 例如： // CPU核心数：8 // 等待时间：90ms（I/O） // 计算时间：10ms（CPU） // corePoolSize = 8 * (1 + 90/10) = 8 * 10 = 80 int corePoolSize = cpuCount * (1 + waitTime / computeTime); 4.2 队列选择与容量 // 1. LinkedBlockingQueue（无界队列） // 优点：无限容量 // 缺点：可能OOM // 适用：内存充足，任务数可控 BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(); // 2. LinkedBlockingQueue（有界队列） // 优点：防止OOM // 缺点：任务过多会拒绝 // 适用：需要控制内存 BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(1000); // 3. ArrayBlockingQueue（有界队列） // 优点：数组实现，性能好 // 缺点：容量固定 // 适用：高性能场景 BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(1000); // 4. SynchronousQueue（无缓冲队列） // 优点：直接交付，吞吐量高 // 缺点：无缓冲，容易拒绝 // 适用：任务执行快，maximumPoolSize大 BlockingQueue<Runnable> queue = new SynchronousQueue<>(); // 5. PriorityBlockingQueue（优先级队列） // 优点：支持优先级 // 缺点：性能开销大 // 适用：需要优先级调度 BlockingQueue<Runnable> queue = new PriorityBlockingQueue<>(); 队列容量计算： ...