URI 路由分发
最小服务器只有一个 handle() 回调，真实服务需要按路径分发。最朴素的实现是用 req.path() 做分支：
import reactor.core.publisher.Mono;
import reactor.netty.http.server.HttpServer;
import reactor.netty.http.server.HttpServerRequest;
import reactor.netty.http.server.HttpServerResponse;
public class RoutedServer {
public static void main(String[] args) {
HttpServer.create().port(8080)
.handle(RoutedServer::route)
.bindNow();
System.out.println("http://localhost:8080");
try { System.in.read(); } catch (Exception ignored) {}
}
static Mono<Void> route(HttpServerRequest req, HttpServerResponse res) {
return switch (req.path()) {
case "/health" -> health(req, res);
case "/users" -> users(req, res);
default -> res.status(404).send();
};
}
static Mono<Void> health(HttpServerRequest req, HttpServerResponse res) {
return res.status(200)
.header("Content-Type", "application/json")
.sendString(Mono.just("{\"status\":\"up\"}"))
.then(); // sendString 返回 NettyOutbound，不是 Mono
}
static Mono<Void> users(HttpServerRequest req, HttpServerResponse res) {
return res.status(200)
.header("Content-Type", "application/json")
.sendString(Mono.just("[\"alice\",\"bob\"]"))
.then();
}
}
这种 switch 写法适合演示和小服务。第 3 章的 eventd v2 网关会用更正式的中间件链替代它。
路径参数提取
REST API 需要路径参数，例如 /users/123。手动解析：
static Mono<Void> userById(HttpServerRequest req, HttpServerResponse res) {
String path = req.path(); // /users/123
String id = path.substring(path.lastIndexOf('/') + 1);
return res.status(200)
.sendString(Mono.just("User: " + id));
}
更健壮的方案是使用正则或预编译模式匹配。网关层（第 3 章）会封装通用的路由表，避免每个 handler 自己解析。
请求体读取与 JSON 解码
请求体是异步字节流，需要用 receive() 读取并聚合：
import reactor.core.publisher.Mono;
import java.nio.charset.StandardCharsets;
static Mono<Void> createUser(HttpServerRequest req, HttpServerResponse res) {
// 1. 先把字节流聚合成一个 ByteBuf，再按 UTF-8 解码
// 不要直接对 ByteBufFlux 调 asString()：它按每个 ByteBuf 分片各自解码，
// 一个中文字符如果跨 TCP 分片就会被拆坏
Mono<String> bodyMono = req.receive()
.aggregate() // ByteBufFlux → ByteBufMono
.asString(StandardCharsets.UTF_8); // 整体解码
// 2. 解析 JSON（这里用手动解析演示；生产用 Jackson）
return bodyMono.flatMap(json -> {
String name = extractName(json); // 简易解析：找 "name":"..."
System.out.println("Creating user: " + name);
return res.status(201)
.header("Content-Type", "application/json")
.sendString(Mono.just("{\"id\":1,\"name\":\"" + name + "\"}"))
.then(); // flatMap 要求返回 Mono，NettyOutbound 不是 Mono
});
}
static String extractName(String json) {
int start = json.indexOf("\"name\":\"");
if (start < 0) {
throw new IllegalArgumentException("missing name field");
}
start += 8;
int end = json.indexOf('"', start);
if (end < 0) {
throw new IllegalArgumentException("unterminated name field");
}
return json.substring(start, end);
}
req.receive() 返回 ByteBufFlux（即分片的 Flux<ByteBuf>）。关键操作链：
.aggregate()—— 把多个 ByteBuf 合并成一个ByteBufMono。它没有maxInMemorySize之类的总量保护（Reactor Netty 的 HTTP 解码器只有单行、单头、单 chunk 上限），整个 body 会被读进内存，必须自己在应用层限制请求体大小.asString(Charset)—— 对合并后的完整字节做一次解码；直接对ByteBufFlux调asString()是逐分片解码，跨分片的多字节字符会被破坏.flatMap(process)—— 业务处理，返回新的Mono<Void>
请求体不适合整体读入内存时（大文件上传），应改为流式处理 Flux<DataBuffer> / Flux<ByteBuf>，而不是 aggregate()。
Content-Type 与状态码的统一处理
每个 handler 都写 .header("Content-Type", ...) 很重复。用一个包装函数：
static Mono<Void> jsonResponse(HttpServerResponse res, int status, String body) {
return res.status(status)
.header("Content-Type", "application/json; charset=utf-8")
.sendString(Mono.just(body))
.then();
}
static Mono<Void> health(HttpServerRequest req, HttpServerResponse res) {
return jsonResponse(res, 200, "{\"status\":\"up\"}");
}
这里的 charset=utf-8 其实可以省略：application/json 按 RFC 8259 固定使用 UTF-8，客户端不应依赖 charset 参数。真正需要显式声明的是 text/plain、text/html 这类 text/* 响应——没有 charset 时浏览器会用页面或系统的默认编码解析，中文才可能变成乱码。统一写成 application/json 即可，把 charset 留给纯文本响应。
中间件链式组合
真实服务需要在 handler 之前执行日志、认证、跨域等横切逻辑。用函数组合实现中间件：
import java.util.function.BiFunction;
@FunctionalInterface
interface Handler extends BiFunction<HttpServerRequest, HttpServerResponse, Mono<Void>> {}
static Handler withLogging(Handler next) {
return (req, res) -> {
long start = System.nanoTime();
System.out.println("→ " + req.method().name() + " " + req.path());
return next.apply(req, res)
.doFinally(sig -> {
long ms = (System.nanoTime() - start) / 1_000_000;
System.out.println("← " + req.path() + " in " + ms + "ms [" + sig + "]");
});
};
}
static Handler withCors(Handler next) {
return (req, res) -> {
res.header("Access-Control-Allow-Origin", "*");
if ("OPTIONS".equals(req.method().name())) {
res.header("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE");
return res.status(204).send();
}
return next.apply(req, res);
};
}
// 使用
Handler app = withCors(withLogging(RoutedServer::route));
HttpServer.create().port(8080)
.handle(app::apply)
.bindNow();
中间件顺序很重要：CORS 必须排在业务路由之前，OPTIONS 预检请求才不会落到业务 handler 里变成 404。最终按 withErrorHandler(withCors(withLogging(...))) 组合时，错误处理在最外层，CORS 次之，日志在内层。
错误处理
handler 中的异常会传播到 Publisher，Reactor Netty 默认返回 500 空响应。添加统一错误处理：
static Handler withErrorHandler(Handler next) {
return (req, res) ->
next.apply(req, res)
.onErrorResume(err -> {
System.err.println("Error handling " + req.path() + ": " + err.getMessage());
return jsonResponse(res, 500,
"{\"error\":\"" + err.getClass().getSimpleName() + "\"}");
});
}
组合：withErrorHandler(withCors(withLogging(...)))，错误处理在最外层，能捕获所有下游异常。
这个包装只在响应尚未提交时有效。如果 handler 已经开始 sendString 并写出一部分数据，再遇到错误，jsonResponse(res, 500, ...) 无法改回状态码，Reactor Netty 会直接中断连接。需要「写出过程中出错也能给出结构化响应」的接口，要么先聚合成完整字节再一次性发送，要么改用 WebFlux 的 ResponseEntity/ServerResponse，由框架判断响应是否已提交（见第 4 章）。
阻塞操作的隔离
如果你的业务逻辑必须调用阻塞 API（如 JDBC），绝对不能在 EventLoop 线程执行。使用 subscribeOn 切换调度器：
static Mono<Void> slowHandler(HttpServerRequest req, HttpServerResponse res) {
return Mono.fromCallable(() -> {
// 模拟阻塞操作（如 JDBC 查询）
Thread.sleep(100);
return "Result";
})
.subscribeOn(reactor.core.scheduler.Schedulers.boundedElastic())
.flatMap(result -> res.sendString(Mono.just(result)).then());
}
Schedulers.boundedElastic() 创建有上限的弹性线程池，适合包裹阻塞 I/O。不要用 parallel()：它是按 CPU 核数固定的线程池，为 CPU 密集计算设计，线程数少且不弹性；多个阻塞调用会把它占满，反而拖住所有使用 parallel() 的算子。
小结
| 能力 | API | 注意点 |
|---|---|---|
| 路由分发 | req.path() + switch 或模式匹配 | 网关层会封装更正式的路由表 |
| 请求体读取 | req.receive().aggregate().asString() | 先聚合再解码，避免多字节字符跨分片被拆坏 |
| 响应编码 | .header("Content-Type", ...) + .sendString() | JSON 不用带 charset，text/* 才需要 |
| 中间件 | 函数包装 Handler | 顺序：错误处理 → CORS → 日志 → 业务 |
| 阻塞隔离 | subscribeOn(boundedElastic()) | 绝不在 EventLoop 调用阻塞 API |
下一节 1.3 HTTP 层的背压
：TCP 发送窗口、接收缓冲、Reactor Netty 的 writeWith() 背压衔接、慢客户端保护。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。