并发编程的陷阱：六年里，我见过的那些诡异的线程安全问题
AI摘要 作者以六年踩坑经历，揭示并发“幽灵”：SimpleDateFormat共享状态、Check-then-Act空档、迭代器隐式fail-fast、volatile误当原子锁。教训：单线程测试不可信，共享可变必审访问路径，优先使用JUC工具，理解内存语义，方得线程安全。 此内容根据文章生成，仅用于文章内容的解释与总结 如果说编程是教会计算机思考，那并发编程就是教会它“一心多用”。而让我真正对并发产生敬畏的，不是书本上的概念，而是在自己代码里撞见的那些诡异瞬间。它们不像生产事故那样轰轰烈烈，却像程序世界里悄无声息的“幽灵”，一次次挑战我对代码确定性的信仰。我记得早期，我天真地以为线程安全就是synchronized。直到我踩进一个又一个陷阱，才明白并发问题的诡异在于：它绝大多数时候都运行良好，就像一颗藏在代码深处的定时炸弹，只等一个完美的、难以复现的条件组合来引爆。陷阱一：看似“局部”的变量，却引来了“全局”的灾难场景： 一个简单的日期格式化工具类。我写了一个public static String formatDate(Date date)方法，内部使用了SimpleDateFormat实例。在我的单元测试里，百分百通过。但在一个压测脚本中，偶尔会出现格式化后的日期字符串错乱，甚至抛出NumberFormatException。排查过程：轻敌阶段： 首先怀疑是测试数据问题，反复检查，无果。困惑阶段： 在IDE里单步调试，一切正常。这让我无比沮丧。顿悟时刻： 当我用多个线程同时调用这个工具类时，幽灵出现了。日期串了，时间错了。我这才意识到，SimpleDateFormat内部维护了一个Calendar对象用于计算。当线程A正在格式化一个日期，还没完成时，线程B闯了进来，重置了Calendar对象，然后把自己的日期塞了进去。最终，A线程拿到的就是被B“污染”后的结果。教训：我犯了一个经典错误：认为没有“写”操作就是线程安全的。我误以为format方法只是“读”传入的Date对象。但实际上，SimpleDateFormat的内部状态（那个Calendar对象）在格式化过程中被剧烈地“写”改了。任何一个有状态（尤其是可变状态）的对象，如果被多个线程共享，就必须考虑线程安全问题。解决方案：最直接： 每次调用都new SimpleDateFormat()。性能差，但简单有效。线程隔离： 使用ThreadLocal<SimpleDateFormat>，为每个线程分配一个独立的实例。这是性能和安全的良好折衷。拥抱新时代： 换成Java 8引入的DateTimeFormatter，它被明确设计为不可变且线程安全的。陷阱二：“检查-执行”的致命间隙场景： 实现一个简单的缓存MyCache。逻辑是：如果缓存里有，直接返回；如果没有，就去数据库查，然后放入缓存。我写了类似这样的代码：public Object getData(String key) { Object data = cache.get(key); if (data == null) { // 检查通过，认为需要加载 data = loadDataFromDB(key); // 这是一个耗时的操作 cache.put(key, data); // 执行放入操作 } return data; }在低并发下完美工作。但模拟高并发访问时，数据库连接池偶尔会报超限错误。排查过程：日志显示，对于同一个key，在极短的时间内，loadDataFromDB这个耗时操作被执行了多次。这太反直觉了！我的if (data == null)检查是干什么用的？幽灵再现：想象一下，线程A和线程B同时调用getData(“key1”)。A检查缓存，key1不存在。A开始执行耗时的loadDataFromDB(“key1”)。就在A查询数据库的这段时间里，线程B也来检查缓存。此时A还没把结果放进去，所以B也检查到key1不存在。B也开始了loadDataFromDB(“key1”)。结果就是，数据库被同一个查询轰炸了两次，缓存里也被重复放入了两次（虽然结果一样，但浪费了资源）。教训：这就是典型的检查-执行（Check-then-Act） 竞态条件。在“检查”（data == null）和“执行”（cache.put）之间，存在一个时间窗口，其他线程可以趁虚而入。synchronized可以解决，但粒度太粗，影响性能。解决方案：使用ConcurrentHashMap的putIfAbsent方法，或者更优雅的computeIfAbsent方法。它能保证对于同一个key，加载数据的函数只被执行一次。这才是真正的原子操作。public Object getData(String key) { return cache.computeIfAbsent(key, k -> loadDataFromDB(k)); }陷阱三：隐式的迭代器陷阱场景： 一个后台任务，需要遍历一个List，并根据条件删除一些元素。我写出了这样的代码：for (String item : list) { // 这里使用了增强for循环（语法糖，底层是Iterator） if (shouldRemove(item)) { list.remove(item); // 在遍历过程中直接删除 } }在单线程测试里，删得干干净净。但当一个线程遍历，另一个线程同时向这个List添加元素时，程序在测试中偶尔会神秘地抛出ConcurrentModificationException。幽灵的根源：增强for循环背后使用的是Iterator。Iterator在工作时，会期望它正在遍历的集合结构不发生“意外改变”。它内部有一个计数器modCount，记录集合的修改次数。当Iterator被创建后，它会记住当前的modCount值。在每次next()调用时，它会检查当前的modCount是否和它记住的值一致。如果不一致（说明有其他线程或方式修改了集合结构），它就认为发生了“并发修改”，立即抛出异常。教训：任何非线程安全的集合类（如ArrayList, HashMap），在被一个线程迭代时，如果被另一个线程修改（增、删），都可能导致ConcurrentModificationException。即使在单线程中，用Iterator遍历时直接调用集合的remove()方法也会抛出此异常，必须使用Iterator.remove()。解决方案：遍历前复制： List<String> copy = new ArrayList<>(list);然后遍历copy，操作原始的list。简单，但内存开销大。使用线程安全集合： 比如CopyOnWriteArrayList。它在修改时（增、删）会复制整个底层数组，遍历操作在旧数组上进行，因此不会受修改影响。适用于读多写少的场景。显式加锁： 在遍历和修改的代码块前后加同一把锁，保证互斥。陷阱四：volatile的错觉——它不保证原子性场景： 一个简单的计数器，private volatile int count = 0;，有多个线程执行count++。我天真地以为加了volatile，就能保证最终结果是正确的。幽灵的数字：跑完测试，结果值总是小于理论值。为什么？count++这个操作，看起来是一行代码，但底层是三个步骤：从主内存读取count的当前值到线程工作内存。在工作内存中给这个值加1。将新值写回主内存。volatile只能保证变量的可见性（一个线程修改后，新值立即对其他线程可见）和禁止指令重排序。但它不保证复合操作的原子性。幽灵再现：线程A和B同时读取count，值都是0。A计算0+1=1。B也计算0+1=1。（因为B在读的时候，A还没把1写回去）A和B分别将1写回主内存。结果，两次++操作，count只增加了1。教训：volatile不是万能的。对于count++这种“读-改-写”复合操作，需要用synchronized或ReentrantLock保证原子性，或者直接使用AtomicInteger。结语：敬畏之心并发编程，最大的陷阱，是“想当然”的思维惰性。不要相信你的单线程测试。 并发问题在99.9%的时间里都会伪装得很好。养成条件反射： 看到共享的可变对象，立刻思考它的访问路径。优先使用高级工具： 多用java.util.concurrent包下的类（ConcurrentHashMap, AtomicXXX, CountDownLatch等），它们由大师编写，远比我们自己实现的可靠。理解底层原理： 明白synchronized、volatile、final的内存语义，才能真正用好它们。与并发幽灵的斗争，是一场心智的磨砺。它逼着你用最严苛的眼光去审视每一行代码，去思考时间线上无数种可能的交错。这种训练所带来的严谨，最终会渗透到你编程的每一个角落，让你成长为一个更可靠的工程师。
Java并发编程的“坑”与“术”：六年实战经验总结的避坑指南
AI摘要 六年踩坑总结：SimpleDateFormat共享致错乱、synchronized组合非原子、volatile不保证++安全；用ThreadLocal/DateTimeFormatter、ConcurrentHashMap原子方法、Atomic类即可避坑，敬畏并发、多线程测试、代码审查是铁律。 此内容根据文章生成，仅用于文章内容的解释与总结 写了几年Java，尤其是开始接触稍具规模的项目后，并发代码就成了绕不开的坎。它不像空指针异常那么“耿直”，直接崩溃给你看。它更像一个幽灵，大部分时间安静无害，但会在某个你意想不到的时刻，给你带来一些诡异、难以复现的问题。这篇博客，就是我六年来在平常开发中，对并发编程从“畏惧”到“理解”再到“谨慎使用”的一些总结。咱们不搞高深理论，就聊实实在在的“坑”和解决问题的“术”。坑一：那个看似“高效”的全局日期格式化器场景：刚开始学Spring，知道要减少对象创建，利用IoC容器的单例特性。于是，我“聪明”地定义了一个全局的SimpleDateFormat来格式化日志时间：@Component public class DateUtil { public static final SimpleDateFormat GLOBAL_FORMATTER = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); } // 在某个Service里到处用 String logTime = DateUtil.GLOBAL_FORMATTER.format(new Date());问题发现：在本地用main方法测试，一点问题没有。直到有一次，我写了个简单的多线程单元测试，想模拟一下并发日志记录。结果日志里的时间戳时不时就乱套了，出现了2023年变成1923年的诡异情况，甚至偶尔抛出了NumberFormatException。为啥是坑？SimpleDateFormat的内部维护了一个Calendar对象实例，它在执行format或parse方法时，会频繁地修改这个Calendar的内部状态（比如先clear，再set时间）。当多个线程同时调用同一个SimpleDateFormat实例时，线程A刚set好值，还没等输出，线程B就来个clear，数据就乱套了。这被称为非线程安全。解决之术：放弃治疗（最简单）： 每次用时在方法内部new SimpleDateFormat()。对于大多数不极端频繁调用的场景，这点开销完全可以接受，换来的是代码的清晰和安全。线程隔离（经典方案）： 使用ThreadLocal为每个线程绑定一个独立的实例，空间换时间，兼顾性能和安全。public class DateUtil { private static final ThreadLocal<SimpleDateFormat> threadLocalFormatter = ThreadLocal.withInitial(() -> new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")); public static String format(Date date) { return threadLocalFormatter.get().format(date); } // 重要！对于Web应用等使用线程池的场景，用完后最好主动remove，防止内存泄漏。 public static void clear() { threadLocalFormatter.remove(); } }拥抱新时代（最佳实践）： 直接使用Java 8引入的DateTimeFormatter。它是不可变对象，天生线程安全。public class DateUtil { // 放心大胆地声明为public static final public static final DateTimeFormatter FORMatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); public static String format(LocalDateTime dateTime) { return dateTime.format(FORMatter); } }坑二：synchronized锁得住代码，但锁不住粗心场景：在实现一个简单的内存缓存时，我写了下面这样的代码：public class SimpleCache<K, V> { private final Map<K, V> cache = new HashMap<>(); public synchronized void put(K key, V value) { cache.put(key, value); } public synchronized V get(K key) { return cache.get(key); } // 问题方法！ public synchronized V getAndUpdate(K key, Function<V, V> updater) { V value = get(key); // 1. 获取值 if (value != null) { V newValue = updater.apply(value); // 2. 基于旧值计算新值（非原子操作！） put(key, newValue); // 3. 放回缓存 } return value; } }我想实现一个“获取并更新”的操作，比如给某个计数加一。我觉得每个方法都加了synchronized，万无一失。问题发现：代码审查时，一位同事指出：假设线程A和线程B同时调用getAndUpdate来给同一个key的值+1。A执行到第2步，拿到了值1，此时CPU时间片用完，切换到B。B也完整地执行完了所有步骤，将值更新为2。然后A继续执行第3步，它基于自己拿到的旧值1计算出2，又覆盖了B的结果。最终结果变成了2，而不是预期的3。为啥是坑？我错误地认为，把多个同步方法组合起来调用，依然是一个原子操作。但实际上，getAndUpdate方法内部的三个步骤（get, calculate, put）虽然每个步骤内部是同步的，但整个组合对于外部调用者来说并不是一个原子操作。在get和put之间，锁被释放了，其他线程可以趁虚而入。解决之术：真正的原子化： 将整个复合操作在同一把锁的保护下完成，确保中间不释放锁。public V getAndUpdate(K key, Function<V, V> updater) { synchronized (this) { // 使用更明确的同步块，锁住整个复合操作 V value = get(key); if (value != null) { V newValue = updater.apply(value); put(key, newValue); } return value; } }使用并发容器： 直接使用ConcurrentHashMap，并利用其提供的原子方法。public class SimpleCache<K, V> { private final ConcurrentMap<K, V> cache = new ConcurrentHashMap<>(); // 使用compute等原子方法 public V getAndUpdate(K key, Function<V, V> updater) { return cache.computeIfPresent(key, (k, v) -> updater.apply(v)); } }这是更优雅、性能更好的解决方案。ConcurrentHashMap的compute、putIfAbsent等方法保证了单个键上操作的原子性。坑三：volatile的错觉——它不保证原子性场景：学习volatile关键字时，我知道它能保证可见性。于是我写了一个多线程累加的程序：public class Counter { private volatile int count = 0; public void increment() { count++; // 操作：读 -> 改 -> 写 } public int getCount() { return count; } }我以为count++这个操作因为volatile的存在而变得原子了。问题发现：跑个测试，开10个线程，每个线程加1000次，结果永远小于10000。为啥是坑？volatile只解决了两件事：1. 可见性；2. 禁止指令重排序。但它不保证复合操作的原子性。count++看似一行代码，在JVM层面是三个指令：读取当前值、值加1、写回新值。线程A可能在读完值后被挂起，线程B完成了完整的加1操作并写回（由于可见性，A会看到B更新后的值），但当A恢复后，它是在自己之前读到的旧值基础上加1，然后写回，这就覆盖了B的操作。解决之术：使用真正的原子类： 对于这种简单的原子操作，应该使用java.util.concurrent.atomic包下的类，如AtomicInteger。public class Counter { private final AtomicInteger count = new AtomicInteger(0); public void increment() { count.incrementAndGet(); // 底层通过CAS操作保证原子性 } }理解volatile的正确场景： volatile非常适合做状态标志位，如boolean shutdownRequested，一个线程写，其他线程读，这种简单的赋值操作是原子的，volatile保证了修改能立刻被其他线程看见。总结对待并发编程，一定要有“敬畏之心”。不要“我觉得”： 想当然地认为代码没问题，多线程环境下的行为往往反直觉。工具要用对： synchronized、volatile、Lock、原子类、并发容器，每个都有其明确的适用场景。用错工具，就像用螺丝刀去敲钉子，既费劲又危险。多写测试： 一定要编写多线程单元测试，并尝试高并发场景下的压力测试，很多坑在低并发下是暴露不出来的。代码审查： 多人互相审查代码，特别是并发相关的部分，非常有效。别人很容易发现你意识不到的盲点。并发编程是一个深水区，但也是从中级迈向高级的必经之路。希望我踩过的这些坑，能成为你前进路上的垫脚石。共勉！
AI摘要 文章以“主线程等待所有工作线程”需求为引，对比synchronized的局限，深入解析AQS核心原理、CLH队列、共享/独占模式，并逐行剖析CountDownLatch如何利用AQS的state与模板方法实现高效等待与唤醒，最终给出支持超时的自定义同步器实战，强调理解设计哲学优于死记API。 此内容根据文章生成，仅用于文章内容的解释与总结 在四年的Java开发生涯中，我曾认为synchronized就是并发编程的全部。直到遇到那个让我夜不能寐的需求：实现一个自定义的线程协同工具。正是这次经历，让我走进了AQS的世界。一、从synchronized的局限性说起去年在公司消息推送系统中，我需要实现这样一个功能：​主线程等待所有工作线程完成消息预处理后，才能开始批量推送​。我的第一版实现使用了synchronized：public class SimpleSync { private int workerCount; private final Object lock = new Object(); public void await() throws InterruptedException { synchronized(lock) { while (workerCount > 0) { lock.wait(); } } } public void countDown() { synchronized(lock) { if (workerCount > 0) { workerCount--; } if (workerCount == 0) { lock.notifyAll(); } } } }这段代码虽然能用，但在高并发场景下暴露了两个问题：​无法超时等待​：主线程可能永久阻塞​性能瓶颈​：所有线程竞争同一把锁正是这些痛点，促使我去探索更优秀的解决方案。二、AQS是什么？为什么说是并发包的核心？AQS（AbstractQueuedSynchronizer）是JDK并发包的基石。Doug Lea在设计时就定下目标：​提供一个框架，用于实现依赖于先进先出（FIFO）等待队列的阻塞锁和相关同步器​。2.1 AQS的核心理念AQS使用一个int类型的成员变量state来表示同步状态，通过内置的FIFO队列来完成资源获取线程的排队工作。这种设计巧妙地将同步器分解为两个维度：public abstract class AbstractQueuedSynchronizer { // 同步状态 private volatile int state; // 等待队列的头节点 private transient volatile Node head; // 等待队列的尾节点 private transient volatile Node tail; }2.2 模板方法模式的应用AQS使用了经典的模板方法模式，将具体的同步操作留给子类实现：// 尝试获取独占锁 protected boolean tryAcquire(int arg) { throw new UnsupportedOperationException(); } // 尝试释放独占锁 protected boolean tryRelease(int arg) { throw new UnsupportedOperationException(); }这种设计让AQS既可以实现独占锁（如ReentrantLock），也可以实现共享锁（如CountDownLatch）。三、深入AQS的CLH队列实现AQS的核心在于其队列管理机制，这直接决定了线程调度的公平性。3.1 Node节点：队列的基础单元static final class Node { // 节点状态 volatile int waitStatus; // 前驱节点 volatile Node prev; // 后继节点 volatile Node next; // 节点对应的线程 volatile Thread thread; // 指向下一个等待节点 Node nextWaiter; }每个Node节点代表一个等待线程，通过waitStatus标识线程状态（CANCELLED、SIGNAL、CONDITION等）。3.2 入队操作：真实的排队场景让我通过一个真实案例来说明入队过程。在订单处理系统中，多个线程同时竞争处理权限：public class OrderProcessor { private final Sync sync = new Sync(); public void processOrder(String orderId) { sync.acquire(1); // 尝试获取锁 try { // 处理订单逻辑 System.out.println("Processing order: " + orderId); } finally { sync.release(1); // 释放锁 } } private static class Sync extends AbstractQueuedSynchronizer { protected boolean tryAcquire(int acquires) { return compareAndSetState(0, 1); } protected boolean tryRelease(int releases) { setState(0); return true; } } }当线程A首先获取锁时，state从0变为1。此时线程B尝试获取锁，CAS操作失败，开始入队过程。四、CountDownLatch的AQS实现剖析现在回到文章开头的需求，看看CountDownLatch如何基于AQS优雅地解决问题。4.1 状态state的巧妙运用CountDownLatch使用state表示剩余的计数：public class CountDownLatch { private static final class Sync extends AbstractQueuedSynchronizer { Sync(int count) { setState(count); // 初始化状态值 } // 共享式获取同步状态 protected int tryAcquireShared(int acquires) { return (getState() == 0) ? 1 : -1; } // 共享式释放同步状态 protected boolean tryReleaseShared(int releases) { for (;;) { int c = getState(); if (c == 0) return false; int nextc = c - 1; if (compareAndSetState(c, nextc)) { return nextc == 0; // 返回是否全部释放 } } } } }4.2 await()方法的完整执行流程当主线程调用await()时：​检查状态​：如果state == 0，直接返回​状态不为0​：将当前线程加入等待队列​进入等待​：通过LockSupport.park()挂起线程// 这是我调试源码时记录的调用栈 await() → sync.acquireSharedInterruptibly(1) → tryAcquireShared(1)返回-1（需要排队） → doAcquireSharedInterruptibly(1) → parkAndCheckInterrupt() // 线程挂起4.3 countDown()方法的唤醒机制工作线程调用countDown()时的关键逻辑：public void countDown() { sync.releaseShared(1); } // 在AQS中 public final boolean releaseShared(int arg) { if (tryReleaseShared(arg)) { // 如果state减到0 doReleaseShared(); // 唤醒等待队列中的所有线程 return true; } return false; }doReleaseShared()方法会从队头开始，逐个唤醒等待的线程。五、AQS的公平性与非公平性实现理解这一点对实际应用至关重要。以ReentrantLock为例：5.1 非公平锁的"插队"机制final boolean nonfairTryAcquire(int acquires) { final Thread current = Thread.currentThread(); int c = getState(); if (c == 0) { if (compareAndSetState(0, acquires)) { // 直接尝试获取，不管队列 setExclusiveOwnerThread(current); return true; } } // ... 重入逻辑 }非公平锁在锁释放时，新来的线程可以和等待队列中的线程竞争，可能"插队"成功。5.2 公平锁的严格排队protected final boolean tryAcquire(int acquires) { if (getState() == 0) { if (!hasQueuedPredecessors() && // 检查是否有前驱节点 compareAndSetState(0, acquires)) { setExclusiveOwnerThread(current); return true; } } // ... 重入逻辑 }公平锁通过hasQueuedPredecessors()确保严格按FIFO顺序获取锁。六、实战：基于AQS实现自定义同步工具现在，我可以优雅地重构文章开头的需求：public class AdvancedLatch { private final Sync sync; public AdvancedLatch(int count) { this.sync = new Sync(count); } public void await() throws InterruptedException { sync.acquireSharedInterruptibly(1); } public boolean await(long timeout, TimeUnit unit) throws InterruptedException { return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout)); } public void countDown() { sync.releaseShared(1); } public int getCount() { return sync.getCount(); } private static final class Sync extends AbstractQueuedSynchronizer { Sync(int count) { setState(count); } protected int tryAcquireShared(int acquires) { return getState() == 0 ? 1 : -1; } protected boolean tryReleaseShared(int releases) { for (;;) { int c = getState(); if (c == 0) return false; int nextc = c - 1; if (compareAndSetState(c, nextc)) { return nextc == 0; } } } int getCount() { return getState(); } } }这个实现支持超时等待，性能远超最初的synchronized版本。七、AQS的注意事项和最佳实践7.1 避免过度自定义除非有特殊需求，否则优先使用JDK提供的同步工具。AQS的复杂性很容易引入难以发现的bug。7.2 正确实现模板方法实现tryAcquire/tryRelease时，要确保线程安全，通常需要配合CAS操作。7.3 性能考量在低竞争场景下，非公平锁性能更好；高竞争场景下，公平锁能避免线程饥饿。总结理解AQS让我对Java并发编程有了全新的认识。它不仅是技术实现，更体现了一种设计哲学：​通过合理的抽象，将复杂问题分解为可管理的部分​。从synchronized到AQS，就像从手动挡汽车换到自动挡。你仍然需要理解底层原理，但可以更专注于业务逻辑的实现。​技术成长的标志不是知道更多API，而是理解背后的设计思想​。AQS的学习曲线很陡峭，但一旦掌握，你将拥有解决复杂并发问题的能力。希望我的经验分享对你有所帮助。如果你在AQS学习过程中遇到问题，欢迎交流讨论！
AI摘要 文章通过微服务上下文传递问题，系统解析ThreadLocal原理、内存泄漏风险与最佳实践，并深入演示CompletableFuture的链式、并行、异常及超时处理，结合性能测试给出IO/计算密集型线程池配置建议，总结生产环境监控与优雅关闭要点。 此内容根据文章生成，仅用于文章内容的解释与总结 一、引言：一个微服务上下文传递的难题在微服务架构中，我们经常需要在不同线程间传递用户上下文（如用户ID、请求ID等）。最初我们使用静态Map来存储，结果遇到了严重的数据错乱问题：// ❌ 错误示例：静态Map存储上下文 public class UserContextHolder { private static final Map<String, Object> context = new ConcurrentHashMap<>(); public static void set(String key, Object value) { context.put(key, value); } public static Object get(String key) { return context.get(key); } }在多线程环境下，多个请求会相互覆盖上下文数据。这让我深刻认识到：​共享可变状态是并发编程万恶之源​。二、ThreadLocal：线程安全的局部变量2.1 ThreadLocal的核心原理ThreadLocal为每个线程提供独立的变量副本，避免了线程间的数据竞争。ThreadLocal基本使用import java.util.concurrent.TimeUnit; public class ThreadLocalDemo { // 创建ThreadLocal实例 private static final ThreadLocal<String> userContext = new ThreadLocal<>(); private static final ThreadLocal<Integer> requestIdContext = ThreadLocal.withInitial(() -> 0); public static void main(String[] args) throws InterruptedException { // 创建并启动多个线程 Thread thread1 = new Thread(new UserRequest("用户A", 1001), "Thread-1"); Thread thread2 = new Thread(new UserRequest("用户B", 1002), "Thread-2"); Thread thread3 = new Thread(new UserRequest("用户C", 1003), "Thread-3"); thread1.start(); thread2.start(); thread3.start(); // 等待所有线程执行完成 thread1.join(); thread2.join(); thread3.join(); System.out.println("所有线程执行完成"); } // 用户请求处理类 static class UserRequest implements Runnable { private final String userName; private final int requestId; public UserRequest(String userName, int requestId) { this.userName = userName; this.requestId = requestId; } @Override public void run() { try { // 设置线程局部变量 userContext.set(userName); requestIdContext.set(requestId); // 模拟业务处理 processRequest(); } finally { // 重要：必须清理ThreadLocal，防止内存泄漏 userContext.remove(); requestIdContext.remove(); } } private void processRequest() { // 获取当前线程的上下文数据 String currentUser = userContext.get(); Integer currentRequestId = requestIdContext.get(); System.out.println(Thread.currentThread().getName() + " -> 用户: " + currentUser + ", 请求ID: " + currentRequestId); // 模拟嵌套方法调用，上下文依然可以获取 nestedMethod(); } private void nestedMethod() { // 嵌套方法中依然可以获取到ThreadLocal中的数据 System.out.println(Thread.currentThread().getName() + " -> 嵌套方法获取用户: " + userContext.get()); } } }2.2 ThreadLocal的内存泄漏问题与解决方案import java.lang.ref.WeakReference; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; public class ThreadLocalMemoryLeakDemo { // 模拟大对象 static class BigObject { private final byte[] data = new byte[10 * 1024 * 1024]; // 10MB @Override protected void finalize() throws Throwable { System.out.println("BigObject被GC回收"); super.finalize(); } } public static void main(String[] args) throws InterruptedException { // 1. 演示内存泄漏的场景 demonstrateMemoryLeak(); // 等待GC System.gc(); TimeUnit.SECONDS.sleep(3); System.out.println("\n--- 分隔线 ---\n"); // 2. 演示正确使用ThreadLocal demonstrateCorrectUsage(); // 等待GC System.gc(); TimeUnit.SECONDS.sleep(3); } private static void demonstrateMemoryLeak() { System.out.println("=== 演示ThreadLocal内存泄漏 ==="); ThreadLocal<BigObject> threadLocal = new ThreadLocal<>(); ExecutorService executor = Executors.newSingleThreadExecutor(); // 在线程池线程中设置ThreadLocal executor.submit(() -> { threadLocal.set(new BigObject()); System.out.println("线程池线程设置了BigObject"); // 注意：这里没有调用remove()！ }); // 等待任务完成 executor.shutdown(); try { executor.awaitTermination(2, TimeUnit.SECONDS); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } System.out.println("线程池已关闭，但ThreadLocal中的BigObject可能还未被回收"); } private static void demonstrateCorrectUsage() { System.out.println("=== 演示ThreadLocal正确用法 ==="); ThreadLocal<BigObject> threadLocal = new ThreadLocal<>(); ExecutorService executor = Executors.newSingleThreadExecutor(); // 在线程池线程中设置ThreadLocal，并正确清理 executor.submit(() -> { try { threadLocal.set(new BigObject()); System.out.println("线程池线程设置了BigObject"); // 模拟业务处理 TimeUnit.MILLISECONDS.sleep(100); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } finally { // 关键：使用完后一定要remove！ threadLocal.remove(); System.out.println("线程池线程清理了ThreadLocal"); } }); // 等待任务完成 executor.shutdown(); try { executor.awaitTermination(2, TimeUnit.SECONDS); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } System.out.println("线程池已关闭，ThreadLocal已正确清理"); } // 3. 使用InheritableThreadLocal实现父子线程数据传递 private static void demonstrateInheritableThreadLocal() { System.out.println("=== 演示InheritableThreadLocal ==="); InheritableThreadLocal<String> inheritableThreadLocal = new InheritableThreadLocal<>(); inheritableThreadLocal.set("父线程数据"); Thread childThread = new Thread(() -> { System.out.println("子线程获取数据: " + inheritableThreadLocal.get()); }); childThread.start(); try { childThread.join(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } // 清理 inheritableThreadLocal.remove(); } }2.3 ThreadLocal在实际项目中的应用：请求上下文管理import java.util.HashMap; import java.util.Map; import java.util.UUID; /** * 请求上下文管理器 * 用于在同一个请求的整个调用链中传递上下文信息 */ public class RequestContextHolder { private RequestContextHolder() { // 工具类，禁止实例化 } // 使用ThreadLocal存储请求上下文 private static final ThreadLocal<RequestContext> REQUEST_CONTEXT = new ThreadLocal<>(); /** * 请求上下文类 */ public static class RequestContext { private final String requestId; private final long startTime; private final Map<String, Object> attributes; public RequestContext() { this.requestId = generateRequestId(); this.startTime = System.currentTimeMillis(); this.attributes = new HashMap<>(); } public RequestContext(String requestId) { this.requestId = requestId; this.startTime = System.currentTimeMillis(); this.attributes = new HashMap<>(); } private String generateRequestId() { return UUID.randomUUID().toString().replace("-", ""); } public String getRequestId() { return requestId; } public long getStartTime() { return startTime; } public long getElapsedTime() { return System.currentTimeMillis() - startTime; } public void setAttribute(String key, Object value) { attributes.put(key, value); } @SuppressWarnings("unchecked") public <T> T getAttribute(String key) { return (T) attributes.get(key); } public void removeAttribute(String key) { attributes.remove(key); } public Map<String, Object> getAllAttributes() { return new HashMap<>(attributes); } } /** * 初始化请求上下文 */ public static void init() { REQUEST_CONTEXT.set(new RequestContext()); } /** * 初始化请求上下文（指定requestId） */ public static void init(String requestId) { REQUEST_CONTEXT.set(new RequestContext(requestId)); } /** * 获取当前请求上下文 */ public static RequestContext getCurrentContext() { RequestContext context = REQUEST_CONTEXT.get(); if (context == null) { throw new IllegalStateException("请求上下文未初始化，请确保在请求入口处调用RequestContextHolder.init()"); } return context; } /** * 获取当前请求ID */ public static String getCurrentRequestId() { return getCurrentContext().getRequestId(); } /** * 设置上下文属性 */ public static void setAttribute(String key, Object value) { getCurrentContext().setAttribute(key, value); } /** * 获取上下文属性 */ @SuppressWarnings("unchecked") public static <T> T getAttribute(String key) { return getCurrentContext().getAttribute(key); } /** * 获取当前用户ID */ public static String getCurrentUserId() { return getAttribute("userId"); } /** * 设置当前用户ID */ public static void setCurrentUserId(String userId) { setAttribute("userId", userId); } /** * 获取客户端IP */ public static String getClientIp() { return getAttribute("clientIp"); } /** * 设置客户端IP */ public static void setClientIp(String clientIp) { setAttribute("clientIp", clientIp); } /** * 清理请求上下文（必须在请求结束时调用） */ public static void clear() { REQUEST_CONTEXT.remove(); } /** * 模拟业务服务类 */ public static class OrderService { public void createOrder(String productId, int quantity) { String requestId = RequestContextHolder.getCurrentRequestId(); String userId = RequestContextHolder.getCurrentUserId(); System.out.println(String.format("[%s] 用户 %s 创建订单 - 商品: %s, 数量: %d", requestId, userId, productId, quantity)); // 模拟调用其他服务 InventoryService inventoryService = new InventoryService(); inventoryService.deductInventory(productId, quantity); } } /** * 模拟库存服务 */ public static class InventoryService { public void deductInventory(String productId, int quantity) { String requestId = RequestContextHolder.getCurrentRequestId(); System.out.println(String.format("[%s] 扣减库存 - 商品: %s, 数量: %d", requestId, productId, quantity)); // 这里可以继续传递上下文 } } /** * 测试用例 */ public static void main(String[] args) { // 模拟HTTP请求处理 simulateHttpRequest(); // 模拟另一个并发请求 new Thread(RequestContextHolder::simulateHttpRequest).start(); } private static void simulateHttpRequest() { try { // 1. 请求开始时初始化上下文 RequestContextHolder.init(); // 2. 设置请求相关属性 RequestContextHolder.setCurrentUserId("user_" + Thread.currentThread().getId()); RequestContextHolder.setClientIp("192.168.1." + Thread.currentThread().getId()); // 3. 执行业务逻辑 OrderService orderService = new OrderService(); orderService.createOrder("product_001", 2); // 4. 打印当前请求耗时 long elapsed = RequestContextHolder.getCurrentContext().getElapsedTime(); System.out.println(String.format("[%s] 请求处理完成，耗时: %dms", RequestContextHolder.getCurrentRequestId(), elapsed)); } finally { // 5. 请求结束时清理上下文（重要！） RequestContextHolder.clear(); } } }三、CompletableFuture：异步编程的利器3.1 CompletableFuture基础用法import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; public class CompletableFutureBasicDemo { // 自定义线程池（生产环境建议使用） private static final ExecutorService customExecutor = Executors.newFixedThreadPool(4, r -> { Thread thread = new Thread(r); thread.setName("custom-pool-" + thread.getId()); return thread; }); public static void main(String[] args) { System.out.println("主线程开始: " + Thread.currentThread().getName()); // 示例1：基本异步执行 basicAsyncExample(); // 示例2：链式调用 chainCallExample(); // 示例3：组合多个Future combineFutureExample(); // 示例4：异常处理 exceptionHandlingExample(); // 示例5：超时控制 timeoutControlExample(); // 关闭线程池 customExecutor.shutdown(); } private static void basicAsyncExample() { System.out.println("\n=== 示例1：基本异步执行 ==="); // 方式1：使用默认的ForkJoinPool CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> { sleep(1000); System.out.println("future1执行线程: " + Thread.currentThread().getName()); return "Hello"; }); // 方式2：使用自定义线程池 CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> { sleep(500); System.out.println("future2执行线程: " + Thread.currentThread().getName()); return "World"; }, customExecutor); // 阻塞获取结果 try { String result1 = future1.get(); String result2 = future2.get(); System.out.println("结果: " + result1 + " " + result2); } catch (Exception e) { e.printStackTrace(); } } private static void chainCallExample() { System.out.println("\n=== 示例2：链式调用 ==="); CompletableFuture.supplyAsync(() -> { System.out.println("第一步：获取用户ID - " + Thread.currentThread().getName()); sleep(300); return "user_123"; }, customExecutor).thenApply(userId -> { System.out.println("第二步：根据用户ID查询用户信息 - " + Thread.currentThread().getName()); sleep(300); return "用户: " + userId; }).thenAccept(userInfo -> { System.out.println("第三步：处理用户信息 - " + Thread.currentThread().getName()); System.out.println("处理结果: " + userInfo); }).thenRun(() -> { System.out.println("第四步：清理资源 - " + Thread.currentThread().getName()); }); sleep(2000); // 等待异步任务完成 } private static void combineFutureExample() { System.out.println("\n=== 示例3：组合多个Future ==="); // 模拟并行调用多个服务 CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> { sleep(800); System.out.println("调用用户服务完成"); return "用户信息"; }, customExecutor); CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> { sleep(600); System.out.println("调用商品服务完成"); return "商品信息"; }, customExecutor); CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> { sleep(400); System.out.println("调用库存服务完成"); return "库存信息"; }, customExecutor); // 方式1：全部完成 CompletableFuture<Void> allOf = CompletableFuture.allOf(future1, future2, future3); allOf.thenRun(() -> { System.out.println("所有服务调用完成"); try { System.out.println("用户服务结果: " + future1.get()); System.out.println("商品服务结果: " + future2.get()); System.out.println("库存服务结果: " + future3.get()); } catch (Exception e) { e.printStackTrace(); } }); // 方式2：任意一个完成 CompletableFuture<Object> anyOf = CompletableFuture.anyOf(future1, future2, future3); anyOf.thenAccept(result -> { System.out.println("最快返回的服务结果: " + result); }); sleep(2000); } private static void exceptionHandlingExample() { System.out.println("\n=== 示例4：异常处理 ==="); CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> { System.out.println("开始执行可能失败的任务"); if (Math.random() > 0.5) { throw new RuntimeException("随机失败!"); } return "成功结果"; }, customExecutor); // 异常处理方式1：exceptionally（类似catch） future.exceptionally(ex -> { System.out.println("捕获到异常: " + ex.getMessage()); return "默认值"; }).thenAccept(result -> { System.out.println("最终结果: " + result); }); // 异常处理方式2：handle（无论成功失败都会执行） future.handle((result, ex) -> { if (ex != null) { System.out.println("处理异常: " + ex.getMessage()); return "异常处理后的默认值"; } return result + "（处理后的）"; }).thenAccept(result -> { System.out.println("handle处理结果: " + result); }); sleep(1000); } private static void timeoutControlExample() { System.out.println("\n=== 示例5：超时控制 ==="); CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> { sleep(3000); // 模拟耗时操作 return "正常结果"; }, customExecutor); // 使用completeOnTimeout设置超时默认值 CompletableFuture<String> timeoutFuture = future .completeOnTimeout("超时默认值", 1000, TimeUnit.MILLISECONDS) .exceptionally(ex -> "异常默认值"); try { String result = timeoutFuture.get(2000, TimeUnit.MILLISECONDS); System.out.println("超时控制结果: " + result); } catch (TimeoutException e) { System.out.println("获取结果超时"); } catch (Exception e) { e.printStackTrace(); } } private static void sleep(long millis) { try { TimeUnit.MILLISECONDS.sleep(millis); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } }3.2 CompletableFuture在实际项目中的应用：并行调用优化import java.math.BigDecimal; import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; /** * 并行服务调用优化示例 * 场景：订单详情页需要聚合多个服务的数据 */ public class ParallelServiceCallDemo { // 模拟商品信息服务 static class ProductService { public Product getProductById(String productId) { sleep(200); // 模拟网络延迟 System.out.println(Thread.currentThread().getName() + " - 获取商品信息: " + productId); return new Product(productId, "商品" + productId, new BigDecimal("99.99")); } } // 模拟库存服务 static class InventoryService { public Integer getStock(String productId) { sleep(150); // 模拟网络延迟 System.out.println(Thread.currentThread().getName() + " - 获取库存: " + productId); return (int) (Math.random() * 100); // 模拟库存数量 } } // 模拟价格服务 static class PriceService { public BigDecimal getPrice(String productId) { sleep(100); // 模拟网络延迟 System.out.println(Thread.currentThread().getName() + " - 获取价格: " + productId); return new BigDecimal(Math.random() * 100 + 50).setScale(2, BigDecimal.ROUND_HALF_UP); } } // 模拟评价服务 static class ReviewService { public Review getReview(String productId) { sleep(250); // 模拟网络延迟 System.out.println(Thread.currentThread().getName() + " - 获取评价: " + productId); return new Review(productId, (int) (Math.random() * 5) + 1, "评价内容"); } } // 商品类 static class Product { String id; String name; BigDecimal price; public Product(String id, String name, BigDecimal price) { this.id = id; this.name = name; this.price = price; } } // 评价类 static class Review { String productId; int rating; String content; public Review(String productId, int rating, String content) { this.productId = productId; this.rating = rating; this.content = content; } } // 商品详情类 static class ProductDetail { String productId; Product product; Integer stock; BigDecimal price; Review review; @Override public String toString() { return String.format("商品详情{id=%s, 名称=%s, 价格=%.2f, 库存=%d, 评分=%d星}", productId, product.name, price, stock, review != null ? review.rating : 0); } } // 传统的串行调用方式 static class TraditionalService { private final ProductService productService = new ProductService(); private final InventoryService inventoryService = new InventoryService(); private final PriceService priceService = new PriceService(); private final ReviewService reviewService = new ReviewService(); public ProductDetail getProductDetail(String productId) { System.out.println("\n=== 传统串行调用 ==="); long startTime = System.currentTimeMillis(); // 串行调用各个服务 Product product = productService.getProductById(productId); Integer stock = inventoryService.getStock(productId); BigDecimal price = priceService.getPrice(productId); Review review = reviewService.getReview(productId); long endTime = System.currentTimeMillis(); System.out.println("串行调用总耗时: " + (endTime - startTime) + "ms"); ProductDetail detail = new ProductDetail(); detail.productId = productId; detail.product = product; detail.stock = stock; detail.price = price; detail.review = review; return detail; } } // 使用CompletableFuture的并行调用方式 static class ParallelService { private final ProductService productService = new ProductService(); private final InventoryService inventoryService = new InventoryService(); private final PriceService priceService = new PriceService(); private final ReviewService reviewService = new ReviewService(); private final ExecutorService executor = Executors.newFixedThreadPool(4); public ProductDetail getProductDetail(String productId) { System.out.println("\n=== 并行调用 ==="); long startTime = System.currentTimeMillis(); // 并行调用各个服务 CompletableFuture<Product> productFuture = CompletableFuture.supplyAsync(() -> productService.getProductById(productId), executor); CompletableFuture<Integer> stockFuture = CompletableFuture.supplyAsync(() -> inventoryService.getStock(productId), executor); CompletableFuture<BigDecimal> priceFuture = CompletableFuture.supplyAsync(() -> priceService.getPrice(productId), executor); CompletableFuture<Review> reviewFuture = CompletableFuture.supplyAsync(() -> reviewService.getReview(productId), executor); // 等待所有任务完成，并组装结果 CompletableFuture<ProductDetail> detailFuture = CompletableFuture.allOf(productFuture, stockFuture, priceFuture, reviewFuture) .thenApply(v -> { ProductDetail detail = new ProductDetail(); detail.productId = productId; try { detail.product = productFuture.get(); detail.stock = stockFuture.get(); detail.price = priceFuture.get(); detail.review = reviewFuture.get(); } catch (Exception e) { throw new RuntimeException(e); } return detail; }); ProductDetail detail = detailFuture.join(); long endTime = System.currentTimeMillis(); System.out.println("并行调用总耗时: " + (endTime - startTime) + "ms"); return detail; } public void shutdown() { executor.shutdown(); } } // 批量查询优化：同时查询多个商品 static class BatchQueryService { private final ProductService productService = new ProductService(); private final InventoryService inventoryService = new InventoryService(); private final ExecutorService executor = Executors.newFixedThreadPool(8); public Map<String, ProductDetail> batchQueryProductDetails(List<String> productIds) { System.out.println("\n=== 批量查询优化 ==="); long startTime = System.currentTimeMillis(); // 为每个商品创建并行查询任务 List<CompletableFuture<ProductDetail>> futures = productIds.stream() .map(productId -> querySingleProduct(productId)) .collect(Collectors.toList()); // 等待所有查询完成 CompletableFuture<Void> allFutures = CompletableFuture.allOf( futures.toArray(new CompletableFuture[0]) ); // 收集结果 Map<String, ProductDetail> result = new ConcurrentHashMap<>(); allFutures.thenRun(() -> { futures.forEach(future -> { try { ProductDetail detail = future.get(); result.put(detail.productId, detail); } catch (Exception e) { // 记录异常，但不影响其他结果 System.err.println("查询商品失败: " + e.getMessage()); } }); }).join(); // 阻塞直到所有任务完成 long endTime = System.currentTimeMillis(); System.out.println("批量查询" + productIds.size() + "个商品总耗时: " + (endTime - startTime) + "ms"); return result; } private CompletableFuture<ProductDetail> querySingleProduct(String productId) { return CompletableFuture.supplyAsync(() -> { Product product = productService.getProductById(productId); return product; }, executor).thenCombineAsync( CompletableFuture.supplyAsync(() -> inventoryService.getStock(productId), executor), (product, stock) -> { ProductDetail detail = new ProductDetail(); detail.productId = productId; detail.product = product; detail.stock = stock; return detail; } , executor).exceptionally(ex -> { // 异常处理：返回一个空的ProductDetail System.err.println("查询商品" + productId + "失败: " + ex.getMessage()); ProductDetail detail = new ProductDetail(); detail.productId = productId; return detail; }); } public void shutdown() { executor.shutdown(); } } // 超时和降级策略 static class ResilientService { private final PriceService priceService = new PriceService(); private final ExecutorService executor = Executors.newFixedThreadPool(2); public BigDecimal getPriceWithFallback(String productId) { System.out.println("\n=== 带超时和降级的服务调用 ==="); // 主服务调用，设置超时时间 CompletableFuture<BigDecimal> mainService = CompletableFuture.supplyAsync(() -> { return priceService.getPrice(productId); }, executor).orTimeout(300, TimeUnit.MILLISECONDS); // 300ms超时 // 降级服务（缓存或备用服务） CompletableFuture<BigDecimal> fallbackService = CompletableFuture.supplyAsync(() -> { sleep(100); // 降级服务更快 return new BigDecimal("88.88"); // 默认价格 }, executor); // 使用主服务，超时时使用降级服务 return mainService.exceptionally(ex -> { System.out.println("主服务调用失败，使用降级服务: " + ex.getMessage()); return fallbackService.join(); }).join(); } public void shutdown() { executor.shutdown(); } } public static void main(String[] args) { // 1. 对比串行和并行调用 TraditionalService traditionalService = new TraditionalService(); ParallelService parallelService = new ParallelService(); ProductDetail traditionalResult = traditionalService.getProductDetail("P001"); System.out.println("串行结果: " + traditionalResult); ProductDetail parallelResult = parallelService.getProductDetail("P001"); System.out.println("并行结果: " + parallelResult); // 2. 批量查询演示 List<String> productIds = new ArrayList<>(); for (int i = 1; i <= 10; i++) { productIds.add("P00" + i); } BatchQueryService batchService = new BatchQueryService(); Map<String, ProductDetail> batchResults = batchService.batchQueryProductDetails(productIds); System.out.println("\n批量查询结果:"); batchResults.forEach((id, detail) -> { System.out.println(" " + id + ": " + detail); }); // 3. 带超时和降级的服务调用 ResilientService resilientService = new ResilientService(); BigDecimal price = resilientService.getPriceWithFallback("P001"); System.out.println("\n最终价格: " + price); // 清理资源 parallelService.shutdown(); batchService.shutdown(); resilientService.shutdown(); } private static void sleep(long millis) { try { TimeUnit.MILLISECONDS.sleep(millis); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } }3.3 性能对比测试与最佳实践import java.util.ArrayList; import java.util.List; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; import java.util.stream.IntStream; /** * 并发性能对比测试 */ public class ConcurrencyPerformanceTest { // 模拟一个耗时服务 static class SlowService { private final int delayMs; public SlowService(int delayMs) { this.delayMs = delayMs; } public String process(int id) { try { // 模拟业务处理时间 Thread.sleep(delayMs); return "Result-" + id; } catch (InterruptedException e) { Thread.currentThread().interrupt(); return "Error-" + id; } } } // 测试配置 static class TestConfig { int taskCount; // 任务数量 int serviceDelayMs; // 每个任务耗时(ms) int threadPoolSize; // 线程池大小 public TestConfig(int taskCount, int serviceDelayMs, int threadPoolSize) { this.taskCount = taskCount; this.serviceDelayMs = serviceDelayMs; this.threadPoolSize = threadPoolSize; } } // 测试结果 static class TestResult { String methodName; long totalTime; int successCount; int failedCount; public TestResult(String methodName, long totalTime, int successCount, int failedCount) { this.methodName = methodName; this.totalTime = totalTime; this.successCount = successCount; this.failedCount = failedCount; } @Override public String toString() { return String.format("%-20s 总耗时: %6dms 成功: %4d 失败: %4d QPS: %.2f", methodName, totalTime, successCount, failedCount, successCount * 1000.0 / totalTime); } } /** * 方法1：传统串行处理 */ static TestResult testSerial(TestConfig config) { SlowService service = new SlowService(config.serviceDelayMs); List<String> results = new ArrayList<>(); long startTime = System.currentTimeMillis(); for (int i = 0; i < config.taskCount; i++) { try { results.add(service.process(i)); } catch (Exception e) { // 处理异常 } } long endTime = System.currentTimeMillis(); return new TestResult("串行处理", endTime - startTime, results.size(), 0); } /** * 方法2：使用ExecutorService */ static TestResult testExecutorService(TestConfig config) throws InterruptedException { SlowService service = new SlowService(config.serviceDelayMs); ExecutorService executor = Executors.newFixedThreadPool(config.threadPoolSize); AtomicInteger successCount = new AtomicInteger(0); AtomicInteger failedCount = new AtomicInteger(0); long startTime = System.currentTimeMillis(); List<Future<String>> futures = new ArrayList<>(); for (int i = 0; i < config.taskCount; i++) { final int taskId = i; Future<String> future = executor.submit(() -> service.process(taskId)); futures.add(future); } // 等待所有任务完成 for (Future<String> future : futures) { try { future.get(); successCount.incrementAndGet(); } catch (Exception e) { failedCount.incrementAndGet(); } } long endTime = System.currentTimeMillis(); executor.shutdown(); return new TestResult("ExecutorService", endTime - startTime, successCount.get(), failedCount.get()); } /** * 方法3：使用CompletableFuture（默认线程池） */ static TestResult testCompletableFutureDefault(TestConfig config) { SlowService service = new SlowService(config.serviceDelayMs); AtomicInteger successCount = new AtomicInteger(0); AtomicInteger failedCount = new AtomicInteger(0); long startTime = System.currentTimeMillis(); List<CompletableFuture<Void>> futures = IntStream.range(0, config.taskCount) .mapToObj(i -> CompletableFuture.supplyAsync(() -> service.process(i)) .thenAccept(result -> successCount.incrementAndGet()) .exceptionally(ex -> { failedCount.incrementAndGet(); return null; }) ).collect(Collectors.toList()); // 等待所有任务完成 CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join(); long endTime = System.currentTimeMillis(); return new TestResult("CompletableFuture(默认)", endTime - startTime, successCount.get(), failedCount.get()); } /** * 方法4：使用CompletableFuture（自定义线程池） */ static TestResult testCompletableFutureCustom(TestConfig config) { SlowService service = new SlowService(config.serviceDelayMs); ExecutorService executor = Executors.newFixedThreadPool(config.threadPoolSize); AtomicInteger successCount = new AtomicInteger(0); AtomicInteger failedCount = new AtomicInteger(0); long startTime = System.currentTimeMillis(); List<CompletableFuture<Void>> futures = IntStream.range(0, config.taskCount) .mapToObj(i -> CompletableFuture.supplyAsync(() -> service.process(i), executor) .thenAccept(result -> successCount.incrementAndGet()) .exceptionally(ex -> { failedCount.incrementAndGet(); return null; }) ).collect(Collectors.toList()); // 等待所有任务完成 CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join(); long endTime = System.currentTimeMillis(); executor.shutdown(); return new TestResult("CompletableFuture(自定义)", endTime - startTime, successCount.get(), failedCount.get()); } /** * 方法5：使用并行流（Parallel Stream） */ static TestResult testParallelStream(TestConfig config) { SlowService service = new SlowService(config.serviceDelayMs); AtomicInteger successCount = new AtomicInteger(0); AtomicInteger failedCount = new AtomicInteger(0); long startTime = System.currentTimeMillis(); IntStream.range(0, config.taskCount) .parallel() .forEach(i -> { try { service.process(i); successCount.incrementAndGet(); } catch (Exception e) { failedCount.incrementAndGet(); } }); long endTime = System.currentTimeMillis(); return new TestResult("并行流(ParallelStream)", endTime - startTime, successCount.get(), failedCount.get()); } /** * 运行性能测试 */ static void runPerformanceTest(TestConfig config) throws InterruptedException { System.out.println("\n" + "=".repeat(80)); System.out.println("性能测试配置: 任务数=" + config.taskCount + ", 任务耗时=" + config.serviceDelayMs + "ms" + ", 线程数=" + config.threadPoolSize); System.out.println("-".repeat(80)); List<TestResult> results = new ArrayList<>(); // 运行各种测试 results.add(testSerial(config)); results.add(testExecutorService(config)); results.add(testCompletableFutureDefault(config)); results.add(testCompletableFutureCustom(config)); results.add(testParallelStream(config)); // 打印结果 results.forEach(System.out::println); // 找出最佳方案 TestResult best = results.stream() .min((r1, r2) -> Long.compare(r1.totalTime, r2.totalTime)) .orElseThrow(); System.out.println("-".repeat(80)); System.out.println("最佳方案: " + best.methodName + " (耗时最少)"); System.out.println("=".repeat(80)); } /** * CompletableFuture最佳实践总结 */ static class CompletableFutureBestPractices { // 最佳实践1：合理配置线程池 public static ExecutorService createOptimalThreadPool() { int coreCount = Runtime.getRuntime().availableProcessors(); return new ThreadPoolExecutor( coreCount * 2, // 核心线程数：CPU核心数 * 2 coreCount * 4, // 最大线程数：CPU核心数 * 4 60L, TimeUnit.SECONDS, // 空闲线程存活时间 new LinkedBlockingQueue<>(1000), // 任务队列容量 new ThreadFactoryBuilder() .setNameFormat("business-pool-%d") .setUncaughtExceptionHandler((t, e) -> System.err.println("线程" + t.getName() + "异常: " + e.getMessage())) .build(), new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略：由调用者线程执行 ); } // 最佳实践2：链式调用优化 public static CompletableFuture<String> optimizedChainCall() { return CompletableFuture.supplyAsync(() -> { System.out.println("第一步：获取原始数据"); return "原始数据"; }).thenApplyAsync(data -> { System.out.println("第二步：处理数据"); return data + " -> 处理后"; }).thenApplyAsync(data -> { System.out.println("第三步：转换格式"); return data + " -> 转换后"; }).exceptionally(ex -> { System.err.println("处理失败: " + ex.getMessage()); return "默认值"; }); } // 最佳实践3：超时控制 public static CompletableFuture<String> withTimeoutControl() { ExecutorService executor = createOptimalThreadPool(); return CompletableFuture.supplyAsync(() -> { try { Thread.sleep(2000); // 模拟耗时操作 return "正常结果"; } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new RuntimeException(e); } }, executor) .completeOnTimeout("超时默认值", 1000, TimeUnit.MILLISECONDS) .whenComplete((result, ex) -> { if (ex != null) { System.err.println("任务异常完成: " + ex.getMessage()); } else { System.out.println("任务完成，结果: " + result); } }); } // 最佳实践4：资源清理 public static void withResourceCleanup() { ExecutorService executor = createOptimalThreadPool(); CompletableFuture.runAsync(() -> { System.out.println("执行任务..."); // 模拟资源使用 try { // 业务逻辑 Thread.sleep(1000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } finally { // 确保资源清理 System.out.println("清理资源..."); } }, executor); // 优雅关闭 executor.shutdown(); try { if (!executor.awaitTermination(5, TimeUnit.SECONDS)) { executor.shutdownNow(); } } catch (InterruptedException e) { executor.shutdownNow(); Thread.currentThread().interrupt(); } } } public static void main(String[] args) throws InterruptedException { System.out.println("Java并发性能对比测试"); System.out.println("测试环境: CPU核心数=" + Runtime.getRuntime().availableProcessors()); // 测试场景1：IO密集型任务（短耗时，多任务） System.out.println("\n场景1：IO密集型（100个任务，每个耗时50ms）"); TestConfig ioIntensiveConfig = new TestConfig(100, 50, 20); runPerformanceTest(ioIntensiveConfig); // 测试场景2：计算密集型任务（长耗时，少任务） System.out.println("\n场景2：计算密集型（20个任务，每个耗时500ms）"); TestConfig cpuIntensiveConfig = new TestConfig(20, 500, Runtime.getRuntime().availableProcessors()); runPerformanceTest(cpuIntensiveConfig); // 测试场景3：混合型任务 System.out.println("\n场景3：混合型（50个任务，每个耗时200ms）"); TestConfig mixedConfig = new TestConfig(50, 200, Runtime.getRuntime().availableProcessors() * 2); runPerformanceTest(mixedConfig); // 演示最佳实践 System.out.println("\n" + "=".repeat(80)); System.out.println("CompletableFuture最佳实践示例"); System.out.println("-".repeat(80)); // 示例1：链式调用优化 System.out.println("1. 链式调用优化示例:"); CompletableFutureBestPractices.optimizedChainCall() .thenAccept(result -> System.out.println("最终结果: " + result)) .join(); // 示例2：超时控制 System.out.println("\n2. 超时控制示例:"); CompletableFutureBestPractices.withTimeoutControl() .thenAccept(result -> System.out.println("超时控制结果: " + result)) .join(); // 示例3：资源清理 System.out.println("\n3. 资源清理示例:"); CompletableFutureBestPractices.withResourceCleanup(); System.out.println("=".repeat(80)); } }四、总结与最佳实践关键技术点总结：ThreadLocal的核心要点：每个线程有独立的变量副本，线程安全必须在使用后调用remove()避免内存泄漏在线程池中使用时要格外小心InheritableThreadLocal可以实现父子线程数据传递CompletableFuture的核心要点：提供了强大的异步编程能力支持链式调用、组合、异常处理等合理使用线程池，避免耗尽资源注意异常处理和超时控制性能优化建议：IO密集型任务：使用较大的线程池（核心数 * 2 \~ *4）计算密集型任务：使用较小的线程池（≈ 核心数）合理设置任务队列大小，避免内存溢出监控线程池状态，及时调整参数生产环境注意事项：为线程池设置合理的命名，便于监控和排查实现优雅的关闭逻辑，避免任务丢失添加完善的监控和告警机制定期进行性能压测，优化线程池参数通过深入理解ThreadLocal和CompletableFuture的原理和最佳实践，我们可以构建出高性能、可维护的并发Java应用。记住：并发编程没有银弹，只有合适的场景选择合适的工具和技术。