第 31 章 并发工具、线程池与 AQS

1
2
3
4
5
6
7

public ThreadPoolExecutor(int corePoolSize,     // 常驻线程数下限，空闲也不回收
                          int maximumPoolSize,  // 线程数上限，只在队列满时才被逼近
                          long keepAliveTime,   // 超出 corePoolSize 的空闲线程存活多久后退出
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue, // 满 core 后任务的第一个去处
                          ThreadFactory threadFactory,       // 造线程的地方，命名与异常处理器在此挂
                          RejectedExecutionHandler handler)  // 线程数与队列都满时的最后一道闸

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

        execute(task)
             ▼
① workerCount < corePoolSize ?
        ├── 是 ──▶ addWorker(task, true)   新建【核心】线程直接执行该任务，不入队
        └── 否 ▼
② workQueue.offer(task) 成功 ?
        ├── 是 ──▶ 入队；再校验池状态与线程数，必要时补一个线程
        └── 否（队列满）▼
③ workerCount < maximumPoolSize ?
        ├── 是 ──▶ addWorker(task, false)  新建【非核心】线程执行该任务
        └── 否 ▼
④ reject(task) → handler.rejectedExecution(task, this)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

// 拒绝路径由提交线程同步执行，所以这里只打点、不阻塞、不写重日志
public class BusinessRejectHandler implements RejectedExecutionHandler {
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        REJECT_COUNTER.increment();          // 拒绝率本身就是最该告警的指标
        if (r instanceof RetryableTask) {    // 核心任务抛出让上游重试
            throw new RejectedExecutionException("order pool saturated");
        }
        Metrics.markDegraded();              // 非核心任务降级，绝不静默丢失
    }
}

1
2
3
4
5
6
7
8

// 有界队列 + 有界线程数 + 命名线程 + 自定义拒绝，四个"有界"缺一不可
ThreadPoolExecutor pool = new ThreadPoolExecutor(
        8,                              // 核心数按实测拐点设，不按核数拍
        32,                             // 上限留突发余量，到达上限即拒绝，绝不无限扩张
        60L, TimeUnit.SECONDS,          // 非核心线程空闲 60 s 回收，缩容靠它
        new ArrayBlockingQueue<>(500),  // 必须显式给容量，让"满"成为可观测事件
        new NamedThreadFactory("order-io-"),   // 命名 + 守护线程 + 异常兜底
        new BusinessRejectHandler());   // 拒绝时记录业务标识与拒绝量，不做重操作

1
2
3
4
5
6

设单任务：计算时间 W_c，等待时间 W_w（网络、磁盘、下游 RPC）
阻塞系数（blocking coefficient）：p = W_w / (W_c + W_w)
要让"等待期间 CPU 不空转"，并发线程数需放大约 1/(1 − p) 倍：
        N_threads = N × 1/(1 − p) = N × (W_c + W_w)/W_c = N × (1 + W_w/W_c)
算例：8 核、单任务 50 ms（计算 5 ms、等待 45 ms）⟹ N_threads = 8 × (1 + 45/5) = 80
校验：80 个线程里平均只有 80 × (5/50) = 8 个在计算，恰好铺满 8 核

1
2
3
4

        L = λ × W
L = 系统中的平均请求数（即并发数）    λ = 到达率（QPS）    W = 单个请求的平均逗留时间（RT）
线程池是瓶颈时，所需线程数 ≈ 并发数 = 目标 QPS × 平均 RT
算例：目标 2000 QPS、平均 RT 50 ms = 0.05 s ⟹ L = 2000 × 0.05 = 100

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

// 钩子在工作线程内执行，因此能做 ThreadLocal 与耗时统计而无需额外同步
@Override
protected void beforeExecute(Thread t, Runnable r) {
    WAIT_HISTOGRAM.record(System.nanoTime() - ((TaskWrapper) r).enqueueNanos); // 排队耗时
    MDC.put("taskId", ((TaskWrapper) r).id);
}
@Override
protected void afterExecute(Runnable r, Throwable t) {
    MDC.clear(); // 线程会被复用，不清理就会把上一个任务的上下文留给下一个任务
}

1
2
3
4
5
6
7
8
9

pool.submit(() -> { throw new IllegalStateException("boom"); }); // 反例：异常既不到日志也不触发钩子
Future<?> f = pool.submit(() -> doSomething());
try {
    f.get(3, TimeUnit.SECONDS);
} catch (ExecutionException e) {
    log.error("async task failed", e.getCause()); // 必须取 getCause()，外层只是包装
} catch (TimeoutException e) {
    f.cancel(true); // cancel 只是发中断，任务不响应中断就仍在跑，见 31.7 节
}

1
2
3
4
5
6
7
8
9

                       AbstractQueuedSynchronizer
  volatile int state;          ← 语义由子类解释，AQS 只提供 CAS 与读写
  Thread exclusiveOwnerThread; ← 独占模式记录持有者，支持重入判定
  Node head;  Node tail;       ← 同步队列（双向），懒初始化
  独占语义：state 0 = 空闲；1 = 已占用；n = 同一线程重入 n 次
            ReentrantLock.tryAcquire → compareAndSetState(0, 1)；
            tryRelease 把 state 减到 0 才算真正释放（重入时返回 false，故不唤醒后继）
  共享语义：state = 可用许可数，获取递减、释放递增
            Semaphore(3) 初始 state=3；CountDownLatch(5) 初始 state=5，countDown 减 1

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

        head                                                     tail
   ┌───────────┐   prev   ┌───────────┐   prev   ┌───────────┐
   │  Node(哑) │◀────────│  Node(T2) │◀────────│  Node(T3) │
   │ ws = 0    │ ───────▶│ ws=SIGNAL │ ───────▶│ ws=SIGNAL │   next
   └───────────┘   next   └───────────┘   next   └───────────┘
   已释放/当前持有者          排队中，将要被唤醒        队尾，刚入队
  Node.waitStatus（int 常量，新节点默认 0）：
    CANCELLED  =  1  节点已放弃（超时/中断/异常），必须从队列中剔除
    SIGNAL     = -1  本节点释放时【有义务】唤醒 next；后继 park 前的"约定"
    CONDITION  = -2  该节点当前在【条件队列】里，不在同步队列上
    PROPAGATE  = -3  共享模式下，唤醒仍需继续往后传播

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

// 模板方法：排队、阻塞、中断、超时的逻辑全在基类，子类只实现"能不能拿到"
public final void acquire(int arg) {
    if (!tryAcquire(arg)                                  // ① 快路径
        && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) // ② 入队 ③ 队列内循环
        selfInterrupt();  // 不在这里抛：把锁语义与中断语义分开，只在最后补中断标记
}
final boolean acquireQueued(final Node node, int arg) {
    boolean interrupted = false;
    for (;;) {
        final Node p = node.predecessor();
        // 只有前驱是 head 才有资格抢锁：head 代表已释放，可省掉无谓的 park/unpark
        if (p == head && tryAcquire(arg)) {
            setHead(node);   // 自己成为 head；断开 p.next 避免正向遍历看到脏指针
            p.next = null;
            return interrupted;
        }
        // 必须"先置 SIGNAL 再 park"：顺序反了，释放方的 unpark 会早于 park 而丢弃许可
        if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) {
            interrupted = true;  // 被中断不退出循环，只记账 —— 这就是"不可中断获取"
        }
    }
}
public final boolean release(int arg) {
    if (!tryRelease(arg)) {       // 子类：state 减到 0 才返回 true（重入时返回 false，不唤醒后继）
        return false;
    }
    if (head != null && head.waitStatus != 0) {
        unparkSuccessor(head);    // head 的 ws 为 0 说明没有后继在等，无需唤醒
    }
    return true;
}
private void unparkSuccessor(Node node) {
    if (node.waitStatus < 0) {
        node.compareAndSetWaitStatus(node.waitStatus, 0); // 清 SIGNAL：唤醒义务已履行
    }
    Node s = node.next;
    if (s == null || s.waitStatus > 0) {   // 后继不存在或已取消，从尾部反向找有效节点
        for (Node p = tail; p != node && p != null; p = p.prev) {
            if (p.waitStatus <= 0) {
                s = p;
            }
        }
    }
    if (s != null) {
        LockSupport.unpark(s.thread);      // unpark 早于对方 park 也有效，许可不会丢
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

private void doReleaseShared() {
    for (;;) {
        Node h = head;
        if (h != null && h != tail) {          // 队列中至少还有一个后继节点
            if (h.waitStatus == Node.SIGNAL) {
                if (h.compareAndSetWaitStatus(Node.SIGNAL, 0)) {
                    unparkSuccessor(h);        // 交给被唤醒者继续传播（它会再触发本方法）
                }
                continue;                      // CAS 失败说明别人改过状态，重读再判断
            } else if (h.waitStatus == 0 && !h.compareAndSetWaitStatus(0, Node.PROPAGATE)) {
                continue;                      // 关键：把 0 改成 PROPAGATE，而不是"什么都不做"
            }
        }
        if (h == head) {                       // head 没变说明传播已到队尾，退出
            break;
        }
    }
}

1
2
3
4
5
6
7

      同步队列（双向，抢锁用）                       条件队列（单向，等条件用）
   head ──▶ [T2] ⇄ [T3] ──▶ tail        firstWaiter ──▶ [T7] ─▶ [T8] ──▶ lastWaiter
                                          ws=CONDITION，用 nextWaiter 串联
                     ▲                                        │
                     │   await: 释放锁 + 入条件队列 + park
                     │   signal: 把首节点摘下，用 CAS 接回同步队列尾部（不转移锁）
                     └────────────────────────────────────────┘

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

// 非公平：不检查队列，上来就抢一次 —— 插队就发生在这一行
final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    if (getState() == 0 && compareAndSetState(0, acquires)) { // 完全无视队列中已等待的线程
        setExclusiveOwnerThread(current);
        return true;
    }
    if (current == getExclusiveOwnerThread()) {
        setState(getState() + acquires);   // 重入 count++：只有持有者能走到这里，无需 CAS
        return true;
    }
    return false;
}
// 公平：唯一的差别是"锁空闲时还要求队列里没有前辈"
protected final boolean tryAcquire(int acquires) {
    if (getState() != 0 || hasQueuedPredecessors() || !compareAndSetState(0, acquires)) {
        return false;   // 队列里有前辈就退出，不去抢那一行 CAS
    }
    setExclusiveOwnerThread(Thread.currentThread());
    return true;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

// CountDownLatch：主线程等初始化步骤；计数只减不增，要复位必须新建对象
CountDownLatch latch = new CountDownLatch(3);
pool.execute(() -> { try { init(svc); } finally { latch.countDown(); } }); // finally 里减：抛异常也要放行
latch.await(5, TimeUnit.SECONDS);  // 必须带超时：子任务卡死时无超时的 await 会永久拖住启动流程
// CyclicBarrier：线程互等，最后一个到达者执行汇总；可重用，适合分代迭代
new CyclicBarrier(4, () -> mergePartialResults()).await(); // 任一人超时会让其他人收到 BrokenBarrierException
// Semaphore：限流，是"线程数公式"的运行期版本
Semaphore permits = new Semaphore(50);   // 并发 50、单请求 100 ms ⇒ 吞吐上限 50/0.1 = 500 QPS
permits.acquire();                       // 拿不到就排队，天然形成对下游的背压
try { callDownstream(); } finally { permits.release(); } // 必须 finally 释放，否则永久少一个许可