java

《Netty 高性能网络编程》8.2 连接池与多路复用

连接池的关键不是池化本身，而是何时借、何时还、何时判定连接已坏。本节拆开 FixedChannelPool 的三个生命周期回调，配合 ChannelHealthChecker 与获取超时，用 AbstractChannelPoolMap 按 host:port 建池，并用 Little 定律说明池大小以及 HTTP/1.1 与 HTTP/2 的取舍。

Leeting Yan 9 分钟阅读 4186 字

本节目标：知道什么时候该池化、池的每个回调该做什么、池多大合适，以及为什么 linkd-v2 的两种后端需要两套并发模型。
适用版本：JDK 25 + Netty 4.2.19.Final。

承接的问题

8.1 的 ReconnectHandler 维护一条到 9093 的长连接，断线重连、请求排队都在这一条连接上完成。这套模型放到网关就不成立了：GatewayHandler 收到 GET /books/{id} 后要转发到 libraryd 的 8080 端口，HTTP/1.1 一条连接同一时刻只能处理一个请求，如果所有请求共用一条 channel，响应会互相覆盖；如果每个请求新建一次 Bootstrap.connect()，每次都要付 TCP 握手，QPS 一高就把 fd 和端口耗光。

连接池解决的是「物理连接的复用」，但它不解决所有并发问题。linkd-v2 同时存在两种后端：

  • libraryd（HTTP/1.1）：一个请求占用一条连接直到响应回来，适合 FixedChannelPool 的借还模型。
  • linkd 9093（自定义协议）：7.1 的帧头带 requestId，一条连接上可以同时有多个在途请求，池只需要维持固定几条长连接，不需要按请求借还。

这一节先把 FixedChannelPool 的机制拆开，再说明两种模型的分界，最后给出网关侧按 host:port 建池的代码。HTTP/2 的流级多路复用只在取舍表里对比，具体实现留给第 11 章。

本机口径：Apple M1 Pro，10 核，32 GB，macOS 26.3 (25D125)；Temurin 25.0.5+8，Netty 4.2.19.Final。池大小相关的数字都是算式示例，不是压测结论。

核心机制一：FixedChannelPool 的构造与三个回调

FixedChannelPool 在 netty-transport 模块，构造参数决定了池的上限、排队上限和获取超时：

FixedChannelPool pool = new FixedChannelPool(
        bootstrap,                              // 已配置 remoteAddress 的 Bootstrap
        new BackendPoolHandler(),               // 连接生命周期回调
        ChannelHealthChecker.ACTIVE,            // 获取时健康检查
        FixedChannelPool.AcquireTimeoutAction.FAIL,
        500,                                    // acquireTimeoutMillis
        32,                                     // maxConnections
        512);                                   // maxPendingAcquires
参数作用调错的后果
maxConnections物理连接上限太小排队，太大压垮后端
maxPendingAcquires等待获取的请求上限无界会让内存随积压上涨
acquireTimeoutMillis等待可用连接的最长时间太长会把超时压力推给上游
AcquireTimeoutAction超时后的动作NEW 会突破 maxConnections
healthChecker借出前判断连接是否可用只查 isActive 会借出半死连接

AcquireTimeoutAction.FAIL 让获取超时的请求立刻失败；NEW 会新建一条连接交给调用方，即使已经达到 maxConnections。NEW 适合突发流量，但它把「池有上限」这个前提打破了，后端看到的连接数可能远超配置，使用前要确认后端能承受。

ChannelPoolHandler 的三个回调分工不同：

回调调用时机该做什么
channelCreated物理连接建好、第一次交给池时，一次装每连接 handler、打日志
channelAcquired每次 acquire() 成功重置请求级属性、检查可写
channelReleased每次 release()清理属性、关闭不该回池的连接
package com.plume.linkd.v2.client;

import io.netty.channel.Channel;
import io.netty.channel.pool.ChannelPoolHandler;

public final class BackendPoolHandler implements ChannelPoolHandler {

    @Override
    public void channelCreated(Channel ch) {
        ch.pipeline().addLast(new BackendResponseHandler());
    }

    @Override
    public void channelAcquired(Channel ch) {
        ch.attr(BackendResponseHandler.RESPONSE).set(null);
    }

    @Override
    public void channelReleased(Channel ch) {
        if (!ch.isActive()) {
            ch.close();
        }
    }
}

channelCreated 在 Bootstrap.handler 的 ChannelInitializer 执行之后调用，编解码器已经在 pipeline 里；在这里再加一个 HttpClientCodec 会重复解码，是常见错误。channelAcquired 和 channelReleased 成对出现，请求级状态必须在这两个回调里清掉，否则下一个借用者会读到上一个请求的残留。

核心机制二：ChannelHealthChecker 与获取超时

ChannelHealthChecker.ACTIVE 只判断 channel.isActive()。TCP 连接还在、但后端应用已经不处理请求时，它仍然返回健康，调用方借到一条「能写但不会有响应」的连接。要更严格，就把握手结果或最近一次心跳写进 channel 属性：

package com.plume.linkd.v2.client;

import io.netty.channel.Channel;
import io.netty.channel.pool.ChannelHealthChecker;
import io.netty.util.concurrent.Future;

public final class HandshakeHealthChecker implements ChannelHealthChecker {

    @Override
    public Future<Boolean> isHealthy(Channel channel) {
        boolean healthy = channel.isActive()
                && Boolean.TRUE.equals(channel.attr(ClientHandshakeHandler.READY).get());
        return channel.eventLoop().newSucceededFuture(healthy);
    }
}

isHealthy 返回 Netty 的 Future<Boolean>，允许异步判定；不要在这里发一个 PING 然后 sync() 等 PONG，那会阻塞 EventLoop。READY 是客户端 ClientHandshakeHandler 暴露的 AttributeKey<Boolean>，握手成功时置 true；需要网络往返的健康检查由 7.2 的 IdleStateHandler 加心跳维护属性，检查器只读属性。

获取超时有两层，别混在一起：

Future<Channel> acquired = pool.acquire();
Channel channel;
try {
    channel = acquired.get(1, TimeUnit.SECONDS);
} catch (ExecutionException e) {
    throw new IllegalStateException("acquire failed", e.getCause());
} catch (TimeoutException e) {
    acquired.cancel(false);
    throw new IllegalStateException("caller wait timeout", e);
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();
    throw new IllegalStateException("interrupted", e);
}

池的 acquireTimeoutMillis 是 500 ms，调用方等 1 秒。这样超时先由池触发，get() 抛出的 ExecutionException 里能拿到池给的原因；如果两层都设 500 ms，调用方自己的 TimeoutException 会盖掉池的失败信息。池的等待队列满了（maxPendingAcquires）时，acquire() 的 future 会立即失败，不会等到 500 ms。

核心机制三：AbstractChannelPoolMap 按 host:port 建池

网关可能转发到多个 libraryd 实例，池必须按目标地址分开：不同 host 的连接不能混用，池上限也要各自计算。AbstractChannelPoolMap 把「key → 池」的懒创建和关闭收在一起：

package com.plume.linkd.v2.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.pool.AbstractChannelPoolMap;
import io.netty.channel.pool.ChannelHealthChecker;
import io.netty.channel.pool.FixedChannelPool;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.http.HttpObjectAggregator;
import java.net.InetSocketAddress;

public final class HttpBackendPools
        extends AbstractChannelPoolMap<InetSocketAddress, FixedChannelPool> {

    private final EventLoopGroup group;
    private final int maxConnections;
    private final int maxPendingAcquires;

    public HttpBackendPools(EventLoopGroup group, int maxConnections, int maxPendingAcquires) {
        this.group = group;
        this.maxConnections = maxConnections;
        this.maxPendingAcquires = maxPendingAcquires;
    }

    @Override
    protected FixedChannelPool newPool(InetSocketAddress key) {
        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));
                    }
                });
        return new FixedChannelPool(bootstrap, new BackendPoolHandler(),
                ChannelHealthChecker.ACTIVE, FixedChannelPool.AcquireTimeoutAction.FAIL,
                500, maxConnections, maxPendingAcquires);
    }
}

三个要点：

  • 每个池一个 Bootstrap。 AbstractBootstrap 不是线程安全的，池的建连又可能发生在不同 EventLoop 上；共用同一个 Bootstrap 实例会让配置和 remoteAddress 互相覆盖。在 newPool 里新建，成本可以忽略。
  • remoteAddress(key) 必须设置。 FixedChannelPool 内部调用无参 bootstrap.connect()，地址来自 Bootstrap 的 remoteAddress；不设置会在第一次建连时报「远程地址未指定」。
  • key 用 InetSocketAddress.createUnresolved(host, port)。 new InetSocketAddress(host, port) 会在调用线程做 DNS 解析，网关的 EventLoop 线程不应该被 DNS 阻塞；未解析地址的 equals 只比较 host 和 port，适合做 map key。

HttpBackendPools 继承的 close() 会关闭所有池和池里的连接，必须在 8.3 的优雅关闭里调用；只关 EventLoopGroup 不会逐个释放池的引用计数和属性。

核心机制四：池大小与多路复用的取舍

池大小不是「越大越好」，也不是「等于 QPS」。用 Little 定律估下界：

L = λ × W

λ 是到达率，W 是请求在后端的平均驻留时间，L 是平均在途请求数。池至少要覆盖 L，否则请求会在 acquire() 上排队，客户端看到的延迟里混进了池等待时间。

场景λWL = λW建议 maxConnections
网关转发图书详情500 req/s40 ms2032
批量导入代理100 req/s300 ms3048

W 要用后端处理时间，不是客户端观测到的 p99；把排队时间算进 W 会高估所需连接。32 是在 20 之上留出的余量，用于吸收抖动，不是算出来的精确值。

三种并发模型要分开看：

维度HTTP/1.1 连接池HTTP/2 单连接多流linkd 9093 requestId
并发单位物理连接stream帧里的 requestId
每请求借还是否否
典型连接数多个每目标 1 条每目标固定几条
队头阻塞连接级TCP 层仍有应用层自行排序
连接故障影响该连接上的一个请求该连接上的所有流该连接上的所有在途请求

HTTP/2 的多路复用把并发单位从连接降到了流，但它没有消除 TCP 层的队头阻塞：一个丢包仍会拖住这条连接上的所有流。所以「用了 HTTP/2 就不需要池」只对「每个目标一条连接」成立，不对「一条连接扛无限并发」成立。

linkd-v2 的 9093 协议属于第三种：requestId 已经在应用层多路复用，FixedChannelPool 的按请求借还反而会增加开销。正确做法是每个后端维持 2–4 条长连接，RpcClient 按 requestId 轮转发帧；池只负责建连和健康检查，不参与请求调度。把 FixedChannelPool 硬套到这条链路上，会把「一条连接多请求」退化成「一个请求一条连接」。

在 linkd-v2 里落地：网关的借还与响应关联

GatewayHandler 不直接持有 channel，它通过 HttpBackendPools 借一条、写请求、等响应、还回去。响应通过 channel 属性关联到调用方的 future：

package com.plume.linkd.v2.client;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.util.AttributeKey;
import java.util.concurrent.CompletableFuture;

public final class BackendResponseHandler extends SimpleChannelInboundHandler<FullHttpResponse> {

    static final AttributeKey<CompletableFuture<FullHttpResponse>> RESPONSE =
            AttributeKey.valueOf("linkd.pool.response");

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, FullHttpResponse response) {
        CompletableFuture<FullHttpResponse> future =
                ctx.channel().attr(RESPONSE).getAndSet(null);
        if (future != null) {
            future.complete(response.retain());
        }
    }
}

SimpleChannelInboundHandler 会在 channelRead0 返回后释放 response，所以交给 future 的必须是 retain() 之后的引用；消费方读完要自己 release()。如果这里直接 future.complete(response)，调用方拿到的对象已经失效，表现为 IllegalReferenceCountException 或读到脏数据。

借还的完整流程：

public CompletableFuture<FullHttpResponse> get(String host, int port, String path, Duration timeout) {
    FixedChannelPool pool = pools.get(InetSocketAddress.createUnresolved(host, port));
    CompletableFuture<FullHttpResponse> result = new CompletableFuture<>();
    pool.acquire().addListener((Future<Channel> f) -> {
        if (!f.isSuccess()) {
            result.completeExceptionally(f.cause());
            return;
        }
        Channel channel = f.getNow();
        channel.attr(BackendResponseHandler.RESPONSE).set(result);
        result.whenComplete((response, error) -> pool.release(channel));
        FullHttpRequest request = new DefaultFullHttpRequest(
                HttpVersion.HTTP_1_1, HttpMethod.GET, path);
        HttpUtil.setKeepAlive(request, true);
        channel.writeAndFlush(request).addListener((ChannelFutureListener) w -> {
            if (!w.isSuccess()) {
                result.completeExceptionally(w.cause());
            }
        });
    });
    return result.orTimeout(timeout.toMillis(), TimeUnit.MILLISECONDS);
}

流程里的所有权和时序：

  • whenComplete 负责归还。 正常响应、写失败、超时三条路径都会让 result 完成，release 只执行一次；不要同时在写监听器和读处理器里 release。
  • 超时由 orTimeout 触发。 超时后 future 异常完成，whenComplete 立即归还连接；此时响应可能还在路上，回池前 channelReleased 的健康检查只检查 isActive，无法判断响应是否已经被丢弃。对时序敏感的后端，超时后应该关闭连接而不是回池。
  • HttpUtil.setKeepAlive(request, true) 与后端能力匹配。 如果 libraryd 返回 Connection: close，这条连接不该再借出；可以在响应处理器里记一个属性，channelReleased 看到后 close()。
  • result 只完成一次。 写失败后再收到响应、或超时后响应到达，complete 都是空操作，不会重复 release。

常见错误与边界

错误一：在 channelCreated 里重复添加编解码器。

// 反例：Bootstrap.handler 已经加过 HttpClientCodec
@Override
public void channelCreated(Channel ch) {
    ch.pipeline().addLast(new HttpClientCodec());
}

channelCreated 在 ChannelInitializer 之后执行，再加一次会让同一条连接上出现两个 HttpClientCodec，响应被解码两次，第二个解码器读到的是空缓冲，日志里会出现莫名的 DecoderException。

错误二：把请求级 handler 加到池化连接上。

// 反例：每次 acquire 都加一个 handler
pool.acquire().addListener((Future<Channel> f) -> {
    f.getNow().pipeline().addLast(new ResponseCollector(result));
});

handler 不会随 release 移除，同一条连接被借 N 次就累积 N 个 ResponseCollector，响应会被第一个实例消费，后面的实例永远等不到，pipeline 也越来越长。请求级状态放 AttributeKey，handler 在 channelCreated 里只加一次。

错误三：把 maxConnections 设成 QPS。

1000 QPS 不等于 1000 条连接。连接数由「在途请求数」决定，在途请求数由到达率和后端处理时间共同决定。按 QPS 配置会让后端瞬间收到上千条连接，每条都要握手、分配缓冲区、占用一个 worker；实际并发可能只有几十。

边界：连接池不修复慢后端。 池只限制并发上限，后端变慢时，请求会在 acquire() 或响应等待上排队，池满后新请求快速失败。这是保护后端的手段，不是提升吞吐的手段；吞吐问题要看后端处理时间和连接上的并发模型。

小结

  • FixedChannelPool 用 maxConnections 限制物理连接，用 maxPendingAcquires 限制等待队列，用 acquireTimeoutMillis 限制等待时间。
  • channelCreated 每连接一次，channelAcquired 和 channelReleased 每次借还各一次；请求级状态必须在后两者里清理。
  • ChannelHealthChecker.ACTIVE 只查 isActive()，要识别半死连接就把握手或心跳结果放进属性，检查器只读属性，不做阻塞 IO。
  • AbstractChannelPoolMap 按 host:port 懒建池，每个池要有独立的 Bootstrap 和 remoteAddress，key 用未解析地址避免调用线程做 DNS。
  • 池大小按 Little 定律 L = λW 估下界再加余量，不按 QPS 配置，也不是越大越好。
  • HTTP/1.1 按请求借还，HTTP/2 按流复用，linkd 9093 按 requestId 复用；三种模型不能混用同一套池语义。

连接借出、用完、归还之后，还剩一个问题：这条连接到底该回池还是该关闭，半关闭的连接怎么算，进程退出时怎么在期限内把连接收干净。8.3 处理关闭语义与优雅回收。

阅读导航：上一节：8.1 客户端连接与重连 · 下一节：8.3 关闭、半关闭与连接回收 。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「java」更多文章

  1. 《Netty 高性能网络编程》15.3 linkd-v2 验收与复盘
  2. 《Netty 高性能网络编程》15.2 限流、超时与降级
  3. 《Netty 高性能网络编程》15.1 网关路由与转发