汀的知识碎片

09 基于Netty的RPC框架设计——序列化、路由与连接管理

27 min read

基于Netty的RPC框架设计——序列化、路由与连接管理

摘要

RPC（Remote Procedure Call，远程过程调用）框架是现代微服务架构的基础设施，而 Netty 是构建高性能 RPC 框架的事实标准网络层。本文以”从零设计一个生产级 RPC 框架”为主线，系统梳理 RPC 框架的核心问题域：序列化（如何高效地在二进制字节流与 Java 对象之间转换）、协议设计（如何在自定义二进制协议中携带完整的调用语义）、连接管理（连接池的设计与连接的健康检测）、服务路由（如何从多个服务实例中选择一个）、超时与重试（如何在分布式不可靠网络中保证语义正确性）。通过这些问题的讨论，揭示 Dubbo、gRPC 等主流 RPC 框架在 Netty 层面的设计思路，并指出生产实践中最容易踩的坑。

第 1 章 RPC 的本质与挑战

1.1 RPC 要解决的核心问题

RPC 的目标是让跨网络的函数调用”看起来像”本地函数调用。调用方写下 userService.getUserById(123)，RPC 框架负责完成以下一系列工作：

  1. 序列化：将参数 123（和方法名、服务名等元信息）序列化为二进制字节流；
  2. 传输：通过 TCP 连接将字节流发送给服务提供方；
  3. 反序列化：服务提供方将字节流还原为方法调用信息；
  4. 执行：调用 userService.getUserById(123) 的真实实现，得到结果；
  5. 序列化返回值：将返回的 User 对象序列化；
  6. 传输：将响应字节流发回调用方；
  7. 反序列化：调用方将字节流还原为 User 对象，返回给业务代码。

这 7 个步骤，每一步都有复杂的工程问题要解决。而 RPC 的”欺骗性”在于：让调用方感知不到这些中间步骤——这要求框架的异常处理、超时机制、网络抖动处理都足够透明和健壮。

1.2 分布式系统的 8 大谬误

1991 年，Sun Microsystems 的 Peter Deutsch 列出了”分布式计算的 8 大谬误”，每一条都是 RPC 框架必须面对的现实：

  1. 网络是可靠的：事实上，网络抖动、包丢失、连接断开随时发生；
  2. 延迟为零：本地调用纳秒级，网络调用毫秒级，差距是数万倍；
  3. 带宽无限：大对象传输受带宽限制；
  4. 网络是安全的：需要 TLS 等安全机制；
  5. 拓扑结构不变：服务实例会上线/下线/迁移；
  6. 只有一个管理员：多团队维护的系统可能有不一致的配置；
  7. 传输成本为零：序列化/反序列化、网络 I/O 都有 CPU/IO 成本；
  8. 网络是同质的：不同机器/机房的网络特性不同。

一个成熟的 RPC 框架必须在设计层面应对这 8 个谬误，这也是为什么 RPC 框架的代码往往比想象中复杂得多。

第 2 章 序列化方案的选择与权衡

2.1 序列化的核心指标

评估一个序列化方案，需要从四个维度考量：

维度说明高重要性场景
序列化速度单位时间内能序列化/反序列化的对象数量高 QPS 的内部 RPC
序列化大小同一对象序列化后的字节数带宽敏感（跨机房调用、移动端）
跨语言兼容性是否支持多语言客户端调用异构系统集成
可读性与调试序列化结果是否人类可读开发调试阶段

2.2 主流序列化方案对比

JSON（Jackson / Fastjson2）：

文本格式，可读性最好，跨语言兼容性极强。缺点：序列化体积大（字段名重复传输），速度慢（文本解析开销）。适合对外的 HTTP API（如 REST）、对调试友好度要求高的内部接口。

Protobuf（Protocol Buffers）：

Google 开源的二进制序列化格式，需要预先定义 .proto 文件（IDL），代码生成。序列化体积极小（不传字段名，用字段编号），速度极快，跨语言支持完善（官方支持 Java/Go/C++/Python 等 10+ 语言）。gRPC 默认使用 Protobuf。

缺点：IDL 文件增加了额外的维护成本，对象结构变更需要同步更新 .proto 文件和重新生成代码。

Hessian2：

Dubbo 默认使用的序列化格式，二进制格式，无需 IDL，直接序列化 Java 对象。相比 JSON 体积小、速度快；相比 Protobuf 体积稍大、速度稍慢，但无需 IDL 文件，开发体验好。

缺点：跨语言支持有限（主要是 Java），不适合多语言异构场景。

Kryo：

纯 Java 的高性能序列化库，速度和体积都接近 Protobuf，但不需要 IDL。缺点：不支持跨语言，序列化格式没有版本控制（字段顺序/类型变更后无法兼容旧数据）。

性能横向对比（JMH 基准测试，序列化 + 反序列化 1000次，小型 POJO）：

序列化方案耗时（μs）序列化大小（字节）
Protobuf868
Kryo1282
Hessian245156
Jackson JSON95245
JDK Serializable380480

禁止在 RPC 框架中使用 JDK 原生序列化

JDK ObjectOutputStream/ObjectInputStream 是所有序列化方案中性能最差的，体积最大，而且存在严重的安全漏洞（反序列化攻击）。生产 RPC 框架无一使用它。

2.3 序列化扩展点设计

成熟的 RPC 框架会将序列化方案设计为可插拔的扩展点，通过在协议头中携带序列化类型，允许调用方和服务方协商使用哪种序列化：

// 序列化接口（SPI 扩展点）
public interface Serializer {
    byte getType();  // 序列化类型标识（写入协议头）
    <T> byte[] serialize(T obj);
    <T> T deserialize(byte[] bytes, Class<T> clazz);
}
 
// JSON 实现
public class JsonSerializer implements Serializer {
    @Override
    public byte getType() { return 0x01; }
    
    @Override
    public <T> byte[] serialize(T obj) {
        return JSON.toJSONBytes(obj);
    }
    
    @Override
    public <T> T deserialize(byte[] bytes, Class<T> clazz) {
        return JSON.parseObject(bytes, clazz);
    }
}
 
// Protobuf 实现
public class ProtobufSerializer implements Serializer {
    @Override
    public byte getType() { return 0x02; }
    // ...
}

第 3 章 RPC 协议设计

3.1 协议头需要携带哪些信息

RPC 的协议头不仅要携带序列化后的业务数据，还要携带足够的元信息让接收方能正确处理：

┌─────────────────────────────────────────────────────────────────────────────┐
│ magic(2) │ version(1) │ msgType(1) │ serializer(1) │ status(1) │ reqId(8)   │
├─────────────────────────────────────────────────────────────────────────────┤
│                          bodyLength(4)                                       │
├─────────────────────────────────────────────────────────────────────────────┤
│                          body (N 字节)                                       │
└─────────────────────────────────────────────────────────────────────────────┘
固定头部：2 + 1 + 1 + 1 + 1 + 8 + 4 = 18 字节

字段说明：

  • magic（2字节）：协议魔数，如 0xCAFE，识别协议类型，防止解析非法连接；
  • version（1字节）：协议版本，支持协议升级时的兼容处理；
  • msgType（1字节）：消息类型（0=请求，1=响应，2=心跳请求，3=心跳响应）；
  • serializer（1字节）：序列化类型（0=JSON，1=Hessian，2=Protobuf）；
  • status（1字节）：响应状态码（0=成功，1=业务异常，2=框架异常，3=超时）；
  • reqId（8字节）：请求 ID，用于在异步场景下匹配请求与响应（见第 4 章）；
  • bodyLength（4字节）：消息体字节数；
  • body（N字节）：序列化后的请求/响应对象。

3.2 请求/响应对象设计

// RPC 请求对象（序列化为 body 发送）
@Data
public class RpcRequest {
    private String requestId;      // 请求唯一 ID
    private String interfaceName;  // 服务接口名（如 "com.example.UserService"）
    private String methodName;     // 方法名（如 "getUserById"）
    private Class<?>[] paramTypes; // 参数类型数组（用于反射查找方法）
    private Object[] params;       // 参数值数组
    private String version;        // 服务版本（支持同接口多版本）
    private String group;          // 服务分组
}
 
// RPC 响应对象
@Data
public class RpcResponse<T> {
    private String requestId;  // 对应的请求 ID
    private int code;          // 状态码（200=成功，500=服务端异常）
    private String message;    // 错误信息（仅在错误时非空）
    private T data;            // 返回值
    
    public static <T> RpcResponse<T> success(String requestId, T data) {
        return new RpcResponse<>(requestId, 200, null, data);
    }
    
    public static <T> RpcResponse<T> fail(String requestId, String message) {
        return new RpcResponse<>(requestId, 500, message, null);
    }
}

第 4 章 异步调用的请求-响应匹配

4.1 异步调用的核心难题

Netty 的网络通信是全异步的：调用方发出请求后不阻塞，等响应到来时才处理。但”响应到来时”可能有多个响应同时到达（不同请求的响应在 TCP 层面按到达顺序排列），调用方必须知道”这个响应对应的是哪个请求”。

解决方案是 requestId：每个请求分配一个唯一 ID，响应中携带相同的 requestId，调用方通过 requestId 匹配。

4.2 CompletableFuture 实现异步等待

调用方维护一个 Map<String, CompletableFuture<RpcResponse>>，以 requestId 为键：

// 客户端的异步请求-响应匹配机制
public class PendingRequestMap {
    // 等待中的请求：requestId → CompletableFuture
    private final Map<String, CompletableFuture<RpcResponse<?>>> pendingRequests 
        = new ConcurrentHashMap<>();
    
    // 发送请求前，先注册 Future
    public CompletableFuture<RpcResponse<?>> register(String requestId) {
        CompletableFuture<RpcResponse<?>> future = new CompletableFuture<>();
        pendingRequests.put(requestId, future);
        return future;
    }
    
    // 响应到达时（在 Handler 的 channelRead 中调用），完成 Future
    public void complete(String requestId, RpcResponse<?> response) {
        CompletableFuture<RpcResponse<?>> future = pendingRequests.remove(requestId);
        if (future != null) {
            future.complete(response);
        }
        // 如果 future 为 null：说明请求已超时被移除，响应被丢弃
    }
    
    // 超时时移除 Future 并让其完成异常
    public void timeout(String requestId) {
        CompletableFuture<RpcResponse<?>> future = pendingRequests.remove(requestId);
        if (future != null) {
            future.completeExceptionally(
                new RpcTimeoutException("Request " + requestId + " timed out"));
        }
    }
}

完整的异步调用流程：

// RPC 客户端发起调用
public <T> T invoke(RpcRequest request, Class<T> returnType) throws Exception {
    String requestId = UUID.randomUUID().toString();
    request.setRequestId(requestId);
    
    // 1. 注册等待 Future
    CompletableFuture<RpcResponse<?>> future = pendingRequestMap.register(requestId);
    
    // 2. 发送请求（异步，立即返回）
    Channel channel = connectionPool.getChannel(serviceAddress);
    channel.writeAndFlush(encodeRequest(request));
    
    // 3. 注册超时定时器（3 秒）
    Timeout timeoutTask = timer.newTimeout(
        t -> pendingRequestMap.timeout(requestId),
        3, TimeUnit.SECONDS
    );
    
    // 4. 等待响应（阻塞调用线程，直到 Future 完成或超时）
    try {
        RpcResponse<?> response = future.get(3, TimeUnit.SECONDS);
        timeoutTask.cancel();  // 取消超时定时器
        
        if (response.getCode() != 200) {
            throw new RpcException(response.getMessage());
        }
        return returnType.cast(response.getData());
    } catch (TimeoutException e) {
        throw new RpcTimeoutException("RPC call timed out: " + request.getMethodName());
    }
}
 
// 响应到达时的 Handler
public class RpcClientHandler extends SimpleChannelInboundHandler<RpcMessage> {
    private final PendingRequestMap pendingRequestMap;
    
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcMessage msg) {
        if (msg.getType() == RpcMessage.TYPE_RESPONSE) {
            RpcResponse<?> response = deserialize(msg);
            // 通知等待的 Future
            pendingRequestMap.complete(response.getRequestId(), response);
        } else if (msg.getType() == RpcMessage.TYPE_HEARTBEAT_PONG) {
            // 心跳响应，更新连接活跃时间
            ctx.channel().attr(LAST_ACTIVE_KEY).set(System.currentTimeMillis());
        }
    }
}

第 5 章 连接管理：连接池的设计

5.1 为什么 RPC 需要连接池

每次 RPC 调用都新建 TCP 连接代价极高：TCP 三次握手约 1～3 个 RTT（局域网 ~0.1ms，跨机房 ~5ms），TLS 握手更需要 2～3 个额外 RTT。对于每秒数千次的 RPC 调用，连接创建开销会完全压垮服务。

复用长连接（Keep-Alive）是标准方案，而管理这些长连接的组件就是连接池。

5.2 连接池的核心设计要素

连接数量的上下限：

  • minConnections（最小连接数）：服务启动时预热建立，保证随时有连接可用；
  • maxConnections（最大连接数）：防止对下游服务造成连接数压力（对方的 acceptQueue 有上限）。

通常的经验值：对同一个服务实例保持 min=2, max=10 的连接数，具体根据 QPS 调整。

连接的健康检测：

TCP 连接可能在没有数据传输的情况下被中间设备（防火墙、NAT 网关）悄悄断开，连接池不知情，还以为连接是好的，导致发送第一个请求时失败。

解决方案：应用层心跳。每隔 N 秒发送一次心跳请求，对方回复心跳响应，以此证明连接是活跃的。Netty 的 IdleStateHandler 提供了连接空闲检测的标准实现：

// 在 Pipeline 中添加空闲检测
pipeline.addLast(new IdleStateHandler(
    0,              // readerIdleTime：多少秒没收到数据触发 READER_IDLE 事件（0=不检测）
    30,             // writerIdleTime：多少秒没发出数据触发 WRITER_IDLE 事件（30秒）
    0,              // allIdleTime：读写都空闲触发 ALL_IDLE 事件（0=不检测）
    TimeUnit.SECONDS
));
pipeline.addLast(new HeartbeatHandler());
 
// 心跳处理器
public class HeartbeatHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state() == IdleState.WRITER_IDLE) {
                // 30 秒没有发出数据：主动发送心跳
                RpcMessage heartbeat = RpcMessage.builder()
                    .type(RpcMessage.TYPE_HEARTBEAT_PING)
                    .requestId(UUID.randomUUID().toString())
                    .build();
                ctx.writeAndFlush(heartbeat).addListener(future -> {
                    if (!future.isSuccess()) {
                        // 心跳发送失败：关闭连接，触发重连
                        ctx.close();
                    }
                });
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }
}

连接的重建（重连机制）：

当连接断开时（channelInactive() 被触发），连接池需要异步重建连接。重连不能立即重试（可能服务端正在重启），而应该使用指数退避（Exponential Backoff）策略：

// 指数退避重连
public class ReconnectHandler extends ChannelInboundHandlerAdapter {
    private int retryCount = 0;
    private static final int MAX_RETRY = 10;
    private static final long INITIAL_DELAY_MS = 1000;  // 1 秒
    
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        // 连接断开：触发重连
        scheduleReconnect(ctx.channel().eventLoop());
        super.channelInactive(ctx);
    }
    
    private void scheduleReconnect(EventLoop eventLoop) {
        if (retryCount >= MAX_RETRY) {
            log.error("Max retry reached, giving up reconnection");
            return;
        }
        // 指数退避：1s, 2s, 4s, 8s, 16s, ...（上限 60s）
        long delay = Math.min(INITIAL_DELAY_MS * (1L << retryCount), 60_000L);
        retryCount++;
        
        eventLoop.schedule(() -> {
            log.info("Attempting reconnection #{}, delay={}ms", retryCount, delay);
            bootstrap.connect(remoteAddress).addListener(future -> {
                if (future.isSuccess()) {
                    log.info("Reconnected successfully");
                    retryCount = 0;  // 重置重试计数
                } else {
                    scheduleReconnect(eventLoop);  // 继续重试
                }
            });
        }, delay, TimeUnit.MILLISECONDS);
    }
}

5.3 连接选择策略

当连接池中有多个连接时，如何选择发送当前请求的连接？

轮询（Round Robin）：最简单，均衡分配请求到各连接。缺点：不考虑各连接的当前负载（某些连接可能有大量待处理请求积压）。

最少待处理请求（Least Pending Requests）：选择当前待处理请求数最少的连接。需要记录每个连接上未收到响应的请求数量，适合请求处理时间差异较大的场景。

第 6 章 服务发现与负载均衡

6.1 服务发现的必要性

在微服务架构中，服务实例的 IP 和端口是动态变化的（弹性伸缩、故障迁移、版本升级）。RPC 框架不能硬编码服务地址，而需要通过服务注册中心动态发现可用的服务实例列表。

主流服务注册中心：

  • ZooKeeper（Dubbo 传统默认）：CP 特性（一致性 > 可用性），节点使用临时节点实现服务注册；
  • Nacos（阿里，现 Dubbo 推荐）：支持 AP/CP 切换，性能更好，界面更友好；
  • Consul（HashiCorp）：内置健康检查，与 Kubernetes 集成良好；
  • Eureka（Netflix，Spring Cloud 传统默认）：AP 特性，简单但精度较低（服务下线后有延迟感知）。

6.2 客户端负载均衡

RPC 框架通常采用客户端负载均衡（区别于 Nginx 这类服务端负载均衡）：客户端从注册中心拿到全部服务实例列表，自己决定调用哪一个，不经过中间代理层，延迟更低，也更灵活。

常见负载均衡算法：

随机（Random）：简单有效，在实例数量多、请求量大时统计上趋于均匀。

轮询（Round Robin）：严格按顺序轮流，比随机更均匀，但所有实例权重相同。

加权轮询（Weighted Round Robin）：不同服务实例赋予不同权重（如高配机器权重 3，低配机器权重 1），按权重比例分配请求。适合机器配置不均匀的场景。

一致性哈希（Consistent Hashing）：根据请求参数（如用户 ID）计算哈希值，将相同参数的请求路由到同一实例。适合有本地缓存的服务（同一用户的请求命中同一实例的缓存）。

最小活跃数（Least Active）：将请求发给当前活跃请求数最少的实例，动态感知实例负载。Dubbo 默认使用这个算法。

第 7 章 超时与重试：分布式语义的正确处理

7.1 超时设置的艺术

超时时间的设置是一个需要认真对待的工程问题：

超时太长：上游的线程（或连接资源）被占用太久，如果下游服务不可用，上游服务的资源会被耗尽（线程池满、连接池满），进而引发级联故障（雪崩）。

超时太短：正常的慢请求被误判为超时，增加业务失败率和不必要的重试压力。

业界常用的超时设置建议：

  • 连接超时（connectionTimeout）：1000ms，连不上就快速失败；
  • 读超时（readTimeout）：根据 TP99 × 3 设置，即服务正常情况下 99% 请求在多少毫秒内返回，乘以 3 作为超时阈值；
  • 建议在压测时测量 TP99，而非经验值。

7.2 重试的语义危险

重试看似简单，实则暗藏语义陷阱。

幂等性问题：

如果一个 RPC 调用是写操作（如”扣减库存”），超时后不确定操作是否已在服务端执行。如果重试，可能导致操作被执行两次（库存被扣减两次）。

安全重试的前提条件：

  1. 操作是幂等的（多次执行与一次执行效果相同），如”查询”、“设置状态为 X”（幂等写）；
  2. 或请求 ID 是唯一的，服务端通过请求 ID 去重（幂等令牌）。

重试只应用于连接错误或超时，不应用于业务错误：

// 重试策略（只在特定异常下重试）
public Object invokeWithRetry(RpcRequest request, int maxRetry) {
    int retry = 0;
    while (retry <= maxRetry) {
        try {
            return invoke(request);
        } catch (RpcTimeoutException | ConnectException e) {
            // 可以重试：网络问题，不确定服务端是否处理了
            if (++retry > maxRetry || !isIdempotent(request)) {
                throw e;  // 非幂等操作不重试
            }
            log.warn("Retry #{} for {}", retry, request.getMethodName());
            // 换一个服务实例重试（避免重试到同一个有问题的实例）
            request.setTargetAddress(loadBalancer.selectExclude(request.getTargetAddress()));
        } catch (RpcException e) {
            // 业务异常：不重试（服务端明确返回了错误）
            throw e;
        }
    }
    throw new RpcException("Max retry exceeded");
}

7.3 熔断器：防止雪崩的最后防线

重试在下游不可用时会加剧问题（大量重试请求涌向已过载的服务）。熔断器（Circuit Breaker） 是解决方案：

熔断器有三个状态：

  • CLOSED（关闭）：正常状态，请求正常通过；
  • OPEN（打开）：熔断状态，请求直接快速失败（不发送到服务端），等待恢复；
  • HALF_OPEN（半开）：试探状态，允许少量请求通过，根据结果决定恢复或继续熔断。

状态转换规则：

  • CLOSED → OPEN：错误率超过阈值（如 50% 的请求在 10 秒内失败）；
  • OPEN → HALF_OPEN：熔断持续时间（如 30 秒）到期后自动进入半开；
  • HALF_OPEN → CLOSED：试探请求成功率高于阈值，恢复正常；
  • HALF_OPEN → OPEN：试探请求仍然高失败率，重新熔断。

总结

基于 Netty 构建 RPC 框架，是对 Netty 所有核心组件的综合运用：

  • 协议层（LengthFieldBasedFrameDecoder + 自定义编解码器）解决 TCP 粘包拆包，在协议头中携带 magic、version、msgType、serializer、requestId 等元信息；

  • 序列化层作为可插拔扩展，Protobuf 适合对性能和跨语言有要求的场景，JSON 适合调试优先的场景，Hessian2 是 Java 内部服务的良好平衡点；

  • 异步请求匹配通过 ConcurrentHashMap<requestId, CompletableFuture> + HashedWheelTimer 超时检测实现：发送请求前注册 Future，响应到达时通过 requestId 查找 Future 并完成，超时时让 Future 完成异常；

  • 连接管理需要连接池（min/max 连接数）、应用层心跳（IdleStateHandler + 自定义心跳帧）、指数退避重连三者协同，保证连接的可用性和稳定性；

  • 服务路由采用客户端负载均衡（从注册中心获取实例列表，本地算法选择），常用算法有轮询、加权轮询、最小活跃数；

  • 超时与重试必须区分幂等操作和非幂等操作，只对幂等操作做重试，且重试时应换用不同的服务实例；熔断器是防止级联故障的最后防线。

下一篇以 Dubbo、RocketMQ、Elasticsearch 三个开源项目为例，具体分析 Netty 在真实生产项目中如何被定制和使用：10 Netty在开源项目中的应用——Dubbo、RocketMQ、Elasticsearch。

参考资料

  • Dubbo 源码：org.apache.dubbo.remoting.transport.netty4
  • gRPC-Java 源码：io.grpc.netty
  • Peter Deutsch,《The Eight Fallacies of Distributed Computing》, 1991
  • Martin Fowler,《CircuitBreaker》, martinfowler.com

思考题

  1. IdleStateHandler 检测读/写空闲超时。它的实现原理是定时任务——每隔一段时间检查上次读/写时间与当前时间的差值。如果系统时钟发生跳变（如 NTP 同步导致时间回退），IdleStateHandler 会错误地触发 IDLE 事件吗？Netty 是使用 System.currentTimeMillis() 还是 System.nanoTime()？
  2. 断线重连的实现通常是在 channelInactive 事件中启动延迟重连任务。但如果网络持续不可用，重连会无限重试——消耗线程和 CPU 资源。你如何实现指数退避重连（初始 1s → 2s → 4s → … → 最大 60s）？如果重连成功后又立即断开，退避时间应该重置还是继续递增？
  3. 在双向心跳机制中，客户端发送 Ping 帧，服务端回复 Pong 帧。如果服务端进程假死（JVM Full GC 长停顿），服务端无法回复 Pong，客户端会检测到超时并断开连接。但 Full GC 结束后服务端恢复正常——此时大量客户端同时重连（reconnect storm）会导致什么问题？如何缓解？

Backlinks