第9章：并发（Concurrency）
一、导读
1.1 本章学习目标
第9章聚焦于 Java 并发编程的最佳实践。并发是 Java 中最复杂也最容易出错的主题之一。本章通过8条经验法则，帮助你掌握并发编程的核心知识，写出正确的并发代码。
本章的核心学习目标包括：
- 理解同步的重要性：掌握 synchronized 的语义和使用。
- 理解 volatile 的作用：掌握可见性和有序性。
- 掌握并发工具类的使用：学会使用
java.util.concurrent包中的工具。 - 理解线程池的设计：掌握 Executor 框架。
- 掌握 CompletableFuture 的使用：学会异步编程。
1.2 为什么第九章如此重要
并发编程是现代软件的必修课。多核处理器、分布式系统、高并发应用——这些场景都需要程序员具备并发编程的能力。错误的并发代码可能导致数据损坏、死锁、性能问题，而且这些问题往往难以复现和调试。
1.3 核心问题引导
为什么需要同步？不同步会导致什么问题？
volatile 能保证什么？不能保证什么？
如何正确使用 java.util.concurrent 中的并发工具？
线程池的核心参数有哪些？如何合理配置？
CompletableFuture 如何简化异步编程？
二、核心概念详解
9.1 同步共享可变数据（Item 78）
9.1.1 synchronized 的作用
synchronized 保证两件事：
原子性：同步块内的操作不会被其他线程打断。
可见性：同步块内的修改对其他线程立即可见。
java
// 反模式：不同步
public class UnsafeCounter {
private long count = 0;
public void increment() {
count++; // 非原子操作：读-改-写
}
public long getCount() {
return count;
}
}
// 正确做法：同步
public class SafeCounter {
private long count = 0;
public synchronized void increment() {
count++;
}
public synchronized long getCount() {
return count;
}
}9.1.2 读写都需要同步
java
// 反模式：只同步写
public class BrokenCounter {
private long count = 0;
public synchronized void increment() {
count++;
}
public long getCount() { // 没有同步！
return count; // 可能读到过期值
}
}
// 正确做法：读写都同步
public class SafeCounter {
private long count = 0;
public synchronized void increment() {
count++;
}
public synchronized long getCount() {
return count;
}
}9.2 避免过度同步（Item 79）
java
// 反模式：过度同步
public class ObservableSet<E> extends HashSet<E> {
private final List<SetObserver<E>> observers = new ArrayList<>();
public void addObserver(SetObserver<E> observer) {
synchronized (observers) {
observers.add(observer);
}
}
public synchronized boolean add(E element) {
boolean added = super.add(element);
if (added) {
// 在同步块内调用外部方法——可能导致死锁
synchronized (observers) {
for (SetObserver<E> observer : observers) {
observer.added(this, element);
}
}
}
return added;
}
}
// 正确做法：开放调用（open call）
public synchronized boolean add(E element) {
boolean added = super.add(element);
if (added) {
notifyElementAdded(element);
}
return added;
}
private void notifyElementAdded(E element) {
List<SetObserver<E>> observersSnapshot;
synchronized (observers) {
observersSnapshot = new ArrayList<>(observers);
}
for (SetObserver<E> observer : observersSnapshot) {
observer.added(this, element);
}
}9.3 优先使用并发工具而非 wait 和 notify（Item 80）
java
// 反模式：使用 wait 和 notify
public class LegacyQueue<T> {
private final List<T> list = new ArrayList<>();
private final int capacity;
public synchronized T take() throws InterruptedException {
while (list.isEmpty()) {
wait();
}
T item = list.remove(0);
notifyAll();
return item;
}
public synchronized void put(T item) throws InterruptedException {
while (list.size() == capacity) {
wait();
}
list.add(item);
notifyAll();
}
}
// 正确做法：使用 BlockingQueue
public class ModernQueue<T> {
private final BlockingQueue<T> queue = new ArrayBlockingQueue<>(100);
public T take() throws InterruptedException {
return queue.take();
}
public void put(T item) throws InterruptedException {
queue.put(item);
}
}9.4 线程安全优于线程安全文档（Item 82）
java
// 反模式：依赖文档说明线程安全
/**
* This class is thread-safe.
* (但实际上不是)
*/
public class NotThreadSafe {
private int count = 0;
public int getCount() { return count; }
public void increment() { count++; }
}
// 正确做法：确保线程安全
public class ThreadSafe {
private final AtomicInteger count = new AtomicInteger(0);
public int getCount() { return count.get(); }
public void increment() { count.incrementAndGet(); }
}9.5 延迟初始化要慎重（Item 83）
java
// 正常初始化（推荐）
private final FieldType field = computeFieldValue();
// 惰性初始化（如果需要）
private volatile FieldType field;
private FieldType getField() {
FieldType result = field;
if (result == null) {
synchronized (this) {
result = field;
if (result == null) {
field = result = computeFieldValue();
}
}
}
return result;
}
// 静态内部类方式（静态字段的惰性初始化）
private static class FieldHolder {
static final FieldType field = computeFieldValue();
}
private static FieldType getField() {
return FieldHolder.field;
}9.6 不要依赖线程调度器（Item 84）
java
// 反模式：依赖 Thread.sleep 和线程优先级
public void doWork() {
Thread thread1 = new Thread(() -> { /* ... */ });
Thread thread2 = new Thread(() -> { /* ... */ });
thread1.setPriority(Thread.MAX_PRIORITY);
thread2.setPriority(Thread.MIN_PRIORITY);
thread1.start();
thread2.start();
// 依赖调度器来决定执行顺序——不可靠！
}
// 正确做法：使用并发工具协调
public void doWork() {
ExecutorService executor = Executors.newFixedThreadPool(2);
CountDownLatch latch = new CountDownLatch(2);
executor.submit(() -> {
try {
// 工作
} finally {
latch.countDown();
}
});
executor.submit(() -> {
try {
// 工作
} finally {
latch.countDown();
}
});
latch.await(); // 等待两个任务完成
}三、实战案例
9.1 线程池实战
java
public class ThreadPoolDemo {
public static void main(String[] args) {
// 自定义线程池
ThreadPoolExecutor executor = new ThreadPoolExecutor(
4, // 核心线程数
8, // 最大线程数
60L, TimeUnit.SECONDS, // 空闲线程存活时间
new LinkedBlockingQueue<>(100), // 工作队列
new ThreadFactoryBuilder()
.setNameFormat("worker-%d")
.setDaemon(false)
.build(),
new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略
);
// 提交任务
List<Future<String>> futures = new ArrayList<>();
for (int i = 0; i < 20; i++) {
final int taskId = i;
Future<String> future = executor.submit(() -> {
Thread.sleep(100);
return "Task " + taskId + " completed";
});
futures.add(future);
}
// 获取结果
for (Future<String> future : futures) {
try {
System.out.println(future.get(5, TimeUnit.SECONDS));
} catch (TimeoutException e) {
System.err.println("Task timed out");
} catch (Exception e) {
System.err.println("Task failed: " + e.getMessage());
}
}
// 关闭线程池
executor.shutdown();
try {
if (!executor.awaitTermination(30, TimeUnit.SECONDS)) {
executor.shutdownNow();
}
} catch (InterruptedException e) {
executor.shutdownNow();
}
}
}9.2 CompletableFuture 实战
java
public class AsyncDemo {
// 串行执行
public CompletableFuture<String> fetchUser(String userId) {
return CompletableFuture.supplyAsync(() -> {
// 模拟网络请求
sleep(100);
return "User:" + userId;
});
}
public CompletableFuture<List<String>> fetchOrders(String userId) {
return CompletableFuture.supplyAsync(() -> {
sleep(200);
return List.of("Order1", "Order2");
});
}
// 组合多个异步操作
public CompletableFuture<String> getUserWithOrders(String userId) {
return fetchUser(userId)
.thenCombine(fetchOrders(userId), (user, orders) -> {
return user + " has " + orders.size() + " orders";
});
}
// 并行执行多个任务
public CompletableFuture<List<String>> fetchAllUsers(List<String> userIds) {
List<CompletableFuture<String>> futures = userIds.stream()
.map(this::fetchUser)
.collect(Collectors.toList());
return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
.thenApply(v -> futures.stream()
.map(CompletableFuture::join)
.collect(Collectors.toList()));
}
// 异常处理
public CompletableFuture<String> fetchWithFallback(String userId) {
return fetchUser(userId)
.exceptionally(ex -> {
System.err.println("Failed: " + ex.getMessage());
return "User:unknown";
});
}
private void sleep(long ms) {
try { Thread.sleep(ms); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
}
}四、本章小结
第9章系统讲解了并发编程的核心知识：
同步共享可变数据：使用 synchronized 保证原子性和可见性。
避免过度同步：不要在同步块内调用外部方法。
优先使用并发工具：BlockingQueue、ConcurrentHashMap 等。
线程安全优于文档：确保类是真正线程安全的。
延迟初始化要慎重：使用双重检查锁或静态内部类。
不要依赖线程调度器：使用并发工具协调线程。
并发编程是 Java 中最复杂的主题。记住：正确的并发代码不是写出来的，而是设计出来的。
五、思考与练习
思考题
synchronized 的锁是什么？实例方法锁的是 this，静态方法锁的是 Class。
volatile 能保证原子性吗？它和 synchronized 有什么区别？
ConcurrentHashMap 是如何实现线程安全的？
线程池的核心参数有哪些？如何合理配置？
CompletableFuture 相比 Future 有什么优势？
编程练习
实现线程安全的单例：使用双重检查锁或枚举。
实现生产者-消费者：使用 BlockingQueue。
实现异步数据聚合：使用 CompletableFuture 并行获取多个数据源并聚合结果。