《Java 服务端通信工程》6.2 客户端连接池设计

本节目标：能判断什么时候需要多条 channel、池的键应该包含什么、上限怎么定，并写出一个有界、可关闭、不会每请求建连的 LibraryChannelPool。
适用版本：JDK 25 + gRPC-Java 1.76.0（grpc-netty-shaded）。

承接的问题

6.1 的结论是「默认一条连接」。但 linkd-v3 的网关同时服务多个租户，有的租户一次导入几万条书目，长请求会把一条连接上的流额度占满，其他租户的 GetBook 只能排队。这时一条 channel 不是「够不够」的问题，而是「吵不吵」的问题：一个租户的批量任务会把整条连接的流控窗口和在途流额度吃光。

另一种情况是并发额度。服务端把 maxConcurrentCallsPerConnection 设为 64，某个租户的峰值在途请求是 200；如果所有租户共用一条连接，第 65 个请求就要等前面的流结束。建第二条 channel 可以让两个租户分别占用 64 条流，但这会放大服务端的连接数和内存。

连接池要解决的是「有界地复用多条 channel」。它和 HTTP/1.1 的连接池不同：HTTP/1.1 是「一个请求借一条连接，用完归还」；gRPC 是「一次借一条 channel，期间可以发很多请求，channel 一直复用」。把 HTTP/1.1 的借还模型照搬过来，会退化成每请求一条连接，正好丢掉 HTTP/2 的多路复用。

这一节先列「什么时候需要多条 channel」，再设计池键和上限，给出完整实现，最后讨论池大小与流额度的换算。关闭与 drain 在 6.3，本节假设池的生命周期由进程管理。

本机口径：Apple M1 Pro，10 核，32 GB，macOS 26.3 (25D125)；Temurin 25.0.5+8，gRPC-Java 1.76.0。本节不给吞吐数字，池大小都是算式示例。

核心机制一：什么时候需要多条 Channel

一条 channel 是默认，多条是例外。例外的理由只有三类：

「想更快」不是理由。多建连接不会让服务端 CPU 变快，也不会让数据库变快；如果瓶颈在服务端处理时间，加连接只会让更多请求同时压进去。判断顺序应该是：先看服务端在途请求和处理时间，再看单连接流额度是否打满，最后才看隔离需求。

凭证和 authority 不同也必须分开建 channel。一个 channel 绑定一组 CallCredentials 和 target，不能在同一条 channel 上切换租户的 TLS 身份；如果每个租户有独立的 mTLS 证书，channel 必须按租户拆。linkd-v3 第一版用统一的内部凭证，租户信息走 metadata，所以这条不适用；10.1 讲 mTLS 时会回到这个约束。

round_robin 的 LB 状态也挂在 channel 上。两条 channel 各自维护一套子通道和健康状态，不会共享；这意味着实例列表变化时，两条 channel 会各自刷新，收敛时间可能不同。池的键不要包含「随机数」或「请求 ID」，否则每次都是新 channel，池就退化成建连器。

核心机制二：池的键、上限与借还语义

池键至少包含三部分：

purpose 很重要。批量导入和在线查询的延迟目标不同，混在一条 channel 上，批量任务的长流会拖慢在线查询。分成两个池，各自有独立的流额度和流控窗口，互不影响。

池的上限有两个层面：channel 总数和每个键的 channel 数。第一版每个键最多一条 channel，channel 总数由 maxChannels 限制；需要给某个大租户两条时，把 tenant 再细分成 tenant-a#1、tenant-a#2，或者把键设计成 (target, tenant, purpose, shard)。不要在一个键上维护多条 channel 的轮询，那会把 LB 的职责复制一遍。

借还语义是「借 channel，不是借连接」：

• acquire(key) 返回一个 Lease，里面是 ManagedChannel；同一个键的多个租约拿到同一条 channel。
• Lease.close() 只释放租约，不关闭 channel；channel 回到池里继续复用。
• 池关闭时统一 shutdown() 所有 channel，等 drain 完成再退出。

下面是一个有界实现。slots 信号量按「已创建的 channel 数」计数，而不是按「并发租约数」计数；这样池里的 channel 总数不会超过 maxChannels。

package com.plume.linkd.v3.core.client;

import io.grpc.InsecureChannelCredentials;
import io.grpc.ManagedChannel;
import io.grpc.Status;
import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

public final class LibraryChannelPool implements AutoCloseable {

    public record PoolKey(String target, String tenant, String purpose) {
        public PoolKey {
            if (target == null || target.isBlank()) {
                throw new IllegalArgumentException("target is blank");
            }
            if (tenant == null || tenant.isBlank()) {
                throw new IllegalArgumentException("tenant is blank");
            }
            if (purpose == null || purpose.isBlank()) {
                throw new IllegalArgumentException("purpose is blank");
            }
        }
    }

    public static final class Lease implements AutoCloseable {
        private final ManagedChannel channel;
        private final AtomicBoolean closed = new AtomicBoolean();

        private Lease(ManagedChannel channel) {
            this.channel = channel;
        }

        public ManagedChannel channel() {
            return channel;
        }

        @Override
        public void close() {
            closed.set(true);
        }
    }

    private final ConcurrentHashMap<PoolKey, ManagedChannel> channels =
            new ConcurrentHashMap<>();
    private final Semaphore slots;
    private final AtomicBoolean closed = new AtomicBoolean();

    public LibraryChannelPool(int maxChannels) {
        if (maxChannels <= 0) {
            throw new IllegalArgumentException("maxChannels must be positive");
        }
        this.slots = new Semaphore(maxChannels);
    }

    public Lease acquire(PoolKey key, Duration wait) {
        if (closed.get()) {
            throw new IllegalStateException("pool is closed");
        }
        ManagedChannel existing = channels.get(key);
        if (existing != null) {
            return new Lease(existing);
        }

        boolean acquired;
        try {
            acquired = slots.tryAcquire(wait.toMillis(), TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException("interrupted while acquiring channel", e);
        }
        if (!acquired) {
            throw Status.RESOURCE_EXHAUSTED
                    .withDescription("channel pool exhausted for " + key.target())
                    .asRuntimeException();
        }

        boolean created = false;
        try {
            ManagedChannel channel = channels.get(key);
            if (channel == null) {
                channel = build(key);
                channels.put(key, channel);
                created = true;
            }
            return new Lease(channel);
        } finally {
            if (!created) {
                slots.release();
            }
        }
    }

    private ManagedChannel build(PoolKey key) {
        return NettyChannelBuilder
                .forTarget(key.target(), InsecureChannelCredentials.create())
                .defaultLoadBalancingPolicy("round_robin")
                .maxInboundMessageSize(4 * 1024 * 1024)
                .idleTimeout(5, TimeUnit.MINUTES)
                .build();
    }

    public List<PoolKey> keys() {
        return List.copyOf(channels.keySet());
    }

    @Override
    public void close() {
        if (!closed.compareAndSet(false, true)) {
            return;
        }
        List<ManagedChannel> snapshot = new ArrayList<>(channels.values());
        channels.clear();
        for (ManagedChannel channel : snapshot) {
            channel.shutdown();
        }
        for (ManagedChannel channel : snapshot) {
            try {
                if (!channel.awaitTermination(5, TimeUnit.SECONDS)) {
                    channel.shutdownNow();
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                channel.shutdownNow();
            }
        }
    }
}

几个设计点：

• 先查 map，再抢 slot。 常见路径（键已存在）不碰信号量，避免每次借 channel 都做一次 tryAcquire。
• 二次检查。 抢到 slot 后可能已经有别的线程创建了同一个键的 channel；这时释放 slot，复用已有的，保证「一个键一条 channel」。
• Lease.close() 不关 channel。 它只标记租约结束，channel 留在池里；如果每次 close 都 shutdown()，池就变成每请求一条连接。
• close() 先 shutdown() 再等待。 这与 6.3 的 drain 语义一致；池关闭后 acquire 立即失败，不会新建连接。
• idleTimeout(5, MINUTES) 让空闲连接进入 IDLE。 channel 对象还在 map 里，但底层 TCP 连接会被释放；下次调用时重新建连。这比在池里手工驱逐 idle channel 简单。

核心机制三：池大小与流额度

池的总 channel 数由 maxChannels 限制，单条 channel 的并发额度由服务端的 maxConcurrentCallsPerConnection 决定。两者相乘是「池能同时承载的在途流数」上限：

池容量 = maxChannels × maxConcurrentCallsPerConnection

但这不是让你把两个数都调大。先算需求：

需要的 channel 数 = ceil(峰值在途流数 / 单连接流额度)

例如服务端单连接 64 条流，某租户峰值 200 条在途流，需要 ceil(200/64)=4 条 channel。给这个租户 4 条，给其他租户各 1 条，maxChannels 至少是 4 + 其他租户数。如果池只有 8 条，10 个租户同时要 channel，第 9 个会拿到 RESOURCE_EXHAUSTED；这个错误应该被网关转成 503，而不是无限等待。

峰值在途流数不能直接用 QPS 乘处理时间，因为 streaming 流会长期占用额度。WatchStock 每条流占一个额度直到客户端取消，按 QPS 算会严重低估。把 unary 和 streaming 分开统计，或者给 streaming 单独一个池。

流额度是「每连接」的，不是「每租户」的。同一租户的两条 channel 各自有 64 条流额度，但服务端看到的是 128 条；如果服务端还有全局并发限制（8.2 的 Bulkhead(64, 128)），多出来的流会在服务端排队或被拒绝。池大小要和 8.2 的全局上限一起算，不能只看单连接。

核心机制四：池的生命周期与空闲回收

池里的 channel 有三条出路：

idleTimeout 只影响底层连接，不会把 channel 从 map 里移除；map 的大小仍然受 maxChannels 限制。要让「长时间不用的租户」彻底释放 channel 对象，需要额外的驱逐逻辑：记录每个键的 lastUsedNanos，定期扫描，shutdown() 并 remove，同时 slots.release()。第一版不做这件事，因为 channel 对象本身很小，真正的资源是 TCP 连接，而 idleTimeout 已经处理了。

驱逐时有一个竞态：扫描线程判断「空闲」，同时业务线程 acquire 拿到同一个 channel，扫描线程把它 shutdown()，业务调用失败。解决办法是给每个键加引用计数或读写锁，或者规定「驱逐只发生在持有池锁的短窗口内」。如果不想引入锁，就不要做主动驱逐，交给 idleTimeout 和进程退出。

池关闭必须在 drain 之后。6.3 的顺序是：停止入口流量 → 等待在途请求 → pool.close() → 进程退出。反过来先 pool.close()，在途请求会收到 UNAVAILABLE，正在写库的操作可能只完成一半。

在 linkd-v3 里落地：按租户隔离的读池

linkd-v3 的网关有两个池：

• 在线读池：键是 (target, tenant, "read")，每个租户最多一条 channel，总数 8 条。
• 批量池：键是 (target, tenant, "bulk")，只给导入任务用，总数 2 条，避免批量流挤占在线查询。

配置从环境变量读取，默认值写死为 8 和 2，超出时 acquire 抛 RESOURCE_EXHAUSTED。网关把这个状态映射成 HTTP 503，并带上 Retry-After: 1：

LibraryChannelPool readPool = new LibraryChannelPool(8);
LibraryChannelPool.PoolKey key =
        new LibraryChannelPool.PoolKey("static-registry:///library", tenantId, "read");

try (LibraryChannelPool.Lease lease = readPool.acquire(key, Duration.ofMillis(100))) {
    Book book = LibraryGrpc.newBlockingStub(lease.channel())
            .withDeadlineAfter(300, TimeUnit.MILLISECONDS)
            .getBook(GetBookRequest.newBuilder().setIsbn(isbn).build());
    return book;
}

acquire 的等待时间 100 ms 是池满时的排队预算，不是调用 deadline。调用本身还有 300 ms 的单跳 deadline 和 800 ms 的总 deadline；三层预算要满足 100 + 300 < 800，否则池等待会把调用预算吃光，业务还没发出去就超时了。

租户 ID 来自 4.2 的 x-tenant-id，必须经过校验，不能直接拼进池键。一个恶意租户每次请求换一个 ID，会把池撑满；网关应该先做租户白名单，再把 ID 作为键。池键的基数要可控，purpose 用枚举而不是自由字符串。

常见错误与边界

错误一：每请求建 channel。 这是把连接池写成建连器。channel 是长生命周期对象，租约借用的是 channel 的引用，不是 TCP 连接；HTTP/2 的一条连接本来就支持多流。

错误二：池大小等于 QPS。 1000 QPS 不需要 1000 条连接；连接数由在途流和单连接额度决定。按 QPS 配置会让服务端瞬间收到上千条连接，实际并发可能只有几十。

错误三：Lease.close() 里 shutdown() channel。 下一个租约拿到的是已关闭的 channel，调用全部 UNAVAILABLE。租约只释放引用，channel 由池统一关闭。

错误四：池键包含请求 ID 或时间戳。 每次都是新键，map 无限增长，maxChannels 也挡不住（因为 slot 会随 close 释放）。池键必须是有限、稳定的维度。

错误五：忽略 RESOURCE_EXHAUSTED。 池满时应该快速失败并让上游重试/降级，不是无限等待。等待时间要小于调用 deadline，错误要映射成明确的上游状态。

错误六：池关闭顺序反了。 先关池再停流量，在途请求会失败。先停入口、再 drain、最后关池。

边界：连接池不提升单请求性能。 它只改变并发和隔离；单请求的延迟由网络、服务端处理和流控决定。池化不会让 300 ms 的调用变成 100 ms。

边界：多 channel 不等于多路复用。 每条 channel 内部仍然多路复用；池只是把流分散到多条连接上。不要在一个 channel 上再手工做「连接池」，那是重复抽象。

小结

• 默认一条 channel；多条的理由是隔离、并发额度或故障半径，不是「想更快」。
• 池键至少包含 target、tenant、purpose；键的基数要可控，不能包含请求 ID。
• LibraryChannelPool 用 Semaphore 限制 channel 总数，用 ConcurrentHashMap 复用同键 channel；Lease.close() 只释放租约，不关闭 channel。
• 池容量按 ceil(峰值在途流 / 单连接流额度) 估算，并和服务端全局并发上限一起算。
• idleTimeout 释放底层连接但保留 channel 对象；主动驱逐有空闲竞态，第一版交给 idle 和进程退出。
• 关闭顺序是「停入口 → drain → 关池」；池满时快速失败，不要无限等待。
• 本节的数字都是算式示例，没有对应的压测结论。

下一节处理 channel 和 server 的关闭语义：shutdown()、shutdownNow()、awaitTermination() 分别做什么，进程收到 SIGTERM 后怎么在期限内把在途请求收干净。

阅读导航：上一节 6.1 HTTP/2 多路复用与连接数 ｜ 下一节 6.3 连接生命周期与 drain