网关定位
eventd v2 网关是整个系统的入口层，承担以下职责：
- 协议升级：HTTP/1.1 请求 → WebSocket 实时连接
- 路由分发：按路径把请求转发到后端 API 服务
- 横切处理：日志、请求 ID、认证预检、错误统一包装
- 背压保护：慢客户端隔离、连接数限制
网关层不做业务判断——用户是否存在、订单是否有效由 API 层处理。网关只负责「把请求送到正确的服务」。
模块结构
eventd-v2/
├── eventd-gateway/
│ ├── src/main/java/com/plume/eventd/gateway/
│ │ ├── GatewayServer.java # 入口：HttpServer 启动
│ │ ├── RouteTable.java # 路由表定义
│ │ ├── HttpProxyHandler.java # HTTP 请求转发
│ │ ├── GatewayWebSocketHandler.java # WebSocket 连接管理
│ │ └── Middleware.java # 中间件链（日志/请求ID/错误）
│ └── pom.xml # 依赖 reactor-netty-http
├── eventd-api/
│ └── WebFlux 接口层（第 4–6 章）
└── eventd-data/
└── R2DBC + 缓存层（第 7–9 章）
路由表设计
路由表把 URL 路径映射到目标服务地址。最简实现：
package com.plume.eventd.gateway;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;
public class RouteTable {
private final Map<String, String> routes = new ConcurrentHashMap<>();
public RouteTable() {
// 内部服务地址
routes.put("/api/users", "http://localhost:8081");
routes.put("/api/orders", "http://localhost:8081");
routes.put("/api/events", "http://localhost:8081");
}
public Optional<String> resolve(String path) {
// 按路径段匹配，避免 /api/usersX 误命中 /api/users；最长前缀优先
return routes.entrySet().stream()
.filter(e -> path.equals(e.getKey()) || path.startsWith(e.getKey() + "/"))
.max(java.util.Comparator.comparingInt(e -> e.getKey().length()))
.map(Map.Entry::getValue);
}
}
生产环境应支持动态路由（如从配置中心刷新），但事件驱动网关的接口稳定，静态表足够。
HTTP 转发 Handler
把外部 HTTP 请求转发到内部 API 服务：
package com.plume.eventd.gateway;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import reactor.netty.http.client.HttpClient;
import reactor.netty.http.server.HttpServerRequest;
import reactor.netty.http.server.HttpServerResponse;
import java.util.Locale;
import java.util.Set;
public class HttpProxyHandler {
// hop-by-hop 头只在单跳有效，不能原样转发；Host 和 Content-Length
// 由 HttpClient/HttpServer 按实际连接和 body 重新计算
private static final Set<String> HOP_BY_HOP = Set.of(
"connection", "keep-alive", "proxy-authenticate", "proxy-authorization",
"te", "trailer", "transfer-encoding", "upgrade", "host", "content-length");
private final RouteTable routes;
private final HttpClient client = HttpClient.create();
public HttpProxyHandler(RouteTable routes) {
this.routes = routes;
}
public Mono<Void> handle(HttpServerRequest req, HttpServerResponse res) {
String path = req.path();
return routes.resolve(path)
.map(target -> proxyTo(target, req, res))
// sendString 返回 NettyOutbound，不是 Mono<Void>，要 then() 收口
.orElseGet(() -> res.status(404)
.sendString(Mono.just("{\"error\":\"Not Found\"}")).then());
}
private Mono<Void> proxyTo(String target, HttpServerRequest req, HttpServerResponse res) {
// 网关生成的请求 ID 在 Reactor Context 里，不修改入站请求头
// （HttpServerRequest.requestHeaders() 是只读视图）
return Mono.deferContextual(ctx -> {
String requestId = ctx.getOrDefault("requestId", "");
String targetUrl = target + req.uri();
return client.request(req.method())
.uri(targetUrl)
// 单次请求头在 send 的 BiFunction 里通过 HttpClientRequest 设置；
// RequestSender/ResponseReceiver 上没有 headers()
.send((proxyReq, out) -> {
req.requestHeaders().forEach(e -> {
if (!HOP_BY_HOP.contains(e.getKey().toLowerCase(Locale.ROOT))) {
proxyReq.addHeader(e.getKey(), e.getValue());
}
});
// 网关生成或透传的 ID 显式写给内部服务
if (!requestId.isEmpty()) {
proxyReq.addHeader("X-Request-ID", requestId);
}
// 请求体跨连接转发，必须 retain：源连接在 onNext 返回后会释放
return out.send(req.receive().retain());
})
.responseConnection((proxyRes, conn) -> {
res.status(proxyRes.status());
proxyRes.responseHeaders().forEach(e -> {
if (!HOP_BY_HOP.contains(e.getKey().toLowerCase(Locale.ROOT))) {
res.header(e.getKey(), e.getValue());
}
});
// 入站 ByteBuf 由客户端连接自动释放，交给下游连接异步写之前必须 retain
return res.send(conn.inbound().receive().retain());
})
// responseConnection 返回 Flux<Void>，而方法签名是 Mono<Void>
.then();
});
}
// WebSocket 网关把单条消息转成一次内部 HTTP 调用，结果写回该连接的 sink
public Mono<Void> proxyDirect(String path, Sinks.Many<String> responseSink) {
return routes.resolve(path)
.map(target -> client.get()
.uri(target + path)
.responseSingle((proxyRes, bytes) -> bytes.asString())
.doOnNext(body -> {
Sinks.EmitResult emitted = responseSink.tryEmitNext(body);
if (emitted.isFailure()) {
// 客户端读得太慢，缓冲区已满：丢弃并记录，不能阻塞内部调用
System.err.println("WS response dropped: " + emitted);
}
})
.then())
.orElseGet(() -> {
responseSink.tryEmitNext("{\"error\":\"Not Found\"}");
return Mono.empty();
});
}
}
responseConnection 是 Reactor Netty 的低级 API，同时给出 HttpClientResponse 和 Connection，可以把响应体字节流直接转交给下游 HttpServerResponse。它是流式透传，没有把整个响应聚合进内存；但每一块 ByteBuf 仍要经过引用计数和编解码，不是零拷贝。它的返回类型是 Flux<V>，不是 Mono，所以收口到 Mono<Void> 的方法时还要 .then()，否则编译不过。
两边都需要 .retain()，原因相同：receive() 发出的 ByteBuf 在 onNext 返回后会被源连接自动释放，而 send() 是异步写，写完才由目标连接释放。同一次转发有两个所有者，不 retain() 就会出现「源已释放、目标还在写」的二次释放，实测表现为 IllegalReferenceCountException: refCnt: 0, decrement: 1 和连接被关闭。
转发时必须过滤 hop-by-hop 头：Connection、Transfer-Encoding、Upgrade 等只对单跳有意义，原样转发会让上游和下游对消息边界产生分歧；Host 和 Content-Length 也不能照抄，前者应按目标地址生成，后者要与实际发送的 body 一致。
中间件链
与第 1.2 节的中间件类似，但增加了请求 ID 透传：
package com.plume.eventd.gateway;
import reactor.core.publisher.Mono;
import reactor.netty.http.server.HttpServerRequest;
import reactor.netty.http.server.HttpServerResponse;
import java.util.function.BiFunction;
@FunctionalInterface
interface Handler extends BiFunction<HttpServerRequest, HttpServerResponse, Mono<Void>> {}
public class Middleware {
static Handler withRequestId(Handler next) {
return (req, res) -> {
String incoming = req.requestHeaders().get("X-Request-ID");
String requestId = incoming == null
? java.util.UUID.randomUUID().toString()
: incoming;
res.header("X-Request-ID", requestId);
// 生成的 ID 放进 Reactor Context，proxyTo 与日志从同一来源读取
return next.apply(req, res)
.contextWrite(ctx -> ctx.put("requestId", requestId));
};
}
static Handler withLogging(Handler next) {
return (req, res) -> Mono.deferContextual(ctx -> {
long start = System.nanoTime();
String reqId = ctx.getOrDefault("requestId", "-");
System.out.println("[" + reqId + "] → " + req.method().name() + " " + req.path());
return next.apply(req, res)
.doFinally(sig -> {
long ms = (System.nanoTime() - start) / 1_000_000;
System.out.println("[" + reqId + "] ← " + req.path() + " in " + ms + "ms");
});
});
}
static Handler withErrorHandler(Handler next) {
return (req, res) ->
next.apply(req, res)
.onErrorResume(err -> {
System.err.println("Gateway error: " + err.getMessage());
return res.status(502)
.header("Content-Type", "application/json")
.sendString(Mono.just(
"{\"error\":\"GatewayError\",\"message\":\"" +
err.getClass().getSimpleName() + "\"}"));
});
}
}
请求 ID 由网关生成或透传后放进 Reactor Context：响应头、访问日志和转发给内部服务的请求头都从这一份值读取。只在响应头里写一次是不够的——入站请求头是只读的，不显式加给 HttpClientRequest，内部服务就收不到 ID。这是分布式追踪（第 10 章）的基础。
WebSocket 协议转换
网关层的 WebSocket 不是简单回显，而是把 WebSocket 消息转换为内部 HTTP 调用：
package com.plume.eventd.gateway;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import reactor.netty.http.websocket.WebsocketInbound;
import reactor.netty.http.websocket.WebsocketOutbound;
import java.time.Duration;
import java.util.concurrent.ArrayBlockingQueue;
public class GatewayWebSocketHandler {
private final HttpProxyHandler proxy;
public GatewayWebSocketHandler(HttpProxyHandler proxy) {
this.proxy = proxy;
}
public Mono<Void> handle(WebsocketInbound in, WebsocketOutbound out) {
// 每个 WebSocket 连接的响应 sink；显式有界队列，客户端读得慢时不会无限积压
Sinks.Many<String> responseSink = Sinks.many().unicast()
.onBackpressureBuffer(new ArrayBlockingQueue<>(64));
// 接收客户端消息，转发为 HTTP 请求
Mono<Void> receive = in.receive()
.asString()
.flatMap(msg -> {
// 解析消息格式: {"path":"/api/events","method":"GET","body":null}
String path = extractPath(msg);
// 简化为直接构造 HTTP 调用；单条消息失败只丢弃这一条，不终止整个连接
return proxy.proxyDirect(path, responseSink)
.onErrorResume(err -> {
System.err.println("WS proxy error: " + err.getMessage());
return Mono.empty();
});
})
.then();
// 把 HTTP 响应发回 WebSocket
Mono<Void> send = out.sendString(responseSink.asFlux()).then();
// 任一端终止（客户端断开、上游出错、responseSink 完成）都会结束 handler；
// 不要用 Mono.when 等两端都完成，对端断开后继续写已关闭的通道没有意义
return Mono.zip(receive, send).then();
}
String extractPath(String msg) {
// 简易 JSON 解析；格式不对时抛出明确异常，由 flatMap 内的 onErrorResume 吞掉这一条
int start = msg.indexOf("\"path\":\"");
if (start < 0) {
throw new IllegalArgumentException("missing path: " + msg);
}
start += 8;
int end = msg.indexOf('"', start);
if (end < 0) {
throw new IllegalArgumentException("unterminated path: " + msg);
}
return msg.substring(start, end);
}
}
这里刻意不用 onErrorContinue：它依赖各操作符内部支持，遇到不支持的算子会被静默忽略，而且无法区分是哪条消息出错。把错误处理放进 flatMap 内部的 onErrorResume，语义是「这一条失败就丢弃这一条」；如果需要通知客户端，可以往 responseSink 里补一条错误响应再返回 Mono.empty()。
这个设计的意义：客户端通过单一 WebSocket 连接发送多种请求（订阅行情、创建订单、查询状态），网关按 path 字段路由到不同内部服务，响应通过同一连接异步返回。
网关整合启动
package com.plume.eventd.gateway;
import reactor.netty.http.server.HttpServer;
public class GatewayServer {
public static void main(String[] args) {
RouteTable routes = new RouteTable();
HttpProxyHandler proxy = new HttpProxyHandler(routes);
// withRequestId 必须包在 withLogging 外层：
// contextWrite 只向内层传播，反过来的话日志读不到 requestId
Middleware.Handler app = Middleware.withErrorHandler(
Middleware.withRequestId(
Middleware.withLogging(
(req, res) -> {
String path = req.path();
if (path.startsWith("/ws")) {
// WebSocket 升级由 route().ws() 处理
return res.sendString(
reactor.core.publisher.Mono.just("WS handled"));
}
return proxy.handle(req, res);
}
)
)
);
HttpServer.create()
.port(8080)
.route(r -> r
.get("/health", (req, res) ->
res.sendString(reactor.core.publisher.Mono.just("ok")))
.ws("/ws/events", new GatewayWebSocketHandler(proxy)::handle)
// route() 没有单参数的 handler 重载，兜底路由也要给 predicate
.route(req -> true, (req, res) -> app.apply(req, res))
)
.bindNow();
System.out.println("Gateway on http://localhost:8080");
System.out.println("WebSocket: ws://localhost:8080/ws/events");
try { System.in.read(); } catch (Exception ignored) {}
}
}
背压保护
网关层有三道背压防线：
1. TCP 层：滑动窗口 + Netty 写水位线（第 1.3 节）
↓
2. 连接层：最大连接数限制（sessions.size() < 10000）
↓
3. 应用层：tryEmitNext + 慢客户端断开（第 3.2 节）
还有一道隐式保护：HTTP 转发使用 responseConnection 流式透传，没有把响应聚合后再发。即使下游返回 1GB 响应，网关也只需要持有当前在途的几个 ByteBuf，不会整体驻留内存。
小结
| 组件 | 职责 | 技术点 |
|---|---|---|
RouteTable | 路径 → 目标服务映射 | ConcurrentHashMap，前缀匹配 |
HttpProxyHandler | HTTP 请求转发 | HttpClient.responseConnection 透传 |
Middleware | 日志、请求 ID、错误处理 | 函数包装，顺序：Error → Log → RequestId |
GatewayWebSocketHandler | WebSocket ↔ HTTP 转换 | Sinks.Many 异步响应、消息解析路由 |
GatewayServer | 启动与路由整合 | HttpServer.route() 混合 HTTP + WS |
| 背压 | 四层保护 | TCP 窗口 + 连接限制 + tryEmitNext + 流式透传 |
第一部分（第 1–3 章）到此结束。你已经可以：
- 搭建最小 HTTP 服务器和客户端
- 配置连接池和弹性重试
- 处理 WebSocket 实时通信
- 构建带路由表和中间件的网关层
下一部分进入 WebFlux：4.1 注解路由 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。