本节目标：把
/books/{id}的代理从「虚拟线程里发阻塞请求」改成「同一条 EventLoop 上的异步转发」，并让路由、头过滤、超时和错误码都有明确归属。
适用版本：JDK 25 + Netty 4.2.19.Final。
承接的问题
3.3 给出的 GatewayHandler 能工作：它把 request.uri() 拼成 http://127.0.0.1:8080 的地址，交给虚拟线程里的 java.net.http.HttpClient.send，拿到响应再 ctx.writeAndFlush 回客户端。它解决的是「不要在 EventLoop 上阻塞」，代价是每个请求跨一次线程边界，还要一个 Semaphore 把并发压回下游能承受的范围。
到了第 15 章，网关要回答的问题更具体：哪些路径该转发，前缀冲突时谁优先；/books/7?fields=title 到上游应该变成什么；哪些请求头不能原样带过去；上游不可达、建连超时、响应超时分别返回 502 还是 504，连接还能不能回池；以及为什么 Netty 已经提供了异步客户端，还要在 EventLoop 里再套一个阻塞客户端。
这一节把 3.3 的临时实现换成 linkd-v2 的正式版本：GatewayRoutes 管匹配与重写，GatewayHandler 管头过滤与转发，连接复用交给 8.2 的 HttpBackendPools，客户端 pipeline 由 Bootstrap、HttpClientCodec、HttpObjectAggregator 组成。
路由表：前缀匹配与路径重写
路由表不做正则，也不做注解扫描。linkd-v2 只有一条代理路由，用「前缀 + 上游基地址 + 预算」三字段足够，以后加 /covers/ 只需加一行。关键是匹配和重写放在同一个对象里，避免「匹配到 A、重写成 B」这种两处维护的错位。
package com.plume.linkd.v2.server;
import java.net.URI;
import java.time.Duration;
import java.util.List;
import java.util.Locale;
import java.util.Optional;
public final class GatewayRoutes {
public record Route(String prefix, String upstreamBase, Duration budget) {
public URI rewrite(String rawUri) {
int q = rawUri.indexOf('?');
String path = q < 0 ? rawUri : rawUri.substring(0, q);
String query = q < 0 ? "" : rawUri.substring(q);
if (!path.startsWith(prefix) || path.length() == prefix.length()) {
throw new IllegalArgumentException("path does not match prefix");
}
String lower = path.toLowerCase(Locale.ROOT);
if (path.contains("..") || path.contains("//") || lower.contains("%2e")) {
throw new IllegalArgumentException("dot segment or empty segment");
}
return URI.create(upstreamBase + path.substring(prefix.length()) + query);
}
}
private static final List<Route> ROUTES = List.of(
new Route("/books/", "http://127.0.0.1:8080/books/", Duration.ofMillis(2000)));
public static Optional<Route> match(String rawUri) {
int q = rawUri.indexOf('?');
String path = q < 0 ? rawUri : rawUri.substring(0, q);
return ROUTES.stream()
.filter(r -> path.startsWith(r.prefix()) && path.length() > r.prefix().length())
.findFirst();
}
private GatewayRoutes() {
}
}
第一，匹配用原始路径，重写保留原始查询串。request.uri() 可能带 %2F、重复参数、空值参数。用 QueryStringDecoder 解析再拼回去，参数顺序和编码都可能变；网关不应该替客户端重排查询串。这里只按第一个 ? 切开，路径做前缀替换，查询串原样拼回。
第二，前缀必须带结尾斜杠。/books/ 能匹配 /books/7，不会匹配 /bookshelf；如果写成 /books，/bookshelf 也会被误转发。匹配后再要求 path.length() > prefix.length()，GET /books/ 本身就不会进代理，而是走 404。
第三，点段检查放在网关，不指望上游。/books/../../etc/passwd 在到达 libraryd 之前就该拒绝；%2e%2e 是同一件事的编码形式，按小写后的原始路径检查。完整 RFC 3986 归一化更复杂，生产网关应该用经过审计的 URI 解析库；这里的规则只覆盖 linkd-v2 的一条路由。
Hop-by-hop 头过滤
HTTP/1.1 的头部有两类：端到端头和逐跳头。逐跳头只对当前这一跳有意义，代理必须消费掉，不能透传。Connection 头还携带一个「名单」，名单里列出的头同样是逐跳的，必须一起删。
private static final Set<String> HOP_BY_HOP = Set.of(
"connection", "keep-alive", "proxy-authenticate", "proxy-authorization",
"te", "trailer", "transfer-encoding", "upgrade", "expect");
expect 不在 RFC 的逐跳名单里，但这里也要删：HttpObjectAggregator 已经把请求体读完，上游再看到 Expect: 100-continue 会等一个永远不会来的 100 Continue。这是「按链路实际情况删」，不是照抄名单。
复制请求头时还有三件事：Host 必须改成上游的 authority，否则上游虚拟主机路由会命中错误站点；Content-Length 删掉后按聚合后的 content().readableBytes() 重算，入站请求可能原本是 chunked；X-Forwarded-For 写入客户端地址，X-Forwarded-Proto 写 http。只有可信代理才能决定是否相信客户端自带的 X-Forwarded-For，这一条在 15.2 的限流里还会遇到。
在 linkd-v2 里落地：异步转发的 GatewayHandler
客户端 pipeline 沿用 8.2 的 HttpBackendPools，每个上游地址一个 FixedChannelPool。池里的 Bootstrap 只建一次，配置如下：
Bootstrap bootstrap = new Bootstrap()
.group(group)
.channel(NioSocketChannel.class)
.remoteAddress(key)
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 1000)
.option(ChannelOption.TCP_NODELAY, true)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) {
ch.pipeline().addLast(new HttpClientCodec());
ch.pipeline().addLast(new HttpObjectAggregator(1 << 20));
ch.pipeline().addLast(new ReadTimeoutHandler(1500, TimeUnit.MILLISECONDS));
ch.pipeline().addLast(new BackendResponseHandler());
}
});
HttpClientCodec 同时负责请求编码和响应解码；HttpObjectAggregator 把 HttpResponse 与 HttpContent 聚合成 FullHttpResponse，上限 1 MiB。BackendResponseHandler 是 8.2 里的类，它把响应 retain() 后交给 CompletableFuture，并清空 channel 属性；GatewayHandler 用同名 key linkd.pool.response 重新 valueOf，Netty 的 AttributeKey 按名字复用实例，两边拿到同一个 key，更干净的做法是把常量提到公共类。
GatewayHandler 只保留路由和协议语义，不直接碰 Bootstrap；3.3 的构造参数 ExecutorService 在这里换成 HttpBackendPools，LinkdV2Main 的装配同步替换，虚拟线程执行器只留给 CPU 密集任务：
package com.plume.linkd.v2.server;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.pool.FixedChannelPool;
import io.netty.handler.codec.http.DefaultFullHttpRequest;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpUtil;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.util.AttributeKey;
import io.netty.util.CharsetUtil;
import io.netty.util.ReferenceCountUtil;
import com.plume.linkd.v2.client.HttpBackendPools;
import java.net.InetSocketAddress;
import java.net.URI;
import java.util.HashSet;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
@ChannelHandler.Sharable
public final class GatewayHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
private static final Set<String> HOP_BY_HOP = Set.of(
"connection", "keep-alive", "proxy-authenticate", "proxy-authorization",
"te", "trailer", "transfer-encoding", "upgrade", "expect");
public static final AttributeKey<CompletableFuture<FullHttpResponse>> RESPONSE =
AttributeKey.valueOf("linkd.pool.response");
private final HttpBackendPools pools;
public GatewayHandler(HttpBackendPools pools) {
this.pools = pools;
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) {
if (!request.decoderResult().isSuccess()) {
write(ctx, HttpResponseStatus.BAD_REQUEST, "bad request");
return;
}
GatewayRoutes.match(request.uri()).ifPresentOrElse(
route -> forward(ctx, request, route),
() -> write(ctx, HttpResponseStatus.NOT_FOUND, "no route"));
}
private void forward(ChannelHandlerContext ctx, FullHttpRequest request,
GatewayRoutes.Route route) {
URI target;
try {
target = route.rewrite(request.uri());
} catch (IllegalArgumentException e) {
write(ctx, HttpResponseStatus.BAD_REQUEST, "bad path");
return;
}
FullHttpRequest outbound = new DefaultFullHttpRequest(
HttpVersion.HTTP_1_1, request.method(),
target.getRawPath() + (target.getRawQuery() == null ? "" : "?" + target.getRawQuery()),
request.content().retain());
copyRequestHeaders(ctx, request, outbound, target);
HttpUtil.setKeepAlive(outbound, true);
FixedChannelPool pool = pools.get(
InetSocketAddress.createUnresolved(target.getHost(), target.getPort()));
CompletableFuture<FullHttpResponse> result = new CompletableFuture<>();
AtomicReference<Channel> leased = new AtomicReference<>();
pool.acquire().addListener((io.netty.util.concurrent.Future<Channel> acquired) -> {
if (!acquired.isSuccess()) {
ReferenceCountUtil.release(outbound);
result.completeExceptionally(new PoolBusyException(acquired.cause()));
return;
}
Channel backend = acquired.getNow();
leased.set(backend);
backend.attr(RESPONSE).set(result);
backend.writeAndFlush(outbound).addListener((ChannelFutureListener) write -> {
if (!write.isSuccess()) {
result.completeExceptionally(write.cause());
}
});
});
result.orTimeout(route.budget().toMillis(), TimeUnit.MILLISECONDS)
.whenComplete((response, error) -> {
Channel backend = leased.get();
if (backend != null) {
backend.attr(RESPONSE).set(null);
}
if (error != null) {
if (backend != null) {
backend.close();
}
if (error instanceof PoolBusyException) {
write(ctx, HttpResponseStatus.SERVICE_UNAVAILABLE, "backend pool busy");
} else if (error instanceof TimeoutException || error instanceof io.netty.handler.timeout.TimeoutException) {
write(ctx, HttpResponseStatus.GATEWAY_TIMEOUT, "upstream timeout");
} else {
write(ctx, HttpResponseStatus.BAD_GATEWAY, "proxy failed");
}
return;
}
boolean reusable = backend != null && backend.isActive()
&& HttpUtil.isKeepAlive(response);
try {
writeResponse(ctx, request, response);
} finally {
response.release();
if (reusable) {
pool.release(backend);
} else if (backend != null) {
backend.close();
}
}
});
}
private static void copyRequestHeaders(ChannelHandlerContext ctx, FullHttpRequest in,
FullHttpRequest out, URI target) {
Set<String> named = new HashSet<>();
for (String value : in.headers().getAll(HttpHeaderNames.CONNECTION)) {
for (String token : value.split(",")) {
if (!token.isBlank()) {
named.add(token.trim().toLowerCase(Locale.ROOT));
}
}
}
for (Map.Entry<String, String> header : in.headers()) {
String lower = header.getKey().toLowerCase(Locale.ROOT);
if (HOP_BY_HOP.contains(lower) || named.contains(lower)
|| lower.equals("host") || lower.equals("content-length")) {
continue;
}
out.headers().add(header.getKey(), header.getValue());
}
String clientIp = ctx.channel().remoteAddress() instanceof InetSocketAddress remote
? remote.getAddress().getHostAddress() : "unknown";
out.headers().set(HttpHeaderNames.HOST, target.getAuthority());
out.headers().set("X-Forwarded-Proto", "http");
out.headers().set("X-Forwarded-For", clientIp);
HttpUtil.setContentLength(out, out.content().readableBytes());
}
private static void writeResponse(ChannelHandlerContext ctx, FullHttpRequest request,
FullHttpResponse backend) {
FullHttpResponse response = new DefaultFullHttpResponse(
HttpVersion.HTTP_1_1, backend.status(), backend.content().retain());
for (Map.Entry<String, String> header : backend.headers()) {
String lower = header.getKey().toLowerCase(Locale.ROOT);
if (HOP_BY_HOP.contains(lower) || lower.equals("content-length")) {
continue;
}
response.headers().add(header.getKey(), header.getValue());
}
response.headers().set("Via", "1.1 linkd-v2");
HttpUtil.setContentLength(response, response.content().readableBytes());
boolean keepAlive = HttpUtil.isKeepAlive(request);
if (keepAlive) {
response.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderNames.KEEP_ALIVE);
}
ChannelFuture write = ctx.writeAndFlush(response);
if (!keepAlive) {
write.addListener(ChannelFutureListener.CLOSE);
}
}
private static void write(ChannelHandlerContext ctx, HttpResponseStatus status, String body) {
if (!ctx.channel().isActive()) {
return;
}
FullHttpResponse response = new DefaultFullHttpResponse(
HttpVersion.HTTP_1_1, status,
Unpooled.copiedBuffer(body + "\n", CharsetUtil.UTF_8));
response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/plain; charset=utf-8");
HttpUtil.setContentLength(response, response.content().readableBytes());
ctx.writeAndFlush(response);
}
private static final class PoolBusyException extends RuntimeException {
PoolBusyException(Throwable cause) {
super(cause);
}
}
}
流量经过这条路径时，所有权变化如下：
- 入站
request由SimpleChannelInboundHandler在channelRead0返回后释放，所以出站请求必须request.content().retain()。 - 池借出的
Channel在正常响应后release回池；超时、写失败、上游返回Connection: close时close，不回池。 - 8.2 的
BackendResponseHandler已经对响应retain()过一次，writeResponse再retain()一次构造新响应，最后finally里释放原始响应，引用计数回到 1；池获取失败时outbound还没写出去，必须显式ReferenceCountUtil.release(outbound)。
超时、502 与连接复用
三个时间点要分开设，而且顺序是「外层大于内层」：
| 层 | 配置 | 触发时 | 返回 |
|---|---|---|---|
| 建连 | CONNECT_TIMEOUT_MILLIS = 1000 | 上游端口不可达或握手无响应 | 502 |
| 读 | ReadTimeoutHandler(1500 ms) | 已建连但 1.5 秒没有读到任何响应 | 504，关闭连接 |
| 总预算 | route.budget() = 2000 ms | 从借连接到响应完成超过 2 秒 | 504，关闭连接 |
读超时和总预算不是重复：ReadTimeoutHandler 在每个读事件后重置，一个每秒吐 1 字节的上游永远不会触发它；总预算从请求开始计时，到了就结束。反过来，只有总预算没有读超时，上游卡住时连接会一直被占着，池很快耗尽。
超时后必须 close() 而不是 release()。8.2 已经指出过原因：orTimeout 只让 future 完成，不会取消已经在路上的写和读；把连接放回池，下一个借用者会先读到上一个请求的迟到响应，表现为「返回了别人的数据」。关闭连接让内核回收，代价是一次 TCP 握手，比串响应便宜。
HttpUtil.isKeepAlive(response) 决定能否回池。上游明确返回 Connection: close，或者响应没有 Content-Length 也没有 chunked（聚合器能处理 chunked，但连接语义仍要看头），都不应该复用。FixedChannelPool 的 ChannelHealthChecker.ACTIVE 只检查 isActive()，它不知道「这条连接上还有一个迟到的响应」，所以业务层必须自己把关。
502 和 504 的边界也要说清楚：502 是根本没拿到合法响应，连接被拒、DNS 失败、上游返回非法 HTTP、写失败；504 是拿到了连接，但在预算内没有完成响应；503 是网关自己没能力处理，比如池等待队列满、熔断打开（15.2）。不要把三者都写成 502，否则客户端和监控无法区分「上游挂了」和「上游慢」。
为什么不在 EventLoop 里用阻塞 HTTP 客户端
EventLoop 是单线程执行器，一个 Channel 注册后终身绑定同一条 EventLoop。默认线程数是 2 × 可用核数，本机 10 核就是 20 条；每条线程上通常挂着几十到上千个连接。在 channelRead0 里调用 HttpClient.send，这条线程会停在 socket 读上，它上面的所有连接一起停摆：/healthz 超时让编排系统以为进程死了，WebSocket 心跳发不出去，其他 /books/{id} 连 channelRead 都排不上。这不是「Netty 慢」，是「把阻塞调用放在了不允许阻塞的线程上」。ChannelFuture.sync() 在 EventLoop 里还可能直接死锁：等待的 future 需要同一条 EventLoop 推进，而这条线程正卡在 sync() 上。
3.3 的虚拟线程方案是对这个问题的正确回答之一：阻塞时载体线程可以卸载，EventLoop 继续跑。但它仍有成本：每个请求一次线程切换和一次 Semaphore 配额管理；虚拟线程里的异常、上下文、超时都要单独传递，HttpClient 的连接池和 Netty 的连接池是两套；下游容量仍然要靠信号量或限流压住，虚拟线程不增加下游连接数。
Netty 客户端把这件事放回同一条 EventLoop：Bootstrap.connect() 返回 future，writeAndFlush 返回 future，响应到达时在同一个线程回调。没有线程切换，连接池就是 8.2 的 FixedChannelPool，超时就是 orTimeout 加 ReadTimeoutHandler。代价是代码必须全程异步，任何一步忘了 addListener 都会变成「请求发出去了，没人处理响应」。
常见错误与边界
错误一：把 Transfer-Encoding 原样转发。 入站请求是 chunked，HttpObjectAggregator 已经把它聚合成完整 body；出站再带 Transfer-Encoding: chunked，HttpClientCodec 会再分块一次，上游看到的是「chunked 里套 chunked」，或者同时出现 Content-Length 与 Transfer-Encoding 而被拒绝。逐跳头必须删，长度重新算。
错误二：超时后把连接放回池。 反例是 whenComplete 里不区分正常与异常，统一 pool.release(channel)。迟到的响应会留在连接上，下一个请求读到它，日志里出现「A 用户拿到 B 用户的书」。正确做法是异常路径 close()，正常路径再检查 isActive() 与 isKeepAlive()。
小结
- 路由表把「前缀、上游基地址、预算」放在同一个
Route里；查询串原样保留，只替换路径前缀，点段和空段在网关拒绝。 - 逐跳头按 RFC 名单加
Connection展开名单删除，Expect也要删；Host、Content-Length、X-Forwarded-*重新生成。 - 建连超时、读超时、总预算分层设置；超时和
Connection: close关闭连接，只有健康且 keep-alive 的连接才回池。 - 502 是没拿到合法响应，504 是预算内没完成，503 是网关自身没能力；三者不能混用。
阅读导航：上一节：14.3 调优案例与回归 · 下一节：15.2 限流、超时与降级 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。