本节目标：在服务端把「同时处理多少个请求」变成一个明确的数字，并让超限的客户端拿到可解释、可重试的错误，而不是一个没有上下文的连接断开。
适用版本：JDK 25 + gRPC-Java 1.76.0（grpc-netty-shaded）。
承接的问题
8.1 的令牌桶控制入口速率：200 rps。速率合规不代表并发可控。一个 SearchBooks 请求如果下游卡住 2 秒，200 rps 的放行速度会在两秒内堆出 400 个在途请求；线程池、堆内存和下游连接都会被慢慢吃光。服务端还需要第二道闸：同一时刻最多处理 64 个请求，超出的直接拒绝。
这道闸放在哪里，决定了它是否有效。放在业务方法里，请求已经占用了线程，拒绝只是早点结束，救不了线程池；放在 ServerInterceptor 里，可以在 startCall 之前判断，超限时连业务代码都不进。linkd-v3 的选择是拦截器计数 64，执行器队列 128，两者分别对应「执行中」和「等待调度」。
先区分三个容易混淆的数字：
| 数字 | 含义 | 超限后的表现 |
|---|---|---|
| 64 | 拦截器统计的在途调用上限 | 返回 RESOURCE_EXHAUSTED + ErrorInfo |
| 128 | 执行器队列容量 | 队列满后 RejectedExecutionException |
| 200 rps | 入口令牌桶速率 | 返回 RESOURCE_EXHAUSTED + RetryInfo |
三者不是同一个限额，也不能用一个数字代替另外两个。8.1 管速率，本节管执行并发，8.3 管队列与流控。
核心机制一：在拦截器里精确计数
计数要满足三个条件：进入时加一、结束时减一、任何路径都不漏。gRPC 的 ServerCall.Listener 提供两个终止回调：onComplete 表示调用正常结束，onCancel 表示客户端取消或连接断开。两者在一次调用中只会发生一个，但实现里仍然建议用 AtomicBoolean 做幂等保护，避免未来包装层变化导致重复释放。
package com.plume.linkd.v3.server;
import com.google.protobuf.Any;
import com.google.rpc.ErrorInfo;
import io.grpc.ForwardingServerCall;
import io.grpc.Metadata;
import io.grpc.ServerCall;
import io.grpc.ServerCallHandler;
import io.grpc.ServerInterceptor;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
import io.grpc.protobuf.StatusProto;
import java.util.concurrent.atomic.AtomicInteger;
public final class ConcurrencyLimitInterceptor implements ServerInterceptor {
private static final Metadata.Key<String> TENANT =
Metadata.Key.of("x-tenant-id", Metadata.ASCII_STRING_MARSHALLER);
private final int maxConcurrent;
private final AtomicInteger inFlight = new AtomicInteger();
public ConcurrencyLimitInterceptor(int maxConcurrent) {
if (maxConcurrent < 1) {
throw new IllegalArgumentException("maxConcurrent 必须为正");
}
this.maxConcurrent = maxConcurrent;
}
@Override
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
ServerCall<ReqT, RespT> call,
Metadata headers,
ServerCallHandler<ReqT, RespT> next) {
if (inFlight.incrementAndGet() > maxConcurrent) {
inFlight.decrementAndGet();
reject(call, headers);
return new ServerCall.Listener<ReqT>() {
};
}
ServerCall.Listener<ReqT> delegate;
try {
delegate = next.startCall(
new ForwardingServerCall.SimpleForwardingServerCall<>(call) {
}, headers);
} catch (RuntimeException e) {
inFlight.decrementAndGet();
throw e;
}
return new ServerCall.Listener<ReqT>() {
private boolean released;
private void release() {
if (!released) {
released = true;
inFlight.decrementAndGet();
}
}
@Override
public void onMessage(ReqT message) {
delegate.onMessage(message);
}
@Override
public void onHalfClose() {
delegate.onHalfClose();
}
@Override
public void onReady() {
delegate.onReady();
}
@Override
public void onCancel() {
try {
delegate.onCancel();
} finally {
release();
}
}
@Override
public void onComplete() {
try {
delegate.onComplete();
} finally {
release();
}
}
};
}
public int inFlight() {
return inFlight.get();
}
private <ReqT, RespT> void reject(ServerCall<ReqT, RespT> call, Metadata headers) {
String tenant = headers.get(TENANT);
ErrorInfo info = ErrorInfo.newBuilder()
.setReason("RATE_LIMITED")
.setDomain("linkd.v3")
.putMetadata("limit", Integer.toString(maxConcurrent))
.putMetadata("tenant", tenant == null ? "" : tenant)
.build();
com.google.rpc.Status status = com.google.rpc.Status.newBuilder()
.setCode(Status.Code.RESOURCE_EXHAUSTED.value())
.setMessage("server concurrency limit reached")
.addDetails(Any.pack(info))
.build();
StatusRuntimeException error = StatusProto.toStatusRuntimeException(status);
call.close(error.getStatus(), error.getTrailers());
}
}
inFlight.incrementAndGet() > maxConcurrent 这一段先加再判，是为了让「判断是否超限」和「占用一个名额」成为一个原子动作。如果先 get() 再 incrementAndGet()，两个并发请求可能同时看到 63，然后都加到 64，实际变成 65。先加后判最多让超限的线程短暂把计数推到 65，然后立刻减回去，不会漏掉任何一次拒绝。
用 AtomicInteger 而不是 LongAdder：LongAdder.sum() 不是原子快照，在并发更新时可能读到一个中间值，用它做准入判断会让 64 这个上限漂移。LongAdder 适合累计计数（总请求数、总拒绝数），不适合表示「当前在途」这种 gauge。
拒绝时返回一个空的 ServerCall.Listener，并且只调用 call.close。不要返回 null，gRPC 会在后续回调里对 listener 解引用；也不要先 startCall 再关闭，业务代码已经执行就没有「提前拒绝」的意义了。
StatusProto.toStatusRuntimeException(com.google.rpc.Status) 会把整个 google.rpc.Status 序列化进 grpc-status-details-bin trailer；error.getTrailers() 拿到的就是这份 metadata。call.close(error.getStatus(), error.getTrailers()) 同时设置状态码、描述和 details，客户端用 StatusProto.fromThrowable 可以还原出 ErrorInfo。
核心机制二：64 个执行线程与 128 个等待位
拦截器只管「已经开始执行的调用」，管不到「还在执行器队列里排队」的调用。理解这一点，才能把 64 和 128 配对。
package com.plume.linkd.v3.server;
import io.grpc.Server;
import io.grpc.ServerBuilder;
import java.io.IOException;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public final class LibraryServer {
public static Server start(int port) throws IOException {
ThreadPoolExecutor executor = new ThreadPoolExecutor(
64, 64, 0L, TimeUnit.MILLISECONDS,
new ArrayBlockingQueue<>(128),
runnable -> {
Thread thread = new Thread(runnable, "linkd-grpc");
thread.setDaemon(true);
return thread;
},
new ThreadPoolExecutor.AbortPolicy());
return ServerBuilder.forPort(port)
.executor(executor)
.addService(new LibraryService())
.intercept(new ConcurrencyLimitInterceptor(64))
.maxInboundMessageSize(4 * 1024 * 1024)
.build()
.start();
}
}
corePoolSize 和 maximumPoolSize 都设 64，是为了避开 ThreadPoolExecutor 的扩容顺序：它先填满核心线程，再把任务塞进队列，队列满了才扩到最大线程。如果核心是 16、最大是 64、队列是 128，实际并发会长期停在 16，直到 128 个等待位全部占满才扩线程；这会让「64 并发」的配置名不副实。核心等于最大，语义最简单：64 个线程，128 个等待位，队列满就拒绝。
AbortPolicy 让队列满显式抛 RejectedExecutionException。不要用 CallerRunsPolicy：它会让提交任务的线程自己执行，而在 gRPC 服务端，这个线程通常是传输线程或调度线程，执行一个 2 秒的业务调用会把整条连接的其他流一起拖住。也不要用 DiscardPolicy，静默丢任务会让客户端等到 deadline，却不知道请求根本没执行。
ServerBuilder.executor(Executor) 是真实存在的 API，接受任意 Executor；这里传 ThreadPoolExecutor。ServerBuilder 没有「队列长度」参数，队列长度属于执行器，所以 128 必须在 ThreadPoolExecutor 的构造器里配置，不能在 ServerBuilder 上配置。
一个必须说明的边界：执行器队列里的调用还没有进入拦截器，因此不计入 64。极端情况下，进程里可能有 64 个执行中的调用加 128 个排队调用。队列满之后，AbortPolicy 抛出的异常由 gRPC 内部处理，客户端可能拿到 INTERNAL 或 UNAVAILABLE，而不是带 ErrorInfo 的 RESOURCE_EXHAUSTED。这是拦截器方案的固有局限：它只能管已经调度到业务线程的调用，管不到调度之前的队列。要严格在入口拒绝，需要传输层过滤器或自定义执行器，复杂度高得多；linkd-v3 选择接受这个边界，用 8.3 的队列指标和 429/503 监控来发现它。
核心机制三：流式响应的 isReady 与 onReadyHandler
并发限制管的是「同时多少个调用」，不解决「一个调用产生数据太快」。SearchBooks 一次返回上千本书，如果服务端不检查客户端是否读得动，onNext 会把消息堆在 gRPC 的发送缓冲区里，堆到流控窗口满了还在堆，内存一路涨。ServerCallStreamObserver 提供了三个工具：
| 方法 | 作用 |
|---|---|
isReady() | 当前是否适合继续 onNext，是提示不是保证 |
setOnReadyHandler(Runnable) | 从不可写翻回可写时回调，用来恢复生产 |
setOnReadyThreshold(int) | 以字节为单位设置恢复阈值，低于它 isReady() 翻回 true |
setOnCancelHandler(Runnable) | 客户端取消时停止生产者，释放资源 |
setOnReadyThreshold 的参数是字节数，不是消息条数。默认阈值由 gRPC 按流控窗口计算，通常在窗口的一半附近；显式设置 32 KiB 表示「待发字节降到 32 KiB 以下时再叫我」。阈值太小，onReady 会频繁触发；阈值太大，恢复太晚，吞吐下降。要按消息大小分布调，不是固定值。
package com.plume.linkd.v3.server;
import com.plume.linkd.v3.api.Book;
import com.plume.linkd.v3.api.LibraryGrpc;
import com.plume.linkd.v3.api.SearchBooksRequest;
import io.grpc.stub.ServerCallStreamObserver;
import io.grpc.stub.StreamObserver;
import java.util.Iterator;
import java.util.concurrent.atomic.AtomicBoolean;
public final class LibraryService extends LibraryGrpc.LibraryImplBase {
@Override
public void searchBooks(SearchBooksRequest request, StreamObserver<Book> observer) {
ServerCallStreamObserver<Book> out = (ServerCallStreamObserver<Book>) observer;
ReadyPump pump = new ReadyPump(out, search(request));
out.setOnReadyThreshold(32 * 1024);
out.setOnReadyHandler(pump::drain);
out.setOnCancelHandler(pump::cancel);
pump.drain();
}
private Iterator<Book> search(SearchBooksRequest request) {
return BookRepository.instance().search(request.getQuery(), request.getPageSize());
}
private static final class ReadyPump {
private final ServerCallStreamObserver<Book> out;
private final Iterator<Book> source;
private final AtomicBoolean draining = new AtomicBoolean();
private boolean done;
ReadyPump(ServerCallStreamObserver<Book> out, Iterator<Book> source) {
this.out = out;
this.source = source;
}
void drain() {
if (!draining.compareAndSet(false, true)) {
return;
}
try {
synchronized (this) {
if (done) {
return;
}
while (out.isReady() && source.hasNext()) {
out.onNext(source.next());
}
if (!source.hasNext()) {
done = true;
out.onCompleted();
}
}
} finally {
draining.set(false);
}
}
void cancel() {
synchronized (this) {
done = true;
}
}
}
}
isReady() 只是「当前缓冲区没满」的提示，不是「调用 onNext 一定不会缓冲」的保证。在 isReady() 为 true 和实际调用 onNext() 之间，缓冲区可能被其他线程写满；所以生产循环要在每次 onNext 前重新检查，不能一次判断后连发一万条。
onReadyHandler 由传输线程调用，可能和业务线程的 drain 并发进入。AtomicBoolean 保证同一时刻只有一个线程在泵数据，synchronized 保护 done 和 Iterator。两者都需要：前者防止重入，后者防止取消与发送竞争。onNext、onCompleted、onError 这三个方法本身不是线程安全的，必须串行调用。
客户端取消时要停止生产者。setOnCancelHandler(pump::cancel) 只把 done 置真，不再调用 onCompleted；对一个已经取消的流调用 onCompleted 会抛 IllegalStateException。取消后数据源也要关闭，BookRepository 的游标要释放，否则数据库连接会泄漏。
入站方向也有流控。默认情况下 gRPC 自动请求消息，客户端发多快服务端就收多快；对 Chat 这种客户端流，可以调用 disableAutoRequest()，再在处理完一条消息后 request(1)，把压力反传给客户端。入站流控和出站 isReady 是两套独立机制，不能用一个替代另一个。
在 linkd-v3 里落地：三层限额的排序
服务端入口的顺序是：
HTTP/2 连接
-> 令牌桶 200 rps / burst 400 （8.1）
-> 执行器队列 128 （8.3）
-> 并发拦截器 64 （本节）
-> 业务方法
顺序不能颠倒。令牌桶在最前面，决定「要不要接这个请求」；队列在中间，吸收毫秒级抖动；并发拦截器在业务线程入口，决定「现在有没有执行名额」。如果并发拦截器放在令牌桶前面，被限流的请求也会短暂占用一个并发名额，浪费容量；如果队列放在并发拦截器后面，队列里的请求已经算进 64，实际执行并发会超过 64。
拦截器的计数是全局的，不区分方法。GetBook 很快，SearchBooks 很慢，两者共用一个 64 的上限，慢方法会把名额吃光，快方法跟着排队。需要按方法隔离时，用 call.getMethodDescriptor().getFullMethodName() 做 key，给每个方法一个独立的计数器；代价是总量不再有全局上限，要额外维护一个总计数器。linkd-v3 先做全局 64，等压测发现方法间互相影响明显，再拆成「读 48 / 写 16」两个池。
指标至少要有四个：grpc_server_in_flight（当前在途）、grpc_server_rejected_total（拦截器拒绝数）、grpc_server_executor_queue（队列长度）、grpc_server_executor_rejected_total（执行器拒绝数）。前两个来自拦截器，后两个来自 ThreadPoolExecutor.getQueue().size() 和一个自定义 RejectedExecutionHandler。把四个放在一起看，才能区分「并发到了上限」和「队列先满了」。
常见错误与边界
错误一：在 ServerCall.close() 里减计数。 close() 是服务端主动关闭时调用的，客户端取消时不一定走到这里；而且一次调用可能先 close() 再 onComplete()，减两次。计数释放只放在 onComplete / onCancel，并用幂等保护。
错误二：拒绝时返回 null listener。 gRPC 会在后续回调里使用 listener，null 会变成 NullPointerException，客户端看到的是 UNKNOWN 而不是 RESOURCE_EXHAUSTED。返回一个空的 ServerCall.Listener 实现。
错误三：拒绝时忘记 call.close。 只返回空 listener 不关闭 call，客户端会一直等到 deadline，期间还占着连接和流。拒绝路径必须先 close，再返回空 listener。
错误四：用 LongAdder 做在途 gauge。 sum() 不是原子快照，并发下可能读到偏小或偏大的值，准入判断会漂移。AtomicInteger 的 incrementAndGet 是原子的，适合做上限判断；累计指标再用 LongAdder。
错误五：拦截器里做阻塞操作。 拦截器运行在调度线程上，Thread.sleep、同步 HTTP 调用、等锁都会拖住后续调用的分发。拦截器只做计数和拒绝，业务逻辑留给服务方法。
错误六：执行器用无界队列。 new ThreadPoolExecutor(64, 64, 0, MILLISECONDS, new LinkedBlockingQueue<>()) 永远不会触发拒绝，队列会一直涨到 OOM。128 必须是硬上限。
错误七：忽略 isReady() 直接 onNext。 对一个小响应没问题，对上千条的服务端流会把发送缓冲区撑爆。每次 onNext 前检查，onReadyHandler 里恢复。
边界：拦截器管不到执行器队列。 64 是执行中的上限，128 是排队上限，队列里的调用还没进拦截器。队列满时 AbortPolicy 的异常在 gRPC 内部变成什么状态码，取决于版本和调用阶段，不能假设一定是 RESOURCE_EXHAUSTED。要稳定映射，需要在客户端侧也做重试和状态码兜底。
边界：64 是每进程的。 10 个实例就是 640 个并发。下游的总容量要按实例数一起算，不能把单机 64 当成全局承诺。
边界：平台线程不是唯一选择。 JDK 25 可以用虚拟线程，但虚拟线程的并发上限要靠 Semaphore 自己加，不能指望线程池大小。虚拟线程适合阻塞 IO 多的场景；gRPC 的异步服务方法本身不阻塞，平台线程 64 个通常够用。
小结
- 服务端并发限制放在
ServerInterceptor，先incrementAndGet再判断，超限时call.close返回RESOURCE_EXHAUSTED+ErrorInfo。 - 计数释放只放在
onComplete/onCancel，用幂等保护防止重复减；拒绝路径返回空 listener，不返回null。 - 在途 gauge 用
AtomicInteger，累计指标用LongAdder；两者语义不同，不能互换。 - 执行器
core = max = 64、ArrayBlockingQueue(128)、AbortPolicy；队列长度属于执行器，不属于ServerBuilder。 - 拦截器管不到执行器队列，队列满的拒绝可能没有
ErrorInfo；这是方案边界，要用指标监控。 - 流式响应用
isReady+onReadyHandler+setOnReadyThreshold(32 * 1024)回压生产者，onCancelHandler停止数据源，所有onNext串行调用。
并发限制解决「同时执行多少个」，队列解决「排队等多久、排不下怎么办」。8.3 把 ArrayBlockingQueue 的拒绝策略、队列深度和 gRPC 流控接起来，给出完整的背压链路。
阅读导航：上一节 8.1 令牌桶与漏桶 ｜ 下一节 8.3 队列与背压
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。