8.2 排队与协调：同步、通信与线程池

8.2 排队与协调：同步、通信与线程池

本节摘要：wait 与 notifyAll 配合 while 条件实现窗口间的叫号协调——生产者消费者是标准模型。线程池是调度台：固定窗口数加任务队列，避免为每个任务新建窗口的开销；核心参数是窗口数、排队容量与拒绝策略，提交用 execute 与 submit，关闭用 shutdown 与 shutdownNow。Callable 是带回执的任务规程，Future 是取回执的凭据。本节实现生产者消费者并用线程池跑一批任务。

窗口之间怎么递材料

8.1 节的锁解决"不互相踩"，但只靠锁写不出"一个窗口产出、另一个窗口消费"的配合——材料没到时消费窗口该等，材料到了该有人叫醒它。这套协调机制就是 wait 与 notifyAll，它们必须写在 synchronized 块里（wait 的前提是持有锁），配对的铁律是"条件判断用 while 循环、唤醒用 notifyAll"。经典模型是生产者消费者：

import java.util.ArrayDeque;
import java.util.Deque;

class Counter {                                  // 叫号台：窗口间递材料的中间站
    private final Deque<Integer> tray = new ArrayDeque<>();
    private final int capacity = 3;              // 托盘容量：最多压三份材料

    synchronized void put(int item) throws InterruptedException {
        while (tray.size() == capacity) {        // 满：等 为什么用 while 见下方解读
            wait();                              // 放手锁 挂起 等消费窗口叫号
        }
        tray.addLast(item);
        System.out.println("生产了 " + item + " 剩余 " + tray.size());
        notifyAll();                             // 叫醒所有等待者：有材料了
    }

    synchronized int take() throws InterruptedException {
        while (tray.isEmpty()) {                 // 空：等
            wait();
        }
        int item = tray.removeFirst();
        System.out.println("消费了 " + item + " 剩余 " + tray.size());
        notifyAll();                             // 叫醒所有等待者：有空间了
        return item;
    }
}

public class PcDemo {
    public static void main(String[] args) throws InterruptedException {
        Counter counter = new Counter();
        Thread producer = new Thread(() -> {     // 生产窗口：连续投放
            try {
                for (int i = 1; i <= 5; i++) counter.put(i);
            } catch (InterruptedException e) {
                System.out.println("生产窗口被中断 收工");
            }
        });
        Thread consumer = new Thread(() -> {     // 消费窗口：慢速取用
            try {
                for (int i = 1; i <= 5; i++) {
                    Thread.sleep(80);            // 模拟消费比生产慢
                    counter.take();
                }
            } catch (InterruptedException e) {
                System.out.println("消费窗口被中断 收工");
            }
        });
        producer.start();
        consumer.start();
        producer.join();
        consumer.join();
        System.out.println("全部材料交接完毕");
    }
}
​

运行输出可见生产三条后被托盘容量卡住、等消费窗口取走再继续——协调生效。为什么条件判断必须用 while 不能用 if：被叫醒的窗口从 wait 处恢复、重新拿锁，但从"被叫醒"到"拿到锁"之间，条件可能又被别的窗口改掉（别人抢先消费了）——这叫虚假唤醒，while 会在恢复后重查条件、不满足继续等，if 则会带着过期的假设往下走。为什么 notifyAll 而不是 notify：notify 只叫醒一个，但它不区分叫醒的是同类还是异类窗口——生产窗口可能叫醒的也是生产窗口，白叫一次还可能全员饿死；notifyAll 叫醒所有人让大家重查条件，正确性有保障（代价是惊群，高效场景有专门的高级工具替代，本教程不展开）。

线程池：窗口调度台

为每个任务 new 一个窗口、办完就拆，开销大且窗口数量失控——高并发下窗口数能冲垮系统。线程池的逻辑是调度台：固定开几个窗口，任务提交进队列，窗口办完一个自动取下一个。创建池最直观的方式是手动给参数：

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;

public class PoolDemo {
    public static void main(String[] args) throws Exception {
        // 手动三参数：窗口数二到四 队列容量五十 满了之后调用方自己执行
        ExecutorService pool = new java.util.concurrent.ThreadPoolExecutor(
                2, 4, 60, TimeUnit.SECONDS,
                new java.util.concurrent.LinkedBlockingQueue<>(50),
                new java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy());

        // execute：提交无回执任务
        for (int i = 1; i <= 3; i++) {
            final int no = i;
            pool.execute(() -> System.out.println("窗口 " +
                    Thread.currentThread().getName() + " 办理任务 " + no));
        }

        // submit：提交 Callable 带回执的任务 拿 Future 取结果
        List<Future<Long>> receipts = new ArrayList<>();
        for (int i = 1; i <= 4; i++) {
            final int n = i * 10;
            receipts.add(pool.submit(() -> {      // Callable 有返回值 还能抛受检异常
                long sum = 0;
                for (int k = 1; k <= n; k++) sum += k;
                return sum;                      // 这就是回执上的结果
            }));
        }
        for (Future<Long> f : receipts) {
            System.out.println("回执结果：" + f.get());   // get 阻塞到办结
            // 输出四行：55 210 465 820（一到十 二十 三十 四十 的累加和）
        }

        pool.shutdown();                          // 温和关闭：不收新任务 办完存量后谢幕
        // pool.shutdownNow();                    // 强硬关闭：尝试中断在办任务
    }
}
​

Callable 与 Runnable 的差别正好对应"要不要回执"：Runnable 的 run 无返回值、抛不了受检异常（第 7 章的约束在规程上的体现——异常只能在规程内部消化）；Callable 的 call 有返回值、能抛受检异常，submit 后拿到 Future，get 时阻塞取结果。get 也支持带超时，避免无限等。

内置的几种工厂快捷方式各有用途：固定窗口数的池（任务量稳定）、缓存式池（窗口按需增减、空闲回收，适合大量短任务）、单窗口池（任务严格排队、保证顺序）、定时池（周期执行）。但有一个名声在外的坑必须提示：某个单窗口快捷池的队列无界——任务堆积没有上限，来多少排多少，内存迟早被队列撑爆；资源敏感的系统应手动建池、显式给出队列容量与拒绝策略，就像上面的三参数写法。

图 8-2 线程池调度台：窗口 队列 与回执

完整案例：批量档案核对

背景：档案科每晚要核对一百份档案，单份核对耗时数秒，串行要跑好几分钟；机器四核，希望压到几十秒且最后汇总每份的核对结果。操作：固定四窗口的池加 Callable 回执：

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class AuditCase {
    public static void main(String[] args) throws Exception {
        ExecutorService pool = Executors.newFixedThreadPool(4);   // 四窗口 与核数匹配
        List<Future<String>> receipts = new ArrayList<>();

        for (int id = 1; id <= 10; id++) {          // 十份档案做演示（实际一百份）
            final int docId = id;
            receipts.add(pool.submit(() -> {
                Thread.sleep(200);                  // 模拟单份核对耗时
                if (docId == 7) return "档案" + docId + " 缺页 需补录";
                return "档案" + docId + " 完整";
            }));
        }
        pool.shutdown();
        int problems = 0;
        for (Future<String> f : receipts) {         // 逐份收回执
            String result = f.get();
            if (!result.contains("完整")) problems++;
            System.out.println(result);
        }
        System.out.println("异常档案数：" + problems);   // 输出：异常档案数：1
    }
}
​

结果：十份档案约半秒跑完（串行需两秒），结果按提交顺序汇总，缺页档案被精确点名。解读：窗口数取核数附近是计算密集型任务的常用起点（过多窗口只增加切换开销）；Callable 让"并发执行"与"结果汇总"解耦——提交时不用等、收据齐全后统一取。变式：若要"完成一份处理一份"而不是最后统一收，用完成通知服务或对每个回执单独安排回调；若任务里有共享计数，回到 8.1 节的锁纪律，或用线程安全的结果容器替代普通集合（普通 ArrayList 被多线程同时 add 会丢数据，与计数丢数同理）。

本节要点回顾

• wait 与 notifyAll 必须在锁内：条件判断用 while 防虚假唤醒、唤醒用 notifyAll 防叫错人——生产者消费者是标准模型
• 线程池是调度台：固定窗口加任务队列，省掉逐任务建拆窗口的开销，也保住系统不被窗口数冲垮
• 核心参数与策略：窗口数、队列容量、拒绝策略都要显式给——某个内置单窗口池的无界队列是内存隐患
• execute 与 submit：前者无回执，后者提交 Callable 拿 Future，get 阻塞取结果可带超时；Callable 能返回值也能抛受检异常
• 关闭两档：温和档办完存量谢幕、强硬档中断在办任务；被中断的线程靠中断异常感知收工

多窗口的协调到此立住。最后一节换一个完全不同的视角：不再新开窗口，而是给工作人员"翻档案"的特权——反射让你在运行时拆开任何类，注解则是档案上的标签纸，两者合起来就是框架魔法的全部底细。