最小 HTTP 客户端
与 HttpServer 一样，HttpClient 也是不可变配置对象：.headers()、.baseUrl() 等方法返回新实例，原对象不变。要复用连接池和 EventLoop，应在应用启动时创建一个 HttpClient 实例到处共享，不要在每次请求时 HttpClient.create()：后者会走全局默认资源，但反复创建仍会带来额外对象和配置开销。
import reactor.netty.http.client.HttpClient;
import reactor.core.publisher.Mono;
public class MinimalClient {
public static void main(String[] args) {
HttpClient client = HttpClient.create();
String response = client.get()
.uri("https://httpbin.org/get")
.responseContent()
.aggregate()
.asString()
.block(); // 阻塞等待结果（仅演示，生产用响应式链）
System.out.println(response);
}
}
关键点：
HttpClient.create()创建默认客户端.get()/.post()/.put()/.delete()指定 HTTP 方法.uri(String)设置请求地址.responseContent()获取响应体的Flux<ByteBuf>.aggregate().asString()聚合为单个字符串
避免 block()
上面的 .block() 方便演示，但在响应式链中会阻塞线程。正确写法是把客户端调用嵌入 Mono/Flux 管道：
import reactor.netty.http.client.HttpClient;
import reactor.core.publisher.Mono;
public class ReactiveClient {
private final HttpClient client = HttpClient.create();
public Mono<String> fetchUser(String userId) {
return client.get()
.uri("https://api.example.com/users/" + userId)
.responseSingle((res, bytes) -> {
if (res.status().code() == 200) {
return bytes.asString();
}
return Mono.error(new RuntimeException("HTTP " + res.status().code()));
});
}
}
responseSingle() 同时给出 HttpClientResponse（状态码、响应头）和 ByteBufMono（聚合后的响应体），便于在解析体之前先做状态码判断。
POST 请求与请求体
发送 JSON 请求体：
import reactor.core.publisher.Mono;
import reactor.netty.ByteBufMono;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.ByteBuf;
import java.nio.charset.StandardCharsets;
public Mono<String> createUser(String name) {
String json = "{\"name\":\"" + name + "\"}";
ByteBuf body = ByteBufAllocator.DEFAULT
.buffer()
.writeBytes(json.getBytes(StandardCharsets.UTF_8));
return client.post()
.uri("https://api.example.com/users")
.send(Mono.just(body))
.responseSingle((res, bytes) -> bytes.asString());
}
更简洁的方式是用 send(ByteBufMono.fromString(...))：
import io.netty.buffer.ByteBufAllocator;
import reactor.netty.ByteBufMono;
return client.post()
.uri("https://api.example.com/users")
// fromString 没有 (Publisher, Charset) 两参重载：要么用默认字符集，
// 要么把 charset 和 allocator 一起传全
.send(ByteBufMono.fromString(
Mono.just(json), StandardCharsets.UTF_8, ByteBufAllocator.DEFAULT))
.responseSingle((res, bytes) -> bytes.asString());
请求头
请求头有两种设置位置，容易混淆：HttpClient.headers(...) 配在客户端上，对之后从该实例发出的所有请求生效，并返回一个新的 HttpClient；单次请求的头则要在 send 的回调里通过 HttpClientRequest 设置。
// 方式一：配在 HttpClient 上，对所有后续请求生效
// headers() 返回新实例，不会修改原来的 client
HttpClient authed = client.headers(h -> {
h.set("Authorization", "Bearer " + token);
h.set("Accept", "application/json");
});
authed.get()
.uri("https://api.example.com/protected")
.responseSingle((res, bytes) -> bytes.asString());
// 方式二：只给这一次请求加头
// send 的 BiFunction 拿到 HttpClientRequest 后才能改请求头
client.request(HttpMethod.GET) // get() 返回 ResponseReceiver，既没有 headers() 也没有 send()
.send((req, out) -> {
req.header("X-Request-ID", UUID.randomUUID().toString());
return out.send(Mono.empty()); // GET 没有请求体
})
.uri("https://api.example.com/protected")
.responseSingle((res, bytes) -> bytes.asString());
io.netty.handler.codec.http.HttpMethod 是 request(...) 的参数类型。这里不能写成 client.get().uri(...).headers(...)：get() 的返回类型是 ResponseReceiver，链上既没有 headers() 也没有 send(...)，编译期就会失败；要给 GET 单次加头，只能先用 request(HttpMethod.GET) 拿到 RequestSender。post() 和 request(...) 返回的 RequestSender 在 uri() 之后仍是 RequestSender，所以 .uri(...).send(...) 和 .uri(...).send((req, out) -> ...) 都是合法的。
注意不要覆盖 Host 头——Reactor Netty 会自动根据 URI 设置。
响应处理：按状态码分支
public Mono<String> safeFetch(String url) {
return client.get()
.uri(url)
.responseSingle((res, bytes) -> switch (res.status().code()) {
case 200 -> bytes.asString();
case 404 -> Mono.error(new NotFoundException(url));
case 500 -> Mono.error(new ServerErrorException(url));
default -> bytes.asString()
.flatMap(body -> Mono.error(
new RuntimeException("HTTP " + res.status().code() + ": " + body)));
});
}
responseSingle() 的回调签名是 BiFunction<HttpClientResponse, ByteBufMono, Mono<T>>。如果回调内部没有订阅 bytes，响应体会被丢弃。上面的代码在每个分支都消费了 bytes（要么转成字符串，要么直接返回错误）。
查询参数编码
手动拼接查询字符串容易出错（特别是中文和特殊字符）。使用 UriComponentsBuilder：
import org.springframework.web.util.UriComponentsBuilder;
String uri = UriComponentsBuilder
.fromUriString("https://api.example.com/search")
.queryParam("q", "响应式编程")
.queryParam("limit", 10)
.build()
.encode() // 不加 encode() 时 toUriString() 不会转义中文
.toUriString();
// https://api.example.com/search?q=%E5%93%8D%E5%BA%94%E5%BC%8F%E7%BC%96%E7%A8%8B&limit=10
注意：UriComponentsBuilder 属于 Spring Web，如果项目不引入 Spring，可以使用 java.net.URLEncoder：
import java.net.URLEncoder;
String uri = "https://api.example.com/search?q="
+ URLEncoder.encode(query, StandardCharsets.UTF_8)
+ "&limit=" + limit;
压缩与内容协商
Reactor Netty 客户端默认不自动解压 Gzip 响应，但内建了压缩协商开关，打开后会自动加请求头并解压响应：
HttpClient gzipClient = client.compress(true);
compress(true) 会自动发送 Accept-Encoding: gzip（同时支持 Brotli、zstd），并在 pipeline 里安装 HttpContentDecompressor，业务代码拿到的已经是解压后的字节。只有需要自定义 Content-Encoding、或流式处理压缩字节时才手写解压。手动方式如下：
// Accept-Encoding 是请求头，配在 client 上；headers() 返回新实例，原 client 不受影响
HttpClient gzipClient = client.headers(h -> h.set("Accept-Encoding", "gzip"));
gzipClient.get()
.uri("https://api.example.com/data")
.responseSingle((res, bytes) -> {
boolean gzipped = "gzip".equalsIgnoreCase(
res.responseHeaders().get("Content-Encoding"));
if (gzipped) {
// responseContent() 给的是压缩后的原始字节，用 JDK 的 GZIPInputStream 解压
return bytes.asByteArray()
.map(arr -> {
try (var in = new java.util.zip.GZIPInputStream(
new java.io.ByteArrayInputStream(arr))) {
return new String(in.readAllBytes(), StandardCharsets.UTF_8);
} catch (java.io.IOException e) {
throw new java.io.UncheckedIOException(e);
}
});
}
return bytes.asString();
});
Spring WebFlux 的 WebClient（同样基于 Reactor Netty，由 spring-boot-starter-webflux 提供）默认就开启了压缩解码。Reactor Netty 原生客户端需要显式调用 compress(true)；手写 GZIPInputStream 只是无法使用内建开关时的兜底，不是首选。
连接复用与 TCP keep-alive
HttpClient 默认使用全局共享的 ConnectionProvider，对同一目标地址的连接会被复用（HTTP/1.1 持久连接），不需要额外设置。
.keepAlive(boolean) 控制的是 TCP 层的 SO_KEEPALIVE（内核定期发探测包判断对端是否存活），不是 HTTP 的 Connection: keep-alive，两者不要混为一谈：
HttpClient.create()
.keepAlive(true) // 打开 SO_KEEPALIVE，默认值
.headers(h -> h.set("Connection", "keep-alive")); // HTTP 层头，默认已持久连接
想强制每个请求新建连接（常用于隔离或压测），应改连接池而不是 keepAlive(false)：把 maxConnections 设为 1 只能串行复用，要真正不复用可以按请求加 Connection: close，或使用一次性 ConnectionProvider。keepAlive(false) 只关掉 TCP 探测，连接仍然会被池复用。
本地服务端测试
结合第 1.1 节的最小服务器，可以写端到端测试：
import reactor.netty.http.server.HttpServer;
import reactor.netty.http.client.HttpClient;
public class ClientServerTest {
public static void main(String[] args) {
// 启动测试服务器
var server = HttpServer.create().port(0) // 随机端口
.handle((req, res) -> res.sendString(
Mono.just("Hello from " + req.path())))
.bindNow();
int port = server.port();
// 客户端访问
String result = HttpClient.create()
.get()
.uri("http://localhost:" + port + "/test")
.responseSingle((res, bytes) -> bytes.asString())
.block();
System.out.println(result); // Hello from /test
server.disposeNow();
}
}
小结
| 能力 | API | 注意 |
|---|---|---|
| GET 请求 | .get().uri(...).responseSingle(...) | 用 responseSingle 同时拿状态和体 |
| POST 请求体 | .send(Mono.just(byteBuf)) 或 .send(ByteBufMono.fromString()) | 避免 .block() |
| 请求头 | client.headers(...) 全局；send((req, out) -> ...) 中 req.header(...) 单次 | get() 之后没有 headers()，不要覆盖 Host |
| 状态码分支 | responseSingle 回调内 switch | 每个分支必须消费 bytes |
| 查询参数 | UriComponentsBuilder 或 URLEncoder | 中文必须编码 |
| 连接复用 | 默认走全局 ConnectionProvider | keepAlive() 是 TCP 探测，不是 HTTP 复用开关 |
下一节 2.2 连接池 ：ConnectionProvider、连接复用、驱逐策略、弹性缩放。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。