## 概述多线程编程是Java开发中的重要技能，它允许程序同时执行多个任务，充分利用现代多核处理器的能力。Java提供了丰富的多线程API和工具，从基础的Thread类到高级的并发工具包。 ### 为什么需要多线程？在单线程程序中，所有任务都必须按顺序执行。当遇到耗时操作（如文件IO、网络请求、数据库查询）时，整个程序会被阻塞，用户体验很差。多线程技术能够： 1. **提高程序响应性**: 主线程可以继续响应用户操作，而后台线程处理耗时任务 2. **充分利用硬件资源**: 现代计算机都是多核处理器，多线程可以并行执行 3. **提高系统吞吐量**: 服务器可以同时处理多个客户端请求 4. **改善资源利用率**: 当某个线程等待IO时，其他线程可以继续使用CPU ### 多线程的核心概念 #### 并发与并行的区别 - **并发(Concurrency)**: 在同一时间段内处理多个任务，但不一定同时执行。就像一个人同时处理多个任务，通过快速切换来给人并行的感觉 - **并行(Parallelism)**: 真正同时执行多个任务，需要多个CPU核心。就像多个人同时工作 #### 线程安全问题 - **竞态条件(Race Condition)**: 多个线程同时访问共享资源时，由于执行顺序不确定而导致的问题 - **数据不一致**: 多个线程同时修改共享数据，可能导致数据状态不正确 - **可见性问题**: 一个线程对共享变量的修改，其他线程可能看不到 #### 常见并发问题 - **死锁(Deadlock)**: 两个或多个线程互相等待对方释放资源，形成循环等待 - **活锁(Livelock)**: 线程不断重试但始终无法成功执行 - **饥饿(Starvation)**: 某些线程长期无法获得所需资源 ### Java多线程的优势 - **提高程序性能**: 充分利用多核CPU，实现真正的并行计算 - **改善用户体验**: 避免UI阻塞，保持界面响应 - **提高系统吞吐量**: 并发处理多个请求，提升服务器性能 - **更好的资源利用**: 在IO等待时执行其他任务，提高CPU利用率 ### 适用场景多线程特别适合以下场景： - **Web服务器**: 同时处理多个HTTP请求 - **图形用户界面**: 后台处理数据，前台保持响应 - **数据处理**: 并行处理大量数据，如图像处理、数据分析 - **生产者-消费者模式**: 如消息队列、任务调度系统 ## 线程基础 ### 什么是线程？线程是程序执行的最小单位，是进程中的一个执行路径。在Java中，每个线程都有自己的栈空间，但共享进程的堆内存、方法区等资源。 #### 线程与进程的区别 | 特性 | 进程 | 线程 | |------|------|------| | 资源占用 | 独立的内存空间 | 共享进程的内存空间 | | 通信成本 | 高（需要IPC机制） | 低（直接访问共享内存） | | 创建开销 | 大 | 小 | | 切换开销 | 大 | 小 | | 稳定性 | 一个进程崩溃不影响其他进程 | 一个线程崩溃可能影响整个进程 | ### 创建和启动线程 Java提供了两种主要方式来创建线程： #### 方式1：继承Thread类这是最直接的方式，但由于Java单继承的限制，不够灵活。 #### 方式2：实现Runnable接口这是推荐的方式，因为： - 避免了单继承的限制 - 更好的代码复用性 - 符合面向接口编程的原则 #### 方式3：使用Lambda表达式（Java 8+）简化了代码编写，特别适合简单的任务。 ```java // 方式1: 继承Thread类 public class MyThread extends Thread { private String threadName; public MyThread(String name) { this.threadName = name; } @Override public void run() { for (int i = 0; i < 5; i++) { System.out.println(threadName + " - 计数: " + i); try { Thread.sleep(1000); // 暂停1秒 } catch (InterruptedException e) { System.out.println(threadName + " 被中断"); return; } } System.out.println(threadName + " 执行完成"); } } // 方式2: 实现Runnable接口 public class MyRunnable implements Runnable { private String taskName; public MyRunnable(String name) { this.taskName = name; } @Override public void run() { for (int i = 0; i < 5; i++) { System.out.println(taskName + " - 计数: " + i); try { Thread.sleep(1000); } catch (InterruptedException e) { System.out.println(taskName + " 被中断"); Thread.currentThread().interrupt(); // 重新设置中断状态 return; } } System.out.println(taskName + " 执行完成"); } } // 使用示例 public class ThreadBasics { public static void main(String[] args) { // 使用Thread子类 MyThread thread1 = new MyThread("线程-1"); thread1.start(); // 注意：调用start()而不是run() // 使用Runnable接口 Thread thread2 = new Thread(new MyRunnable("任务-2")); thread2.start(); // 使用Lambda表达式 Thread thread3 = new Thread(() -> { for (int i = 0; i < 3; i++) { System.out.println("Lambda线程 - 计数: " + i); try { Thread.sleep(1500); } catch (InterruptedException e) { Thread.currentThread().interrupt(); return; } } }); thread3.start(); System.out.println("主线程继续执行"); } } ``` ### 线程状态和生命周期 #### 线程状态详解 Java线程在其生命周期中会经历多个状态，理解这些状态对于调试和优化多线程程序非常重要： 1. **NEW（新建）**: 线程对象已创建但尚未启动 2. **RUNNABLE（可运行）**: 线程正在JVM中运行，但可能正在等待操作系统的其他资源（如CPU时间片） 3. **BLOCKED（阻塞）**: 线程被阻塞，等待获取监视器锁 4. **WAITING（等待）**: 线程无限期等待另一个线程执行某个操作 5. **TIMED_WAITING（计时等待）**: 线程等待另一个线程执行某个操作，但有时间限制 6. **TERMINATED（终止）**: 线程已经执行完毕 #### 状态转换图 ``` NEW → RUNNABLE ⇄ BLOCKED ↓ ⇅ WAITING/TIMED_WAITING ↓ TERMINATED ``` #### 常见的状态转换场景 - **NEW → RUNNABLE**: 调用`start()`方法 - **RUNNABLE → BLOCKED**: 尝试获取synchronized锁失败 - **RUNNABLE → WAITING**: 调用`wait()`、`join()`等方法 - **RUNNABLE → TIMED_WAITING**: 调用`sleep()`、`wait(timeout)`等方法 - **WAITING/TIMED_WAITING → RUNNABLE**: 被`notify()`唤醒或超时 - **RUNNABLE → TERMINATED**: 线程执行完毕或发生异常 ```java public class ThreadStates { public static void demonstrateThreadStates() { Thread thread = new Thread(() -> { System.out.println("线程开始执行"); // RUNNABLE状态 for (int i = 0; i < 3; i++) { System.out.println("工作中: " + i); try { Thread.sleep(1000); // TIMED_WAITING状态 } catch (InterruptedException e) { Thread.currentThread().interrupt(); return; } } System.out.println("线程执行完成"); }); // NEW状态 System.out.println("线程状态 (创建后): " + thread.getState()); thread.start(); // RUNNABLE状态 System.out.println("线程状态 (启动后): " + thread.getState()); try { Thread.sleep(500); // 可能是RUNNABLE或TIMED_WAITING System.out.println("线程状态 (运行中): " + thread.getState()); thread.join(); // 等待线程完成 // TERMINATED状态 System.out.println("线程状态 (完成后): " + thread.getState()); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } public static void main(String[] args) { demonstrateThreadStates(); } } /* 线程状态说明: - NEW: 创建但未启动 - RUNNABLE: 可运行状态(可能正在运行或等待CPU时间片) - BLOCKED: 阻塞状态(等待监视器锁) - WAITING: 等待状态(无限期等待) - TIMED_WAITING: 计时等待状态(有限期等待) - TERMINATED: 终止状态(已完成执行) */ ``` ### 线程中断机制 #### 什么是线程中断？线程中断是Java提供的一种协作式的线程停止机制。它不是强制停止线程，而是给线程发送一个中断信号，由线程自己决定如何响应这个信号。 #### 为什么需要中断机制？ 1. **安全停止**: 直接调用`stop()`方法（已废弃）会导致数据不一致 2. **协作式**: 线程可以在合适的时机检查中断状态并优雅地退出 3. **资源清理**: 线程可以在退出前进行必要的清理工作 #### 中断机制的三个重要方法 1. **`interrupt()`**: 设置线程的中断状态为true 2. **`isInterrupted()`**: 检查线程是否被中断（不清除中断状态） 3. **`interrupted()`**: 检查当前线程是否被中断（清除中断状态） #### 中断处理的最佳实践 1. **定期检查中断状态**: 在循环中检查`Thread.currentThread().isInterrupted()` 2. **正确处理InterruptedException**: 捕获异常后重新设置中断状态 3. **及时响应中断**: 收到中断信号后应尽快退出 ```java public class ThreadInterruption { // 正确处理中断的示例 public static class InterruptibleTask implements Runnable { @Override public void run() { System.out.println("任务开始执行"); for (int i = 0; i < 10; i++) { // 检查中断状态 if (Thread.currentThread().isInterrupted()) { System.out.println("检测到中断信号，任务提前结束"); return; } System.out.println("执行步骤: " + i); try { Thread.sleep(1000); // 可中断的操作 } catch (InterruptedException e) { System.out.println("在sleep中被中断"); // 重新设置中断状态 Thread.currentThread().interrupt(); return; } } System.out.println("任务正常完成"); } } public static void main(String[] args) { Thread workerThread = new Thread(new InterruptibleTask()); workerThread.start(); try { Thread.sleep(3500); // 让任务运行一段时间 System.out.println("主线程发送中断信号"); workerThread.interrupt(); // 发送中断信号 workerThread.join(); // 等待线程结束 System.out.println("工作线程已结束"); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } } ``` ## 同步机制 ### 为什么需要同步？在多线程环境中，当多个线程同时访问和修改共享资源时，可能会出现数据不一致的问题。同步机制确保在任何时刻只有一个线程能够访问共享资源，从而保证数据的一致性和完整性。 ### synchronized关键字 #### synchronized的原理 `synchronized`是Java中最基本的同步机制，它基于对象的内置锁（monitor）实现。每个Java对象都有一个内置锁，当线程进入synchronized代码块时，会自动获取这个锁，退出时自动释放。 #### synchronized的四种使用方式 1. **同步实例方法**: 锁定当前对象实例 2. **同步静态方法**: 锁定当前类的Class对象 3. **同步代码块（this）**: 锁定当前对象实例 4. **同步代码块（自定义对象）**: 锁定指定对象 #### 选择合适的锁对象 - **粒度要合适**: 锁的范围不要太大，避免不必要的阻塞 - **避免锁泄露**: 不要将锁对象暴露给外部 - **一致性**: 对同一资源的访问应该使用同一个锁 ```java public class SynchronizedExample { private int counter = 0; private final Object lock = new Object(); // 同步实例方法 public synchronized void incrementSync() { counter++; System.out.println("同步方法 - 计数器: " + counter + ", 线程: " + Thread.currentThread().getName()); } // 同步静态方法 public static synchronized void staticSyncMethod() { System.out.println("静态同步方法被调用: " + Thread.currentThread().getName()); } // 同步代码块 public void incrementBlock() { synchronized (this) { // 使用this作为锁 counter++; System.out.println("同步代码块(this) - 计数器: " + counter + ", 线程: " + Thread.currentThread().getName()); } } // 使用自定义锁对象 public void incrementCustomLock() { synchronized (lock) { // 使用自定义锁对象 counter++; System.out.println("同步代码块(自定义锁) - 计数器: " + counter + ", 线程: " + Thread.currentThread().getName()); } } // 非同步方法对比 public void incrementUnsafe() { counter++; System.out.println("非同步方法 - 计数器: " + counter + ", 线程: " + Thread.currentThread().getName()); } public int getCounter() { return counter; } public static void main(String[] args) throws InterruptedException { SynchronizedExample example = new SynchronizedExample(); // 创建多个线程测试同步 Thread[] threads = new Thread[5]; for (int i = 0; i < threads.length; i++) { final int threadNum = i; threads[i] = new Thread(() -> { for (int j = 0; j < 3; j++) { if (threadNum % 2 == 0) { example.incrementSync(); // 同步方法 } else { example.incrementBlock(); // 同步代码块 } try { Thread.sleep(100); } catch (InterruptedException e) { Thread.currentThread().interrupt(); return; } } }, "Thread-" + i); } // 启动所有线程 for (Thread thread : threads) { thread.start(); } // 等待所有线程完成 for (Thread thread : threads) { thread.join(); } System.out.println("最终计数器值: " + example.getCounter()); } } ``` ### volatile关键字 #### volatile的作用 `volatile`关键字解决了多线程环境中的可见性问题，但不能解决原子性问题。它有两个主要作用： 1. **保证可见性**: 一个线程对volatile变量的修改，对其他线程立即可见 2. **禁止指令重排**: 防止编译器和处理器对代码进行优化重排 #### volatile的使用场景 - **状态标志**: 如停止标志、初始化标志等 - **双重检查锁定**: 在单例模式中使用 - **读多写少**: 适用于一个线程写，多个线程读的场景 #### volatile vs synchronized | 特性 | volatile | synchronized | |------|----------|-------------| | 原子性 | 不保证 | 保证 | | 可见性 | 保证 | 保证 | | 有序性 | 部分保证 | 保证 | | 阻塞性 | 非阻塞 | 可能阻塞 | | 使用场景 | 简单状态标志 | 复合操作 | ```java public class VolatileExample { // volatile确保可见性，但不保证原子性 private volatile boolean flag = false; private volatile int counter = 0; // 使用volatile的标志位 public void startWork() { new Thread(() -> { System.out.println("工作线程开始，等待信号..."); while (!flag) { // 等待flag变为true // 忙等待 - 在实际应用中应该避免 } System.out.println("工作线程收到信号，开始工作"); // 模拟工作 for (int i = 0; i < 5; i++) { System.out.println("工作进度: " + (i + 1) + "/5"); try { Thread.sleep(500); } catch (InterruptedException e) { Thread.currentThread().interrupt(); return; } } System.out.println("工作完成"); }, "Worker").start(); } public void sendSignal() { try { Thread.sleep(2000); // 模拟准备时间 System.out.println("主线程发送开始信号"); flag = true; // volatile写操作，立即对其他线程可见 } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } // volatile不能保证原子性的例子 public void incrementCounter() { counter++; // 这不是原子操作，即使counter是volatile } public int getCounter() { return counter; } public static void main(String[] args) throws InterruptedException { VolatileExample example = new VolatileExample(); // 演示volatile的可见性 example.startWork(); example.sendSignal(); Thread.sleep(4000); // 等待工作完成 // 演示volatile无法保证原子性 System.out.println("\n测试volatile的原子性问题:"); Thread[] incrementThreads = new Thread[10]; for (int i = 0; i < incrementThreads.length; i++) { incrementThreads[i] = new Thread(() -> { for (int j = 0; j < 1000; j++) { example.incrementCounter(); } }); } for (Thread thread : incrementThreads) { thread.start(); } for (Thread thread : incrementThreads) { thread.join(); } System.out.println("期望计数器值: 10000"); System.out.println("实际计数器值: " + example.getCounter()); System.out.println("差异说明volatile无法保证原子性"); } } ``` ### Lock接口和ReentrantLock #### Lock接口的优势 `Lock`接口是Java 5引入的更灵活的锁机制，相比`synchronized`有以下优势： 1. **可中断**: 可以响应中断请求 2. **可超时**: 可以设置获取锁的超时时间 3. **非阻塞**: 提供`tryLock()`方法尝试获取锁 4. **条件变量**: 支持多个条件变量（Condition） 5. **公平锁**: 可以选择公平锁或非公平锁 #### ReentrantLock特性 `ReentrantLock`是`Lock`接口的主要实现类，具有以下特性： - **可重入**: 同一线程可以多次获取同一个锁 - **公平性选择**: 构造时可以选择是否为公平锁 - **锁信息**: 提供了丰富的锁状态查询方法 #### Lock vs synchronized | 特性 | Lock | synchronized | |------|------|-------------| | 灵活性 | 高 | 低 | | 性能 | 高 | 一般 | | 可中断 | 支持 | 不支持 | | 超时 | 支持 | 不支持 | | 条件变量 | 多个 | 一个 | | 使用复杂度 | 高 | 低 | #### 读写锁（ReadWriteLock）读写锁允许多个读线程同时访问，但写线程是独占的： - **读锁**: 多个线程可以同时获取读锁 - **写锁**: 只有一个线程可以获取写锁 - **适用场景**: 读多写少的场景，如缓存系统 ```java import java.util.concurrent.locks.*; import java.util.concurrent.TimeUnit; public class LockExample { private final ReentrantLock lock = new ReentrantLock(); private final ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock(); private final Lock readLock = readWriteLock.readLock(); private final Lock writeLock = readWriteLock.writeLock(); private int counter = 0; private String data = "初始数据"; // 使用ReentrantLock的基本示例 public void incrementWithLock() { lock.lock(); // 获取锁 try { counter++; System.out.println("使用Lock - 计数器: " + counter + ", 线程: " + Thread.currentThread().getName()); Thread.sleep(100); // 模拟工作 } catch (InterruptedException e) { Thread.currentThread().interrupt(); } finally { lock.unlock(); // 确保释放锁 } } // 尝试获取锁的示例 public boolean tryIncrementWithTimeout() { try { // 尝试在2秒内获取锁 if (lock.tryLock(2, TimeUnit.SECONDS)) { try { counter++; System.out.println("tryLock成功 - 计数器: " + counter + ", 线程: " + Thread.currentThread().getName()); Thread.sleep(3000); // 模拟长时间工作 return true; } finally { lock.unlock(); } } else { System.out.println("tryLock失败 - 超时, 线程: " + Thread.currentThread().getName()); return false; } } catch (InterruptedException e) { Thread.currentThread().interrupt(); return false; } } // 读写锁示例 - 读操作 public String readData() { readLock.lock(); try { System.out.println("读取数据: " + data + ", 线程: " + Thread.currentThread().getName()); Thread.sleep(1000); // 模拟读取操作 return data; } catch (InterruptedException e) { Thread.currentThread().interrupt(); return null; } finally { readLock.unlock(); } } // 读写锁示例 - 写操作 public void writeData(String newData) { writeLock.lock(); try { System.out.println("写入数据: " + newData + ", 线程: " + Thread.currentThread().getName()); this.data = newData; Thread.sleep(2000); // 模拟写入操作 } catch (InterruptedException e) { Thread.currentThread().interrupt(); } finally { writeLock.unlock(); } } // 可重入锁示例 public void recursiveLockExample(int depth) { lock.lock(); try { System.out.println("递归深度: " + depth + ", 持有锁次数: " + lock.getHoldCount()); if (depth > 0) { recursiveLockExample(depth - 1); // 递归调用，重入锁 } } finally { lock.unlock(); } } public static void main(String[] args) throws InterruptedException { LockExample example = new LockExample(); // 测试基本锁功能 System.out.println("=== 基本Lock功能测试 ==="); Thread[] basicThreads = new Thread[3]; for (int i = 0; i < basicThreads.length; i++) { basicThreads[i] = new Thread(() -> { for (int j = 0; j < 2; j++) { example.incrementWithLock(); } }, "Basic-" + i); } for (Thread thread : basicThreads) { thread.start(); } for (Thread thread : basicThreads) { thread.join(); } // 测试tryLock System.out.println("\n=== tryLock测试 ==="); Thread longTask = new Thread(() -> example.tryIncrementWithTimeout(), "Long-Task"); Thread quickTask = new Thread(() -> example.tryIncrementWithTimeout(), "Quick-Task"); longTask.start(); Thread.sleep(500); // 确保longTask先获取锁 quickTask.start(); longTask.join(); quickTask.join(); // 测试读写锁 System.out.println("\n=== 读写锁测试 ==="); // 多个读线程 for (int i = 0; i < 3; i++) { new Thread(() -> example.readData(), "Reader-" + i).start(); } Thread.sleep(500); // 一个写线程 new Thread(() -> example.writeData("新数据-" + System.currentTimeMillis()), "Writer").start(); Thread.sleep(500); // 再来一个读线程 new Thread(() -> example.readData(), "Reader-Final").start(); Thread.sleep(5000); // 测试可重入锁 System.out.println("\n=== 可重入锁测试 ==="); new Thread(() -> example.recursiveLockExample(3), "Recursive").start(); Thread.sleep(2000); } } ``` ## 线程间通信 ### 为什么需要线程间通信？在多线程程序中，线程之间经常需要协调工作，比如： - 生产者-消费者模式：生产者生产数据，消费者消费数据 - 任务依赖：一个任务需要等待另一个任务完成 - 资源共享：多个线程需要协调访问共享资源 ### wait/notify机制 #### wait/notify的工作原理 `wait/notify`是Java中最基本的线程间通信机制，基于对象的内置锁实现： 1. **wait()**: 使当前线程等待，释放锁，进入等待状态 2. **notify()**: 唤醒一个等待的线程 3. **notifyAll()**: 唤醒所有等待的线程 #### 使用wait/notify的注意事项 1. **必须在synchronized块中使用**: wait/notify必须在获取对象锁的情况下调用 2. **使用while循环**: 被唤醒后需要重新检查条件，避免虚假唤醒 3. **选择合适的锁对象**: 通常使用共享资源对象作为锁 #### 经典应用场景：生产者-消费者模式生产者-消费者模式是多线程编程中的经典模式： - **生产者**: 生产数据并放入缓冲区 - **消费者**: 从缓冲区取出数据进行处理 - **缓冲区**: 有限容量的数据存储区域 ```java public class WaitNotifyExample { private final Object lock = new Object(); private boolean dataReady = false; private String sharedData; // 生产者 public void producer() { new Thread(() -> { for (int i = 1; i <= 5; i++) { synchronized (lock) { sharedData = "数据-" + i; dataReady = true; System.out.println("生产者生产: " + sharedData); lock.notify(); // 通知等待的消费者 } try { Thread.sleep(1000); // 模拟生产时间 } catch (InterruptedException e) { Thread.currentThread().interrupt(); return; } } }, "Producer").start(); } // 消费者 public void consumer() { new Thread(() -> { for (int i = 1; i <= 5; i++) { synchronized (lock) { while (!dataReady) { // 使用while而不是if try { System.out.println("消费者等待数据..."); lock.wait(); // 等待生产者通知 } catch (InterruptedException e) { Thread.currentThread().interrupt(); return; } } System.out.println("消费者消费: " + sharedData); dataReady = false; // 重置状态 } try { Thread.sleep(500); // 模拟消费时间 } catch (InterruptedException e) { Thread.currentThread().interrupt(); return; } } }, "Consumer").start(); } public static void main(String[] args) { WaitNotifyExample example = new WaitNotifyExample(); example.consumer(); // 先启动消费者 example.producer(); // 再启动生产者 } } ``` ### Condition接口 #### Condition的优势 `Condition`接口是Lock的配套工具，提供了比wait/notify更强大的线程间通信能力： 1. **多个条件**: 一个Lock可以创建多个Condition 2. **更精确的控制**: 可以精确控制哪些线程被唤醒 3. **更好的性能**: 避免了不必要的线程唤醒 4. **支持中断**: 可以响应中断请求 #### Condition vs wait/notify | 特性 | Condition | wait/notify | |------|-----------|-------------| | 锁机制 | Lock | synchronized | | 条件数量 | 多个 | 一个 | | 精确唤醒 | 支持 | 不支持 | | 中断响应 | 支持 | 支持 | | 使用复杂度 | 高 | 低 | #### 典型应用场景 Condition特别适用于需要多个等待条件的场景，如： - **有界缓冲区**: 需要"不满"和"不空"两个条件 - **读写锁**: 需要"无读者"和"无写者"两个条件 - **任务调度**: 需要多个不同类型的等待条件 ```java import java.util.concurrent.locks.*; import java.util.Queue; import java.util.LinkedList; public class ConditionExample { private final Lock lock = new ReentrantLock(); private final Condition notFull = lock.newCondition(); private final Condition notEmpty = lock.newCondition(); private final Queue queue = new LinkedList<>(); private final int maxSize = 3; // 生产者方法 public void produce(String item) { lock.lock(); try { // 队列满时等待 while (queue.size() == maxSize) { System.out.println("队列已满，生产者等待..."); notFull.await(); // 等待队列不满 } queue.offer(item); System.out.println("生产: " + item + ", 队列大小: " + queue.size()); notEmpty.signal(); // 通知消费者队列不空 } catch (InterruptedException e) { Thread.currentThread().interrupt(); } finally { lock.unlock(); } } // 消费者方法 public String consume() { lock.lock(); try { // 队列空时等待 while (queue.isEmpty()) { System.out.println("队列为空，消费者等待..."); notEmpty.await(); // 等待队列不空 } String item = queue.poll(); System.out.println("消费: " + item + ", 队列大小: " + queue.size()); notFull.signal(); // 通知生产者队列不满 return item; } catch (InterruptedException e) { Thread.currentThread().interrupt(); return null; } finally { lock.unlock(); } } public static void main(String[] args) { ConditionExample example = new ConditionExample(); // 启动生产者线程 Thread producer = new Thread(() -> { for (int i = 1; i <= 8; i++) { example.produce("商品-" + i); try { Thread.sleep(500); } catch (InterruptedException e) { Thread.currentThread().interrupt(); return; } } }, "Producer"); // 启动消费者线程 Thread consumer = new Thread(() -> { for (int i = 1; i <= 8; i++) { example.consume(); try { Thread.sleep(1000); // 消费比生产慢 } catch (InterruptedException e) { Thread.currentThread().interrupt(); return; } } }, "Consumer"); producer.start(); consumer.start(); try { producer.join(); consumer.join(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } System.out.println("所有任务完成"); } } ## 并发工具类 ### 为什么需要并发工具类？ Java并发包（java.util.concurrent）提供了一系列高级并发工具类，这些工具类解决了常见的并发场景问题，相比手动使用锁和条件变量，具有以下优势： - **更简单**: 封装了复杂的同步逻辑 - **更可靠**: 经过充分测试，不容易出错 - **更高效**: 针对特定场景进行了优化 - **更易理解**: 语义清晰，代码可读性好 ### CountDownLatch - 倒计时门闩 #### CountDownLatch的工作原理 `CountDownLatch`是一个同步辅助类，允许一个或多个线程等待其他线程完成操作。它的工作原理类似于倒计时： 1. **初始化**: 设置一个计数器的初始值 2. **等待**: 调用`await()`方法的线程会阻塞，直到计数器归零 3. **倒计时**: 其他线程调用`countDown()`方法，计数器减1 4. **唤醒**: 当计数器归零时，所有等待的线程被唤醒 #### 典型应用场景 - **等待多个任务完成**: 主线程等待所有子任务完成后再继续 - **同时开始**: 多个线程同时等待一个开始信号 - **分阶段执行**: 确保某个阶段的所有任务都完成后再进入下一阶段 #### CountDownLatch的特点 - **一次性**: 计数器不能重置，只能使用一次 - **灵活性**: 可以设置任意的初始计数值 - **高效性**: 基于AQS实现，性能优秀 ```java import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; public class CountDownLatchExample { public static void simulateGameStart() { final int playerCount = 5; CountDownLatch startLatch = new CountDownLatch(1); // 等待游戏开始 CountDownLatch finishLatch = new CountDownLatch(playerCount); // 等待所有玩家完成 // 创建玩家线程 for (int i = 1; i <= playerCount; i++) { final int playerId = i; new Thread(() -> { try { System.out.println("玩家" + playerId + " 准备就绪"); startLatch.await(); // 等待游戏开始信号 // 模拟游戏过程 int gameTime = (int)(Math.random() * 3000) + 1000; Thread.sleep(gameTime); System.out.println("玩家" + playerId + " 完成游戏，用时: " + gameTime + "ms"); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } finally { finishLatch.countDown(); // 完成游戏 } }, "Player-" + i).start(); } try { Thread.sleep(2000); // 模拟准备时间 System.out.println("所有玩家准备完成，游戏开始！"); startLatch.countDown(); // 发送开始信号 // 等待所有玩家完成 if (finishLatch.await(10, TimeUnit.SECONDS)) { System.out.println("所有玩家都完成了游戏！"); } else { System.out.println("游戏超时！"); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } public static void main(String[] args) { simulateGameStart(); } } ``` ### CyclicBarrier - 循环屏障 #### CyclicBarrier的工作原理 `CyclicBarrier`是一个同步辅助类，允许一组线程相互等待，直到所有线程都到达某个公共屏障点。它的名字来源于： - **Cyclic（循环）**: 屏障可以重复使用 - **Barrier（屏障）**: 线程在此点等待其他线程 #### CyclicBarrier vs CountDownLatch | 特性 | CyclicBarrier | CountDownLatch | |------|---------------|----------------| | 可重用性 | 可重用 | 一次性 | | 等待方式 | 相互等待 | 单向等待 | | 触发条件 | 所有线程到达 | 计数器归零 | | 回调支持 | 支持 | 不支持 | | 使用场景 | 多阶段任务 | 一次性同步 | #### 典型应用场景 - **多阶段任务**: 需要分阶段执行的并行任务 - **游戏开发**: 等待所有玩家准备就绪 - **数据处理**: 并行处理数据，每个阶段都需要同步 - **测试场景**: 并发测试中确保所有线程同时开始 #### CyclicBarrier的特点 - **可重用**: 屏障可以被重复使用 - **回调支持**: 可以指定所有线程到达屏障时执行的回调 - **异常处理**: 如果某个线程被中断，其他等待的线程会收到BrokenBarrierException ```java import java.util.concurrent.CyclicBarrier; import java.util.concurrent.BrokenBarrierException; public class CyclicBarrierExample { public static void simulateRace() { final int runnerCount = 4; // 创建屏障，当4个选手都到达时执行裁判动作 CyclicBarrier barrier = new CyclicBarrier(runnerCount, () -> { System.out.println("🏁 裁判：所有选手准备完成，比赛开始！"); }); // 创建选手线程 for (int i = 1; i <= runnerCount; i++) { final int runnerId = i; new Thread(() -> { try { // 模拟选手准备时间 int prepTime = (int)(Math.random() * 2000) + 500; Thread.sleep(prepTime); System.out.println("选手" + runnerId + " 准备完成，等待其他选手..."); barrier.await(); // 等待所有选手到达屏障 // 比赛开始后的行为 System.out.println("选手" + runnerId + " 开始跑步！"); // 模拟跑步时间 int runTime = (int)(Math.random() * 3000) + 1000; Thread.sleep(runTime); System.out.println("选手" + runnerId + " 到达终点，用时: " + runTime + "ms"); } catch (InterruptedException | BrokenBarrierException e) { System.out.println("选手" + runnerId + " 比赛中断"); Thread.currentThread().interrupt(); } }, "Runner-" + i).start(); } } // 演示CyclicBarrier的可重用性 public static void demonstrateReusability() { final int participants = 3; CyclicBarrier barrier = new CyclicBarrier(participants, () -> { System.out.println("--- 一轮会议结束 ---"); }); for (int round = 1; round <= 2; round++) { final int currentRound = round; System.out.println("\n第" + currentRound + "轮会议开始:"); for (int i = 1; i <= participants; i++) { final int participantId = i; new Thread(() -> { try { Thread.sleep((int)(Math.random() * 1000) + 500); System.out.println("第" + currentRound + "轮 - 参与者" + participantId + " 到达会议室"); barrier.await(); // 等待所有人到达 System.out.println("第" + currentRound + "轮 - 参与者" + participantId + " 开始会议"); } catch (InterruptedException | BrokenBarrierException e) { Thread.currentThread().interrupt(); } }, "Participant-" + i + "-Round-" + currentRound).start(); } try { Thread.sleep(3000); // 等待这轮完成 } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } } } public static void main(String[] args) { System.out.println("=== 赛跑示例 ==="); simulateRace(); try { Thread.sleep(6000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } System.out.println("\n=== 可重用性示例 ==="); demonstrateReusability(); } } ``` ### Semaphore - 信号量 #### Semaphore的工作原理 `Semaphore`（信号量）是一个计数信号量，用于控制同时访问特定资源的线程数量。它的工作原理基于许可证（permit）的概念： 1. **初始化**: 设置许可证的数量 2. **获取许可**: 线程调用`acquire()`获取许可证，如果没有可用许可证则阻塞 3. **释放许可**: 线程调用`release()`释放许可证，增加可用许可证数量 4. **限流控制**: 通过许可证数量控制并发访问 #### Semaphore的类型 - **二进制信号量**: 许可证数量为1，类似于互斥锁 - **计数信号量**: 许可证数量大于1，用于限制并发数量 - **公平信号量**: 按照线程请求的顺序分配许可证 #### 典型应用场景 - **连接池**: 限制数据库连接数 - **限流**: 控制API调用频率 - **资源管理**: 限制对有限资源的并发访问 - **停车场**: 控制停车位的使用 #### Semaphore vs 其他同步工具 | 特性 | Semaphore | synchronized | Lock | |------|-----------|-------------|------| | 许可证数量 | 可配置 | 1 | 1 | | 公平性 | 可选择 | 不保证 | 可选择 | | 中断响应 | 支持 | 不支持 | 支持 | | 超时支持 | 支持 | 不支持 | 支持 | | 使用场景 | 限流控制 | 互斥访问 | 复杂同步 | ```java import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; public class SemaphoreExample { // 模拟停车场 public static class ParkingLot { private final Semaphore availableSpaces; private final int totalSpaces; public ParkingLot(int spaces) { this.totalSpaces = spaces; this.availableSpaces = new Semaphore(spaces, true); // 公平模式 } public boolean park(String carId) { try { System.out.println(carId + " 尝试停车..."); // 尝试获取停车位，最多等待3秒 if (availableSpaces.tryAcquire(3, TimeUnit.SECONDS)) { System.out.println(carId + " 成功停车！剩余车位: " + availableSpaces.availablePermits()); return true; } else { System.out.println(carId + " 停车失败，停车场已满且等待超时"); return false; } } catch (InterruptedException e) { Thread.currentThread().interrupt(); return false; } } public void leave(String carId) { System.out.println(carId + " 离开停车场"); availableSpaces.release(); // 释放停车位 System.out.println("车位释放后剩余: " + availableSpaces.availablePermits()); } public int getAvailableSpaces() { return availableSpaces.availablePermits(); } } // 模拟数据库连接池 public static class DatabaseConnectionPool { private final Semaphore connections; private final String[] connectionPool; private final boolean[] used; public DatabaseConnectionPool(int size) { this.connections = new Semaphore(size, true); this.connectionPool = new String[size]; this.used = new boolean[size]; // 初始化连接池 for (int i = 0; i < size; i++) { connectionPool[i] = "Connection-" + i; } } public String getConnection() throws InterruptedException { connections.acquire(); // 获取连接许可 synchronized (this) { for (int i = 0; i < connectionPool.length; i++) { if (!used[i]) { used[i] = true; System.out.println("获取连接: " + connectionPool[i] + " (线程: " + Thread.currentThread().getName() + ")"); return connectionPool[i]; } } } throw new RuntimeException("无可用连接"); // 理论上不会到达这里 } public void releaseConnection(String connectionName) { synchronized (this) { for (int i = 0; i < connectionPool.length; i++) { if (connectionPool[i].equals(connectionName)) { used[i] = false; System.out.println("释放连接: " + connectionName + " (线程: " + Thread.currentThread().getName() + ")"); connections.release(); // 释放连接许可 return; } } } throw new RuntimeException("无效连接: " + connectionName); } } public static void testParkingLot() { System.out.println("=== 停车场示例 ==="); ParkingLot parkingLot = new ParkingLot(3); // 3个停车位 // 模拟5辆车尝试停车 for (int i = 1; i <= 5; i++) { final String carId = "Car-" + i; new Thread(() -> { if (parkingLot.park(carId)) { try { // 停车一段时间 Thread.sleep((int)(Math.random() * 3000) + 1000); parkingLot.leave(carId); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } }).start(); try { Thread.sleep(500); // 错开到达时间 } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } } public static void testConnectionPool() { System.out.println("\n=== 数据库连接池示例 ==="); DatabaseConnectionPool pool = new DatabaseConnectionPool(2); // 2个连接 // 模拟5个任务需要数据库连接 for (int i = 1; i <= 5; i++) { final int taskId = i; new Thread(() -> { try { String connection = pool.getConnection(); // 模拟数据库操作 System.out.println("任务" + taskId + " 使用 " + connection + " 执行数据库操作"); Thread.sleep((int)(Math.random() * 2000) + 1000); pool.releaseConnection(connection); System.out.println("任务" + taskId + " 完成"); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }, "Task-" + i).start(); } } public static void main(String[] args) { testParkingLot(); try { Thread.sleep(6000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } testConnectionPool(); } } ``` ## 线程池 ### 为什么需要线程池？直接创建线程存在以下问题： 1. **资源消耗大**: 创建和销毁线程需要消耗系统资源 2. **响应时间慢**: 每次创建线程都需要时间 3. **缺乏管理**: 无法有效控制线程数量和状态 4. **系统稳定性**: 无限制创建线程可能导致系统崩溃线程池的优势： - **资源复用**: 重复利用已创建的线程 - **控制并发**: 有效控制最大并发线程数 - **管理线程**: 提供定时执行、周期执行等功能 - **提高响应**: 任务到达时不需要等待线程创建 ### ThreadPoolExecutor详解 #### 线程池的核心参数 `ThreadPoolExecutor`是线程池的核心实现类，理解其构造参数对于正确使用线程池至关重要： 1. **corePoolSize（核心线程数）**: 线程池维护的最小线程数 2. **maximumPoolSize（最大线程数）**: 线程池允许的最大线程数 3. **keepAliveTime（存活时间）**: 非核心线程的空闲存活时间 4. **unit（时间单位）**: 存活时间的时间单位 5. **workQueue（工作队列）**: 存储等待执行任务的队列 6. **threadFactory（线程工厂）**: 创建新线程的工厂 7. **handler（拒绝策略）**: 当任务无法执行时的处理策略 #### 线程池的工作流程 1. **提交任务**: 向线程池提交任务 2. **核心线程**: 如果运行线程数 < corePoolSize，创建新线程执行任务 3. **队列缓存**: 如果运行线程数 >= corePoolSize，将任务加入队列 4. **扩展线程**: 如果队列满了且运行线程数 < maximumPoolSize，创建新线程 5. **拒绝任务**: 如果队列满了且运行线程数 >= maximumPoolSize，执行拒绝策略 #### 常见的工作队列类型 - **ArrayBlockingQueue**: 有界队列，基于数组实现 - **LinkedBlockingQueue**: 可选有界队列，基于链表实现 - **SynchronousQueue**: 同步队列，不存储元素 - **PriorityBlockingQueue**: 优先级队列，支持任务优先级 #### 拒绝策略 - **AbortPolicy**: 抛出RejectedExecutionException异常（默认） - **CallerRunsPolicy**: 在调用者线程中执行任务 - **DiscardPolicy**: 静默丢弃任务 - **DiscardOldestPolicy**: 丢弃队列中最老的任务 ```java import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; public class ThreadPoolExample { // 自定义线程工厂 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); return t; } } // 自定义拒绝策略 static class CustomRejectedExecutionHandler implements RejectedExecutionHandler { @Override public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { System.out.println("任务被拒绝: " + r.toString() + ", 线程池状态 - 活跃线程: " + executor.getActiveCount() + ", 队列大小: " + executor.getQueue().size()); // 可以选择不同的处理策略 // 1. 抛出异常 (默认策略) // throw new RejectedExecutionException("任务被拒绝"); // 2. 在调用者线程中执行 if (!executor.isShutdown()) { r.run(); } } } public static void demonstrateBasicThreadPool() { System.out.println("=== 基本线程池示例 ==="); // 创建线程池 ThreadPoolExecutor executor = new ThreadPoolExecutor( 2, // 核心线程数 4, // 最大线程数 60L, // 非核心线程空闲存活时间 TimeUnit.SECONDS, // 时间单位 new LinkedBlockingQueue<>(3), // 任务队列，容量为3 new CustomThreadFactory("MyPool"), // 线程工厂 new CustomRejectedExecutionHandler() // 拒绝策略 ); // 提交任务 for (int i = 1; i <= 10; i++) { final int taskId = i; try { executor.submit(() -> { System.out.println("任务" + taskId + " 开始执行，线程: " + Thread.currentThread().getName()); try { Thread.sleep(2000); // 模拟任务执行 } catch (InterruptedException e) { Thread.currentThread().interrupt(); return; } System.out.println("任务" + taskId + " 执行完成"); }); // 打印线程池状态 System.out.println("提交任务" + taskId + " 后 - 核心线程数: " + executor.getCorePoolSize() + ", 当前线程数: " + executor.getPoolSize() + ", 活跃线程数: " + executor.getActiveCount() + ", 队列大小: " + executor.getQueue().size()); Thread.sleep(300); // 错开提交时间 } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } } // 关闭线程池 executor.shutdown(); try { if (!executor.awaitTermination(30, TimeUnit.SECONDS)) { System.out.println("强制关闭线程池"); executor.shutdownNow(); } } catch (InterruptedException e) { executor.shutdownNow(); Thread.currentThread().interrupt(); } } // 演示不同类型的线程池 public static void demonstrateExecutorTypes() { System.out.println("\n=== 不同类型线程池示例 ==="); // 1. 固定大小线程池 ExecutorService fixedPool = Executors.newFixedThreadPool(3); System.out.println("固定大小线程池 (3个线程):"); for (int i = 1; i <= 5; i++) { final int taskId = i; fixedPool.submit(() -> { System.out.println("FixedPool - 任务" + taskId + ", 线程: " + Thread.currentThread().getName()); try { Thread.sleep(1000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }); } // 2. 缓存线程池 ExecutorService cachedPool = Executors.newCachedThreadPool(); System.out.println("\n缓存线程池:"); for (int i = 1; i <= 5; i++) { final int taskId = i; cachedPool.submit(() -> { System.out.println("CachedPool - 任务" + taskId + ", 线程: " + Thread.currentThread().getName()); try { Thread.sleep(1000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }); } // 3. 单线程池 ExecutorService singlePool = Executors.newSingleThreadExecutor(); System.out.println("\n单线程池:"); for (int i = 1; i <= 3; i++) { final int taskId = i; singlePool.submit(() -> { System.out.println("SinglePool - 任务" + taskId + ", 线程: " + Thread.currentThread().getName()); try { Thread.sleep(1000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }); } // 关闭所有线程池 fixedPool.shutdown(); cachedPool.shutdown(); singlePool.shutdown(); try { fixedPool.awaitTermination(5, TimeUnit.SECONDS); cachedPool.awaitTermination(5, TimeUnit.SECONDS); singlePool.awaitTermination(5, TimeUnit.SECONDS); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } public static void main(String[] args) { demonstrateBasicThreadPool(); try { Thread.sleep(8000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } demonstrateExecutorTypes(); } } ``` ### ScheduledThreadPoolExecutor - 定时任务 #### 定时任务的重要性在实际应用中，我们经常需要执行定时任务： - **系统维护**: 定时清理日志、缓存清理 - **数据同步**: 定时从外部系统同步数据 - **监控检查**: 定时检查系统健康状态 - **业务处理**: 定时发送邮件、生成报表 #### ScheduledThreadPoolExecutor的特点 `ScheduledThreadPoolExecutor`是专门用于执行定时任务的线程池： - **延迟执行**: 可以延迟一定时间后执行任务 - **周期执行**: 支持固定频率或固定延迟的周期性执行 - **异常处理**: 任务异常不会影响后续任务的执行 - **线程复用**: 基于线程池，避免频繁创建销毁线程 #### 两种周期执行方式 1. **scheduleAtFixedRate**: 固定频率执行 - 不管任务执行时间，严格按照固定间隔执行 - 如果任务执行时间超过间隔，会立即执行下一次 2. **scheduleWithFixedDelay**: 固定延迟执行 - 任务执行完成后，等待固定时间再执行下一次 - 确保任务之间有固定的间隔时间 #### 实际应用场景 - **缓存清理**: 定时清理过期的缓存数据 - **健康检查**: 定时检查服务状态并发送告警 - **数据备份**: 定时备份重要数据 - **日志轮转**: 定时切换日志文件 ```java import java.util.concurrent.*; import java.time.LocalDateTime; import java.time.format.DateTimeFormatter; public class ScheduledThreadPoolExample { private static final DateTimeFormatter formatter = DateTimeFormatter.ofPattern("HH:mm:ss"); public static void demonstrateScheduledTasks() { System.out.println("=== 定时任务示例 ==="); ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(3); System.out.println("开始时间: " + LocalDateTime.now().format(formatter)); // 1. 延迟执行一次 scheduler.schedule(() -> { System.out.println("延迟任务执行 - " + LocalDateTime.now().format(formatter)); }, 2, TimeUnit.SECONDS); // 2. 固定频率执行 (每3秒执行一次，无论上次任务是否完成) ScheduledFuture fixedRateTask = scheduler.scheduleAtFixedRate(() -> { System.out.println("固定频率任务开始 - " + LocalDateTime.now().format(formatter)); try { Thread.sleep(1000); // 模拟任务执行时间 } catch (InterruptedException e) { Thread.currentThread().interrupt(); return; } System.out.println("固定频率任务结束 - " + LocalDateTime.now().format(formatter)); }, 1, 3, TimeUnit.SECONDS); // 3. 固定延迟执行 (上次任务完成后等待2秒再执行下次任务) ScheduledFuture fixedDelayTask = scheduler.scheduleWithFixedDelay(() -> { System.out.println("固定延迟任务开始 - " + LocalDateTime.now().format(formatter)); try { Thread.sleep(1500); // 模拟任务执行时间 } catch (InterruptedException e) { Thread.currentThread().interrupt(); return; } System.out.println("固定延迟任务结束 - " + LocalDateTime.now().format(formatter)); }, 1, 2, TimeUnit.SECONDS); // 让任务运行10秒 try { Thread.sleep(10000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } // 取消特定任务 System.out.println("取消定时任务..."); fixedRateTask.cancel(false); fixedDelayTask.cancel(false); // 关闭调度器 scheduler.shutdown(); try { if (!scheduler.awaitTermination(5, TimeUnit.SECONDS)) { scheduler.shutdownNow(); } } catch (InterruptedException e) { scheduler.shutdownNow(); Thread.currentThread().interrupt(); } } // 模拟实际应用场景 public static void simulateRealWorldScenarios() { System.out.println("\n=== 实际应用场景 ==="); ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2); // 场景1: 定时清理缓存 scheduler.scheduleAtFixedRate(() -> { System.out.println("执行缓存清理任务 - " + LocalDateTime.now().format(formatter)); // 模拟清理逻辑 try { Thread.sleep(500); } catch (InterruptedException e) { Thread.currentThread().interrupt(); return; } System.out.println("缓存清理完成"); }, 0, 5, TimeUnit.SECONDS); // 场景2: 健康检查 scheduler.scheduleWithFixedDelay(() -> { System.out.println("执行健康检查 - " + LocalDateTime.now().format(formatter)); // 模拟健康检查 boolean healthy = Math.random() > 0.3; // 70%概率健康 if (healthy) { System.out.println("系统健康 ✓"); } else { System.out.println("系统异常 ✗ - 发送告警"); } }, 2, 3, TimeUnit.SECONDS); // 场景3: 延迟执行清理任务 scheduler.schedule(() -> { System.out.println("执行延迟清理任务 - " + LocalDateTime.now().format(formatter)); System.out.println("清理临时文件..."); }, 8, TimeUnit.SECONDS); // 运行15秒后关闭 try { Thread.sleep(15000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } scheduler.shutdown(); System.out.println("调度器已关闭"); } public static void main(String[] args) { demonstrateScheduledTasks(); simulateRealWorldScenarios(); } } ## Future 和 CompletableFuture ### 异步编程的重要性在现代应用中，异步编程变得越来越重要： - **提高响应性**: 避免阻塞主线程 - **提高吞吐量**: 充分利用系统资源 - **用户体验**: 避免界面卡顿 - **系统性能**: 减少资源浪费 ### Future 基础使用 #### Future接口的作用 `Future`接口代表异步计算的结果，它提供了以下功能： 1. **获取结果**: `get()`方法获取计算结果（阻塞） 2. **检查状态**: `isDone()`检查任务是否完成 3. **取消任务**: `cancel()`取消任务执行 4. **超时控制**: `get(timeout)`设置获取结果的超时时间 #### Future的局限性虽然Future提供了基本的异步编程能力，但它有一些局限： - **阻塞获取**: `get()`方法会阻塞当前线程 - **无法组合**: 难以组合多个Future - **异常处理**: 异常处理不够优雅 - **回调支持**: 不支持完成回调 #### 适用场景 Future适用于以下场景： - **简单异步任务**: 提交任务后等待结果 - **批量处理**: 并行处理多个独立任务 - **超时控制**: 需要设置任务执行超时时间 ```java import java.util.concurrent.*; import java.util.List; import java.util.ArrayList; public class FutureExample { public static void demonstrateBasicFuture() { System.out.println("=== Future基础示例 ==="); ExecutorService executor = Executors.newFixedThreadPool(3); // 提交Callable任务，返回Future Future future1 = executor.submit(() -> { Thread.sleep(2000); return "任务1完成: " + Thread.currentThread().getName(); }); Future future2 = executor.submit(() -> { Thread.sleep(1500); return 42; }); Future future3 = executor.submit(() -> { Thread.sleep(1000); return Math.random() > 0.5; }); try { System.out.println("主线程继续执行其他工作..."); Thread.sleep(500); // 检查任务状态 System.out.println("任务1是否完成: " + future1.isDone()); System.out.println("任务2是否完成: " + future2.isDone()); System.out.println("任务3是否完成: " + future3.isDone()); // 获取结果（阻塞直到完成） System.out.println("获取结果:"); System.out.println("任务1结果: " + future1.get()); System.out.println("任务2结果: " + future2.get()); System.out.println("任务3结果: " + future3.get()); } catch (InterruptedException | ExecutionException e) { e.printStackTrace(); } executor.shutdown(); } public static void demonstrateFutureTimeout() { System.out.println("\n=== Future超时示例 ==="); ExecutorService executor = Executors.newSingleThreadExecutor(); Future slowTask = executor.submit(() -> { Thread.sleep(5000); // 模拟耗时任务 return "慢任务完成"; }); try { // 设置超时时间为2秒 String result = slowTask.get(2, TimeUnit.SECONDS); System.out.println("任务结果: " + result); } catch (TimeoutException e) { System.out.println("任务超时，取消任务"); slowTask.cancel(true); // 中断任务 } catch (InterruptedException | ExecutionException e) { e.printStackTrace(); } executor.shutdown(); } // 批量处理任务 public static void demonstrateBatchProcessing() { System.out.println("\n=== 批量任务处理 ==="); ExecutorService executor = Executors.newFixedThreadPool(3); List> tasks = new ArrayList<>(); for (int i = 1; i <= 5; i++) { final int taskId = i; tasks.add(() -> { Thread.sleep(1000 + (int)(Math.random() * 1000)); int result = taskId * taskId; System.out.println("任务" + taskId + " 完成，结果: " + result); return result; }); } try { // invokeAll: 等待所有任务完成 List> futures = executor.invokeAll(tasks); System.out.println("所有任务提交完成，等待结果..."); int sum = 0; for (Future future : futures) { sum += future.get(); } System.out.println("所有任务完成，结果总和: " + sum); // invokeAny: 返回第一个完成的任务结果 System.out.println("\n测试invokeAny:"); Integer firstResult = executor.invokeAny(tasks); System.out.println("第一个完成的任务结果: " + firstResult); } catch (InterruptedException | ExecutionException e) { e.printStackTrace(); } executor.shutdown(); } public static void main(String[] args) { demonstrateBasicFuture(); demonstrateFutureTimeout(); demonstrateBatchProcessing(); } } ``` ### CompletableFuture 异步编程 #### CompletableFuture的优势 `CompletableFuture`是Java 8引入的增强版Future，解决了传统Future的局限性： 1. **非阻塞**: 支持回调和链式操作，避免阻塞 2. **组合能力**: 可以轻松组合多个异步操作 3. **异常处理**: 提供了优雅的异常处理机制 4. **函数式编程**: 支持Lambda表达式和函数式编程风格 #### CompletableFuture的核心方法 - **创建方法**: `supplyAsync()`, `runAsync()` - **转换方法**: `thenApply()`, `thenApplyAsync()` - **消费方法**: `thenAccept()`, `thenRun()` - **组合方法**: `thenCompose()`, `thenCombine()` - **异常处理**: `exceptionally()`, `handle()` - **等待方法**: `allOf()`, `anyOf()` #### 异步编程模式 1. **流水线模式**: 使用`thenApply()`链接多个转换操作 2. **并行组合**: 使用`thenCombine()`并行执行多个任务 3. **异常恢复**: 使用`exceptionally()`处理异常情况 4. **条件执行**: 使用`thenCompose()`根据结果决定下一步操作 #### 实际应用场景 - **微服务调用**: 并行调用多个微服务API - **数据处理管道**: 构建复杂的数据处理流水线 - **缓存预热**: 异步预加载缓存数据 - **文件处理**: 异步读取和处理文件 ```java import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.List; import java.util.stream.Collectors; import java.util.Arrays; public class CompletableFutureExample { // 模拟异步服务 public static CompletableFuture fetchUserInfo(String userId) { return CompletableFuture.supplyAsync(() -> { try { Thread.sleep(1000); // 模拟网络延迟 } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new RuntimeException(e); } return "用户信息-" + userId; }); } public static CompletableFuture fetchUserPreferences(String userId) { return CompletableFuture.supplyAsync(() -> { try { Thread.sleep(800); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new RuntimeException(e); } return "偏好设置-" + userId; }); } public static CompletableFuture fetchUserOrders(String userId) { return CompletableFuture.supplyAsync(() -> { try { Thread.sleep(1200); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new RuntimeException(e); } return "订单历史-" + userId; }); } public static void demonstrateBasicCompletableFuture() { System.out.println("=== CompletableFuture基础示例 ==="); // 1. 简单的异步任务 CompletableFuture future1 = CompletableFuture.supplyAsync(() -> { try { Thread.sleep(1000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } return "Hello"; }); CompletableFuture future2 = future1.thenApply(s -> s + " World"); CompletableFuture future3 = future2.thenAccept(System.out::println); try { future3.get(); // 等待完成 } catch (InterruptedException | ExecutionException e) { e.printStackTrace(); } // 2. 链式操作 CompletableFuture.supplyAsync(() -> "数据处理") .thenApply(s -> s + " -> 转换") .thenApply(String::toUpperCase) .thenAccept(result -> System.out.println("最终结果: " + result)) .join(); // 等待完成 } public static void demonstrateComposition() { System.out.println("\n=== CompletableFuture组合示例 ==="); String userId = "user123"; // 串行组合 - thenCompose CompletableFuture userProfileFuture = fetchUserInfo(userId) .thenCompose(userInfo -> fetchUserPreferences(userId) .thenApply(preferences -> userInfo + " + " + preferences) ); // 并行组合 - thenCombine CompletableFuture combinedFuture = fetchUserInfo(userId) .thenCombine(fetchUserPreferences(userId), (userInfo, preferences) -> userInfo + " + " + preferences); // 等待任意一个完成 - applyToEither CompletableFuture fastestFuture = fetchUserInfo(userId) .applyToEither(fetchUserPreferences(userId), result -> "最快结果: " + result); try { System.out.println("串行组合结果: " + userProfileFuture.get()); System.out.println("并行组合结果: " + combinedFuture.get()); System.out.println("最快响应结果: " + fastestFuture.get()); } catch (InterruptedException | ExecutionException e) { e.printStackTrace(); } } public static void demonstrateErrorHandling() { System.out.println("\n=== CompletableFuture错误处理 ==="); // 模拟可能失败的任务 CompletableFuture riskyTask = CompletableFuture.supplyAsync(() -> { if (Math.random() > 0.5) { throw new RuntimeException("随机错误!"); } return "成功结果"; }); // 错误处理 CompletableFuture handledFuture = riskyTask .exceptionally(throwable -> { System.out.println("捕获异常: " + throwable.getMessage()); return "默认结果"; }) .thenApply(result -> "处理后: " + result); // 使用handle处理成功和失败情况 CompletableFuture handledFuture2 = CompletableFuture.supplyAsync(() -> { if (Math.random() > 0.7) { throw new RuntimeException("另一个错误!"); } return "成功"; }).handle((result, throwable) -> { if (throwable != null) { return "错误处理: " + throwable.getMessage(); } return "正常结果: " + result; }); try { System.out.println(handledFuture.get()); System.out.println(handledFuture2.get()); } catch (InterruptedException | ExecutionException e) { e.printStackTrace(); } } public static void demonstrateAllOfAnyOf() { System.out.println("\n=== CompletableFuture.allOf/anyOf示例 ==="); String userId = "user456"; // 并行获取用户所有信息 CompletableFuture userInfo = fetchUserInfo(userId); CompletableFuture userPrefs = fetchUserPreferences(userId); CompletableFuture userOrders = fetchUserOrders(userId); // 等待所有任务完成 CompletableFuture allTasks = CompletableFuture.allOf(userInfo, userPrefs, userOrders); CompletableFuture allResultsFuture = allTasks.thenApply(v -> { try { String info = userInfo.get(); String prefs = userPrefs.get(); String orders = userOrders.get(); return String.format("完整用户档案: [%s, %s, %s]", info, prefs, orders); } catch (InterruptedException | ExecutionException e) { throw new RuntimeException(e); } }); // 等待任意任务完成 CompletableFuture

Java 多线程开发指南