第2章：面向对象并发（Object-Oriented Concurrency）

一、导读

1.1 本章学习目标

第2章介绍 Java 的 java.util.concurrent 包——面向对象并发模型的代表。它通过高级并发工具（线程池、Future、并发集合）封装了底层的线程与锁。

本章的核心学习目标包括：

掌握 Executor 框架：线程池、任务提交
理解 Future 和 CompletableFuture
学会使用并发集合
理解 Fork/Join 框架

1.2 为什么本章如此重要

java.util.concurrent 是 Java 并发编程的标准工具集。它封装了底层的线程与锁，提供了更高层次的抽象。理解这些工具，能让你在 Java 生态中写出高效的并发代码。

1.3 核心问题引导

Executor 框架如何简化线程管理？

Future 和 CompletableFuture 有什么区别？

ConcurrentHashMap 是如何实现高并发读写的？

Fork/Join 框架适合什么类型的问题？

二、核心概念详解

2.1 Executor 框架

java

// 线程池
ExecutorService executor = Executors.newFixedThreadPool(4);

// 提交任务
Future<String> future = executor.submit(() -> {
    Thread.sleep(1000);
    return "Result";
});

String result = future.get();  // 阻塞等待

// 批量提交
List<Callable<String>> tasks = List.of(
    () -> "Task 1",
    () -> "Task 2",
    () -> "Task 3"
);

List<Future<String>> futures = executor.invokeAll(tasks);
for (Future<String> f : futures) {
    System.out.println(f.get());
}

executor.shutdown();

2.2 CompletableFuture

java

// 链式异步操作
CompletableFuture.supplyAsync(() -> fetchUser(1))
    .thenApply(user -> user.getName())
    .thenApply(name -> name.toUpperCase())
    .thenAccept(System.out::println);

// 组合多个异步操作
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "Hello");
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "World");

future1.thenCombine(future2, (a, b) -> a + " " + b)
    .thenAccept(System.out::println);

// 异常处理
CompletableFuture.supplyAsync(() -> {
    if (Math.random() > 0.5) throw new RuntimeException("Error");
    return "OK";
}).exceptionally(ex -> "Fallback: " + ex.getMessage())
  .thenAccept(System.out::println);

2.3 并发集合

java

// ConcurrentHashMap
ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();
map.put("a", 1);
map.compute("a", (k, v) -> v == null ? 1 : v + 1);

// ConcurrentLinkedQueue
ConcurrentLinkedQueue<String> queue = new ConcurrentLinkedQueue<>();
queue.offer("item1");
String item = queue.poll();

// CopyOnWriteArrayList
CopyOnWriteArrayList<String> list = new CopyOnWriteArrayList<>();
list.add("item");  // 写时复制整个数组

三、实战案例

3.1 案例一：并行流处理

java

List<Integer> numbers = IntStream.range(1, 1000000).boxed().collect(Collectors.toList());

// 并行流
long sum = numbers.parallelStream()
    .filter(n -> n % 2 == 0)
    .mapToLong(n -> n * n)
    .sum();

// Fork/Join
class SumTask extends RecursiveTask<Long> {
    private final int[] array;
    private final int start, end;
    
    protected Long compute() {
        if (end - start <= 1000) {
            long sum = 0;
            for (int i = start; i < end; i++) sum += array[i];
            return sum;
        }
        int mid = (start + end) / 2;
        SumTask left = new SumTask(array, start, mid);
        SumTask right = new SumTask(array, mid, end);
        left.fork();
        long rightResult = right.compute();
        long leftResult = left.join();
        return leftResult + rightResult;
    }
}

四、本章小结

Executor 框架：封装线程管理，提供线程池。

CompletableFuture：链式异步编程，组合多个异步操作。

并发集合：ConcurrentHashMap、ConcurrentLinkedQueue 等线程安全集合。

Fork/Join：分治并行计算框架。

五、思考与练习

CompletableFuture 和回调地狱有什么关系？它如何解决异步编程的复杂性？

ConcurrentHashMap 的并发性能为什么比 Hashtable 好？

并行流在什么场景下能带来性能提升？什么场景下反而更慢？