本节目标：让队列成为「可计算的等待室」，而不是「看不见的内存黑洞」；过载时明确拒绝，并让客户端通过 gRPC 流控感知到压力。
适用版本：JDK 25 + gRPC-Java 1.76.0（grpc-netty-shaded）。
承接的问题
8.2 用拦截器把执行并发限制在 64，超出的请求直接拒绝。真实流量不是均匀的：每秒 200 个请求，可能前 100 ms 来了 60 个，后 900 ms 只来 140 个。如果第 65 个立刻拒绝，客户端会因为一个毫秒级的抖动收到 429；如果给它一个等待位，等 10 ms 就有线程空出来，请求可以正常完成。队列就是干这个的。
但队列一加，问题从「要不要排队」变成「排多长、排不下怎么办、排久了会不会超过 deadline」。linkd-v3 的默认值是执行器队列 128，配合 64 个执行线程。128 不是拍脑袋的数字：单跳 deadline 300 ms，服务时间按 50 ms 估算，64 个线程每秒能处理 1280 个请求；128 个等待位对应约 100 ms 的排队时间，加上 50 ms 服务时间仍在 300 ms 预算内。超过这个长度，请求就算最终被执行，也已经超过调用方的 deadline，排队没有意义。
队列的另一个作用是背压：当消费者处理不过来时，队列变满，生产者要么阻塞、要么被拒绝、要么收到流控信号。三种反应对应三种实现，选错了会把过载从一处转移到另一处。
核心机制一：有界队列与四种拒绝策略
ThreadPoolExecutor 的队列类型决定了过载行为：
| 队列 | 容量 | 过载表现 | 适用 |
|---|---|---|---|
ArrayBlockingQueue | 固定 | 满后触发拒绝策略 | 服务端请求队列，最常用 |
LinkedBlockingQueue | 可选，默认 Integer.MAX_VALUE | 默认几乎不拒绝，堆慢慢涨 | 不推荐做入口队列 |
SynchronousQueue | 0 | 没有等待位，直接交给线程或拒绝 | 高吞吐、低延迟、允许拒绝 |
PriorityBlockingQueue | 无界 | 按优先级取，仍会涨 | 内部调度，不适合入口 |
Executors.newFixedThreadPool(n) 用的是无界 LinkedBlockingQueue。这不是实现细节，而是最常见的生产事故来源：流量涨上来时它不会拒绝，只会把任务堆在堆里，直到 GC 频繁、延迟飙升，最后 OOM。入口队列必须显式构造 ThreadPoolExecutor 并传入有界队列。
队列满之后的动作由 RejectedExecutionHandler 决定：
| 策略 | 行为 | 评价 |
|---|---|---|
AbortPolicy | 抛 RejectedExecutionException | 默认，失败可见，推荐 |
CallerRunsPolicy | 提交任务的线程自己执行 | 在 gRPC 传输线程上会拖住整条连接，禁用 |
DiscardPolicy | 静默丢弃 | 客户端等到 deadline 也不知道请求没执行，禁用 |
DiscardOldestPolicy | 丢最旧的任务 | 丢掉的可能是已经等了很久的请求，禁用 |
| 自定义 | 计数 + 抛领域异常 | 推荐，能把拒绝映射成 429/503 |
linkd-v3 用自定义策略：先 rejected.incrementAndGet()，再抛 RejectedExecutionException。计数让指标能区分「队列满」和「拦截器拒绝」，异常让上层有机会把它映射成 RESOURCE_EXHAUSTED。不要吞掉异常返回一个假成功，那会让客户端以为请求已经受理。
核心机制二：BoundedDispatcher
把执行器、队列、拒绝计数包在一个类里，业务代码只看到 submit 和几个诊断方法。submit 返回 CompletableFuture，调用方可以组合，也可以在入口统一加超时。
package com.plume.linkd.v3.core;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
public final class BoundedDispatcher {
private final ThreadPoolExecutor executor;
private final AtomicLong rejected = new AtomicLong();
public BoundedDispatcher(int threads, int queueCapacity) {
if (threads < 1 || queueCapacity < 1) {
throw new IllegalArgumentException("线程数和队列容量必须为正");
}
this.executor = new ThreadPoolExecutor(
threads, threads, 0L, TimeUnit.MILLISECONDS,
new ArrayBlockingQueue<>(queueCapacity),
runnable -> {
Thread thread = new Thread(runnable, "linkd-dispatch");
thread.setDaemon(true);
return thread;
},
(runnable, pool) -> {
rejected.incrementAndGet();
throw new RejectedExecutionException("dispatch queue full");
});
}
public <T> CompletableFuture<T> submit(Callable<T> task) {
CompletableFuture<T> result = new CompletableFuture<>();
try {
executor.execute(() -> {
try {
result.complete(task.call());
} catch (Throwable failure) {
result.completeExceptionally(failure);
}
});
} catch (RejectedExecutionException e) {
result.completeExceptionally(e);
}
return result;
}
public int queueDepth() {
return executor.getQueue().size();
}
public int activeThreads() {
return executor.getActiveCount();
}
public long rejectedCount() {
return rejected.get();
}
public void shutdown() {
executor.shutdown();
}
}
threads 同时作为核心和最大线程数，理由和 8.2 一样：避免「先填队列、再扩线程」的顺序让实际并发低于配置。ArrayBlockingQueue 的容量是硬上限，队列满时自定义策略计数并抛异常。getQueue().size() 返回的是近似值，但足够画趋势；getActiveCount() 也是近似值，两者不能相加当成精确的在途数。
submit 里的 CompletableFuture 有一个容易忽略的语义：任务被拒绝时，future 以 RejectedExecutionException 完成；任务开始执行后，即使调用方对 future 调用 orTimeout，任务也不会停止。CompletableFuture.orTimeout 只改变 future 的状态，不中断线程，也不把任务从队列里移除。要真正取消，必须把任务提交成 Future 并调用 cancel(true)，或者让任务自己检查中断标志。这也是为什么队列等待时间应该在 submit 之前用 Bulkhead 或 queueDepth 控制，而不是提交后再靠 orTimeout 补救。
队列深度要和执行线程数一起看。64 个线程、128 个等待位，稳态吞吐是 64 / 平均服务时间。服务时间 50 ms 时是 1280 req/s；如果流量只有 200 rps，队列几乎不会超过几格，128 显得浪费；如果流量涨到 2000 rps，队列会在几秒内填满，然后开始拒绝。队列长度应该按「目标吞吐下的突发窗口」设置，不是按「内存能放多少」设置。
核心机制三：把 gRPC 入站流控接到队列上
执行器队列是服务端自己的队列，客户端看不见。gRPC 的 HTTP/2 流控是另一层：客户端发送受服务端的接收窗口约束，服务端不 request，客户端就发不动。把两层接起来，才能形成端到端背压。
| 机制 | 方向 | 作用 |
|---|---|---|
disableAutoRequest() | 入站 | 关闭自动请求，改为手动 request(n) |
request(n) | 入站 | 告诉客户端「还可以再发 n 条」 |
isReady() | 出站 | 当前发送缓冲区是否有空间 |
setOnReadyHandler(Runnable) | 出站 | 从满翻到不满时恢复生产 |
setOnReadyThreshold(int) | 出站 | 以字节为单位设置恢复水位 |
入站流控的典型用法是 Chat：服务端每处理完一条消息才 request(1)，客户端就不会无限发送。如果处理逻辑本身要排队，队列满时停止 request，客户端的发送窗口耗尽后自然停下；这比服务端把消息收进来再丢掉更省内存，也让客户端能感知到压力。
package com.plume.linkd.v3.server;
import com.plume.linkd.v3.api.ChatMessage;
import com.plume.linkd.v3.api.LibraryGrpc;
import io.grpc.Status;
import io.grpc.stub.ServerCallStreamObserver;
import io.grpc.stub.StreamObserver;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
public final class ChatService extends LibraryGrpc.LibraryImplBase {
private static final ChatMessage POISON =
ChatMessage.newBuilder().setFrom("__close__").build();
@Override
public StreamObserver<ChatMessage> chat(StreamObserver<ChatMessage> responseObserver) {
ServerCallStreamObserver<ChatMessage> out =
(ServerCallStreamObserver<ChatMessage>) responseObserver;
ChatInbound inbound = new ChatInbound(out);
out.setOnCancelHandler(inbound::cancel);
inbound.start();
return inbound;
}
private static final class ChatInbound implements StreamObserver<ChatMessage> {
private final ServerCallStreamObserver<ChatMessage> out;
private final ArrayBlockingQueue<ChatMessage> queue = new ArrayBlockingQueue<>(128);
private final Object outLock = new Object();
private volatile boolean closed;
private volatile boolean failed;
private volatile boolean cancelled;
ChatInbound(ServerCallStreamObserver<ChatMessage> out) {
this.out = out;
}
void start() {
out.disableAutoRequest();
Thread worker = new Thread(this::consume, "chat-worker");
worker.setDaemon(true);
worker.start();
out.request(1);
}
@Override
public void onNext(ChatMessage message) {
if (closed) {
return;
}
if (!queue.offer(message)) {
synchronized (outLock) {
failed = true;
closed = true;
out.onError(Status.RESOURCE_EXHAUSTED
.withDescription("chat inbound queue full")
.asRuntimeException());
}
return;
}
out.request(1);
}
@Override
public void onError(Throwable failure) {
failed = true;
closed = true;
queue.offer(POISON);
}
@Override
public void onCompleted() {
closed = true;
queue.offer(POISON);
}
void cancel() {
cancelled = true;
closed = true;
queue.offer(POISON);
}
private void consume() {
while (true) {
ChatMessage message;
try {
message = queue.poll(100, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return;
}
if (message == POISON) {
finish();
return;
}
if (message != null) {
synchronized (outLock) {
if (!failed) {
out.onNext(reply(message));
}
}
} else if (closed && queue.isEmpty()) {
finish();
return;
}
}
}
private void finish() {
synchronized (outLock) {
if (failed || cancelled) {
return;
}
out.onCompleted();
}
}
private ChatMessage reply(ChatMessage message) {
return ChatMessage.newBuilder()
.setFrom("libraryd")
.setText("echo: " + message.getText())
.setSentAt(System.currentTimeMillis())
.build();
}
}
}
disableAutoRequest() 之后必须自己 request，否则客户端一条消息也发不进来。start() 里先 request(1)，每成功入队一条再 request(1)，队列满时不再请求；客户端的发送窗口会在若干条之后耗尽，形成背压。这个「窗口」不是队列容量本身，而是 gRPC 的流控窗口，可能大于或小于 128，所以不能把 request(1) 当成精确的队列控制。
queue.offer 失败时直接 onError(RESOURCE_EXHAUSTED)，而不是阻塞等待。阻塞在 onNext 里会占住 gRPC 的入站线程，反而让流控失效。队列满是一个明确的过载信号，应该快速失败并记录；如果业务希望等待，可以把 offer 换成带超时的 offer(message, 10, MILLISECONDS)，但等待时间要算进总 deadline。
POISON 是一个哨兵对象，用来唤醒阻塞在 poll 上的消费者。onError、onCompleted、cancel 三条路径都把 closed 置真并投递哨兵，消费者看到哨兵就退出。没有哨兵时，消费者最多等 100 ms 的超时，退出会慢一点，但不会泄漏；加上哨兵是为了让关闭更及时。
出站方向的 onNext 也要检查 isReady()。上面的 consume 直接调用 out.onNext(reply)，在回复很小、客户端读得快时没问题；如果回复很大或客户端读得慢，应该像 8.2 的 ReadyPump 一样，在 isReady() 为 false 时把回复放进一个有界出站队列，等 onReadyHandler 再发。入站队列和出站队列是两套独立的背压，不能只做一边。
在 linkd-v3 里落地：队列预算与指标
linkd-v3 的链路预算是：
入口令牌桶：200 rps，burst 400
执行线程：64
等待队列：128
单跳 deadline：300 ms
总 deadline：800 ms
按 Little’s law，并发 = 吞吐 × 延迟。64 个线程、平均服务时间 50 ms，稳态吞吐是 1280 req/s；队列 128 个等待位对应的排队时间是 128 / 1280 ≈ 100 ms。请求在队列里等 100 ms、执行 50 ms，单跳共 150 ms，仍在 300 ms 预算内。如果服务时间涨到 150 ms，吞吐降到 426 req/s，128 个等待位的排队时间变成 300 ms，加上执行时间已经 450 ms，超过单跳预算；这时队列再长也没有意义，应该让请求快速失败，或者扩容。
队列长度不是越大越好。队列越大，请求在队列里待的时间越长，越容易在轮到自己之前就超过 deadline；这些「已经死了」的请求仍然占着队列位置，把真正有机会完成的请求挤出去。队列的作用是吸收「毫秒级抖动」，不是「秒级堆积」。超过 100 ms 的排队，就要考虑拒绝而不是继续等。
指标至少要有四个：
dispatcher_queue_depth：当前队列长度，按 route 或方法打标签。dispatcher_active_threads：当前执行线程数。dispatcher_rejected_total：队列满的拒绝数，按原因打标签。dispatcher_queue_wait_seconds：从入队到开始执行的等待时间直方图。
前两个是 gauge，后两个是 counter 和 histogram。只看队列深度不够：队列深度 100、执行线程 64、服务时间 10 ms，可能马上就能排空；队列深度 20、执行线程 64、服务时间 2 秒，已经在崩的边缘。深度要和等待时间一起看。
常见错误与边界
错误一：用无界队列。 Executors.newFixedThreadPool、new LinkedBlockingQueue<>() 都不会拒绝，过载时堆内存一路涨到 OOM。入口队列必须 ArrayBlockingQueue 并设硬上限。
错误二：CallerRunsPolicy 放在 gRPC 服务端。 队列满时，提交任务的线程会自己执行任务；在 gRPC 里这个线程可能是传输线程，执行一个慢调用会把整条连接的其他流一起拖住。用 AbortPolicy 或自定义策略。
错误三：DiscardPolicy 静默丢任务。 客户端会一直等到 deadline，然后拿到 DEADLINE_EXCEEDED，完全不知道请求根本没执行。丢弃必须伴随计数和明确的错误响应。
错误四：用 orTimeout 取消排队任务。 orTimeout 只改变 future 状态，任务还在队列里，轮到它时照样执行。要取消必须拿到 Future 并 cancel(true)，或者让任务自己检查中断。
错误五：在 onNext 里阻塞等待队列空位。 gRPC 的入站回调运行在传输/调度线程上，阻塞会让整个流的消息处理停住。用 offer 快速失败，或者带很短超时的 offer，并把等待算进 deadline。
错误六：入站流控开了 disableAutoRequest 却忘记 request。 客户端一条消息也发不进来，表现为「连接建立成功、没有任何数据、最后 deadline 超时」。disableAutoRequest 之后必须在合适时机 request(n)。
错误七：一次 request(Integer.MAX_VALUE)。 这等于关闭入站流控，客户端可以无限发送，队列很快被打满。request(n) 的 n 要按处理能力和队列容量设置。
边界：队列不跨实例。 每个实例 128 个等待位，10 个实例是 1280 个；下游看到的总排队量是实例数乘以单机队列。队列容量要和实例数、下游容量一起算。
边界：队列深度是近似值。 getQueue().size() 在并发下只是一个瞬间读数，不能用来做精确的准入控制。精确控制要用 Bulkhead 的许可或拦截器的原子计数。
边界：流控窗口不等于队列容量。 HTTP/2 的流控窗口由连接和流的窗口共同决定，可能远大于 128；request(1) 只是「再要一条」，不代表客户端只会发一条。队列仍然要有硬上限，不能依赖流控窗口兜底。
边界：队列不能修复慢下游。 下游服务时间从 50 ms 涨到 500 ms，队列只会把请求堆得更久，最后一起超时。慢下游要靠熔断、隔离和扩容，队列只是吸收抖动。
小结
- 入口队列必须是有界
ArrayBlockingQueue；无界队列把过载藏成 OOM。 - 拒绝策略用
AbortPolicy或自定义「计数 + 抛异常」，不要用CallerRunsPolicy和DiscardPolicy。 core = max避免「先填队列再扩线程」，64 线程 + 128 等待位是linkd-v3的基线。orTimeout不取消队列里的任务；队列等待要在提交前用许可或深度控制。- 入站流控用
disableAutoRequest()+request(n)，队列满时停止请求；出站用isReady()+onReadyHandler，两边独立。 - 队列深度要和等待时间、执行线程数一起看；超过 100 ms 的排队已经接近 300 ms 单跳预算，应该拒绝或扩容。
第 8 章把入口的三层——速率、并发、队列——讲完了。下一章换到系统边界：外部客户端说 HTTP，内部服务说 gRPC，9.1 用 HttpServer 和 HttpClient 写出第一版协议转换网关。
阅读导航：上一节 8.2 服务端并发限制 ｜ 下一节 9.1 HTTP 到 gRPC 的网关
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。