180｜RPC 原理：代理、序列化、协议、网络传输、超时和异常映射

调用方（Consumer）                         服务方（Provider）
┌──────────────┐                          ┌──────────────┐
│  Business    │                          │  Business    │
│  Logic       │                          │  Logic       │
├──────────────┤                          ├──────────────┤
│  Stub/Proxy  │  ─── 网络传输 ───→       │  Skeleton    │
├──────────────┤                          ├──────────────┤
│ 序列化/反序列化│                         │ 序列化/反序列化│
└──────────────┘                          └──────────────┘

// 调用方看到的代码——你以为在调本地方法
UserService userService = rpcClient.create(UserService.class);
User user = userService.getUserById(10086L);

// 实际上 userService 是一个代理对象
// 它的 invoke 方法里做了：
// 1. 构造 RpcRequest(methodName="getUserById", args=[10086L], ...)
// 2. 序列化请求
// 3. 发送网络请求
// 4. 等待响应
// 5. 反序列化结果
// 6. 返回给调用方

public class RpcProxy implements InvocationHandler {
    private final Class<?> serviceInterface;
    private final String remoteHost;
    private final int remotePort;

    @Override

    public Object invoke(Object proxy, Method method, Object[ ] args) throws Throwable {

        // 1. 构造请求
        RpcRequest request = new RpcRequest();
        request.setInterfaceName(serviceInterface.getName());
        request.setMethodName(method.getName());
        request.setParameterTypes(method.getParameterTypes());
        request.setParameters(args);

        // 2. 网络调用
        RpcResponse response = networkClient.send(request);

        // 3. 返回结果
        if (response.getException() != null) {
            throw response.getException();
        }
        return response.getResult();
    }

    @SuppressWarnings("unchecked")
    public static <T> T create(Class<T> interfaceClass, String host, int port) {
        return (T) Proxy.newProxyInstance(
            interfaceClass.getClassLoader(),

            new Class[ ]{interfaceClass},

            new RpcProxy(interfaceClass, host, port)
        );
    }
}

// CGLIB 代理示例
Enhancer enhancer = new Enhancer();
enhancer.setSuperclass(TargetClass.class);
enhancer.setCallback((MethodInterceptor) (obj, method, args, proxy) -> {
    // 前置处理：构造 RPC 请求
    RpcRequest request = buildRequest(method, args);
    RpcResponse response = sendRequest(request);
    return response.getResult();
});
TargetClass proxy = (TargetClass) enhancer.create();

// Dubbo 中获取代理的入口
// ProxyFactory 的 SPI 扩展点，默认 JavassistProxyFactory
public interface ProxyFactory {
    <T> T getProxy(Invoker<T> invoker) throws RpcException;
    <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) throws RpcException;
}

调用方 Java 对象 → [序列化] → 字节流 → [网络传输] → 字节流 → [反序列化] → 服务方 Java 对象

// 序列化
User user = new User("张三", 25);
String json = JSON.toJSONString(user);
// → {"name":"张三","age":25}

// 反序列化
User user2 = JSON.parseObject(json, User.class);

// 定义 .proto 文件
message User {
  int64 id = 1;
  string name = 2;
  int32 age = 3;
}

// Dubbo 默认序列化协议
Hessian2Output out = new Hessian2Output(outputStream);
out.writeObject(user);
out.flush();

Kryo kryo = new Kryo();
Output output = new Output(new ByteArrayOutputStream());
kryo.writeObject(output, user);

需要跨语言？ ─┬─ 是 → 选择 Protobuf（性能优先）或 JSON（可读性优先）
              │
              └─ 否（纯 Java 生态）→
                  ├─ 极致性能 → Kryo
                  ├─ 平衡性能与兼容性 → Hessian2
                  └─ 调试/HTTP 接口 → JSON

┌──────────────┬──────────────┬──────────────┬──────────────┬──────────────┐
│  Magic Number│  Header      │  Body Length │  Header Data │  Body Data   │
│  (2 bytes)   │  (1 byte)    │  (4 bytes)   │  (variable)  │  (variable)  │
└──────────────┴──────────────┴──────────────┴──────────────┴──────────────┘

Dubbo 协议帧（共 16 字节固定头部 + 可变部分）
Byte 0-1:  Magic Number (0xdabb)
Byte 2:    Flag (序列化 ID + 消息类型)
Byte 3:    Status (响应状态码)
Byte 4-11: Request ID (8 字节，long 类型)
Byte 12-15: Body Length (4 字节，int 类型)

= 变长部分
Data: 序列化后的 Invocation 数据（接口名、方法名、参数类型、参数值）

// gRPC 的服务定义
service UserService {
  // 一元 RPC
  rpc GetUser (GetUserRequest) returns (GetUserResponse);

  // 服务端流式 RPC
  rpc ListUsers (ListUsersRequest) returns (stream User);

  // 客户端流式 RPC
  rpc BatchCreateUsers (stream CreateUserRequest) returns (BatchCreateUserResponse);

  // 双向流式 RPC
  rpc Chat (stream ChatMessage) returns (stream ChatMessage);
}

┌──────────────────────────────────────┐
│            RPC 框架层                │
│  (Dubbo / gRPC / Thrift)            │
├──────────────────────────────────────┤
│            协议层                    │
│  (Dubbo Protocol / HTTP/2)          │
├──────────────────────────────────────┤
│          网络传输层                  │
│  (Netty / Mina / JDK NIO)           │
├──────────────────────────────────────┤
│          操作系统                    │
│  (TCP/IP Stack / epoll)             │
└──────────────────────────────────────┘

                    ┌─────────────┐
                    │  Boss Group  │  ← 接受连接，注册到 Worker
                    │ (1-2 threads)│
                    └──────┬───────┘
                           │
           ┌───────────────┼───────────────┐
           │               │               │
    ┌──────▼──────┐ ┌──────▼──────┐ ┌──────▼──────┐
    │ Worker      │ │ Worker      │ │ Worker      │  ← 处理 IO 读写
    │ Thread      │ │ Thread      │ │ Thread      │
    └──────┬──────┘ └──────┬──────┘ └──────┬──────┘
           │               │               │
    ┌──────▼──────┐ ┌──────▼──────┐ ┌──────▼──────┐
    │ Channel     │ │ Channel     │ │ Channel     │  ← 每个连接绑定一个线程
    │ Pipeline    │ │ Pipeline    │ │ Pipeline    │
    └─────────────┘ └─────────────┘ └─────────────┘

// Netty 的 Handler 链
pipeline.addLast("decoder", new RpcDecoder());      // 解码
pipeline.addLast("encoder", new RpcEncoder());      // 编码
pipeline.addLast("idleHandler", new IdleStateHandler(60, 0, 0)); // 心跳
pipeline.addLast("handler", new RpcServerHandler()); // 业务处理

单连接复用（Dubbo 默认）：
Consumer ──[单 TCP 连接]──> Provider
  ├─ Request#1 (id=1001)
  ├─ Request#2 (id=1002)
  └─ Request#3 (id=1003)
  → 通过 Request ID 匹配请求和响应

连接池（gRPC 推荐）：
Consumer ──[连接池]──> Provider
  ├─ Connection#1 → Request#1
  ├─ Connection#2 → Request#2
  └─ Connection#3 → Request#3
  → 每个连接串行，多连接并行

Provider 端：IdleStateHandler 检测读空闲
  → 60s 未收到数据 → 关闭连接

Consumer 端：定时发送心跳（通常 30s 间隔）
  → 心跳包：协议头的 Flag 标记为心跳类型
  → 无 Body，仅消耗极少的网络带宽

// 错误做法：在 Netty 的 I/O 线程中执行业务逻辑
// 会阻塞 I/O 线程，导致其他连接无法读写
public class BadHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        // 这里直接调用业务方法——大忌！
        Object result = userServiceImpl.getUserById(request.getId());
        ctx.writeAndFlush(result);
    }
}

// 正确做法：将业务处理提交到业务线程池
public class GoodHandler extends ChannelInboundHandlerAdapter {
    private final ExecutorService bizThreadPool;

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        bizThreadPool.execute(() -> {
            Object result = userServiceImpl.getUserById(request.getId());
            ctx.writeAndFlush(result);
        });
    }
}

没有超时：
  客户端线程 → 发起 RPC 调用 → 服务端挂起/死锁 → 客户端线程永久阻塞
  → 线程池耗尽 → 整个服务不可用 → 级联故障

有超时：
  客户端线程 → 发起 RPC 调用 → 超时(1s) → 抛出 TimeoutException
  → 线程释放 → 降级/重试 → 服务部分降级但可用

层级 1: 客户端调用超时
  userService.getUserById(id, timeout=1000ms)
  → 整个调用链的总超时，包括网络、序列化、服务端处理

层级 2: 连接超时
  → 建立 TCP 连接的超时，通常 100-500ms

层级 3: 读写超时
  → Socket read/write 的超时，避免半开连接永久阻塞

层级 4: 服务端处理超时
  → 服务端执行业务逻辑的超时，防止慢请求拖垮线程池

// Netty 的超时基于 HashedWheelTimer（时间轮算法）
// 发送请求时注册超时任务
public class RpcClient {
    private final HashedWheelTimer timer = new HashedWheelTimer();

    public RpcResponse send(RpcRequest request, long timeoutMs) {
        RpcFuture future = new RpcFuture(request.getRequestId());
        FUTURES.put(request.getRequestId(), future);

        // 注册超时任务
        Timeout timeout = timer.newTimeout(t -> {
            RpcFuture f = FUTURES.remove(request.getRequestId());
            if (f != null) {
                f.setException(new TimeoutException("Request timeout after " + timeoutMs + "ms"));
            }
        }, timeoutMs, TimeUnit.MILLISECONDS);

        // 发送请求
        channel.writeAndFlush(request);

        // 阻塞等待响应或超时
        try {
            return future.get(timeoutMs, TimeUnit.MILLISECONDS);
        } finally {
            timeout.cancel(); // 响应到达后取消超时任务
        }
    }
}

Client (timeout=3s)
  → Service A (timeout=3s - elapsed)
    → Service B (timeout=2.5s - elapsed)
      → Service C (timeout=2s - elapsed)
        → Service D (timeout=1.5s - elapsed)

// gRPC 的 Deadline 传播
// 调用方设置 3 秒超时
UserServiceGrpc.UserServiceBlockingStub stub =
    UserServiceGrpc.newBlockingStub(channel)
        .withDeadlineAfter(3, TimeUnit.SECONDS);

// 服务端收到的请求中携带了 deadline
// 如果服务端调用下游，会自动减去已用时间
public void getUser(GetUserRequest req, StreamObserver<User> resp) {
    // 获取剩余时间
    long remainingNanos = Context.current()
        .getDeadline().timeRemaining(TimeUnit.NANOSECONDS);
    // 使用剩余时间调用下游
    downstreamStub.withDeadlineAfter(remainingNanos, TimeUnit.NANOSECONDS)
        .getRelatedData(request);
}

// 超时重试的风险：服务端可能已经处理成功
// 只是响应超时，重试会导致重复执行

// 解决方案：幂等性设计
// 1. 请求携带幂等键（Idempotency-Key）
// 2. 服务端根据幂等键去重
// 3. 重试时使用相同的幂等键

// 示例：Dubbo 的幂等重试
public Result invoke(Invoker<?> invoker, Invocation invocation) {
    // 请求中携带唯一 ID
    String idempotentKey = invocation.getAttachment("idempotent-key");
    if (idempotentKey != null) {
        // 检查是否已处理过
        if (idempotentCache.contains(idempotentKey)) {
            return idempotentCache.get(idempotentKey);
        }
    }
    Result result = invoker.invoke(invocation);
    if (idempotentKey != null) {
        idempotentCache.put(idempotentKey, result);
    }
    return result;
}

┌────────────────────────────────────────────┐
│              业务异常                       │
│  (UserNotFoundException,                   │
│   InsufficientBalanceException, ...)        │
│  → 服务端业务逻辑抛出的异常                  │
│  → 应该透传给调用方，调用方需要知道业务含义   │
├────────────────────────────────────────────┤
│              RPC 框架异常                   │
│  (RemotingException, TimeoutException,      │
│   SerializationException, ...)              │
│  → 框架层面的异常，与业务无关                │
│  → 通常触发重试或降级                       │
├────────────────────────────────────────────┤
│              运行时异常                     │
│  (NullPointerException,                     │
│   ClassCastException, ...)                  │
│  → 服务端代码 Bug，理论上不应该出现          │
│  → 需要告警和修复                           │
└────────────────────────────────────────────┘

// Dubbo 的 ExceptionFilter 处理异常
// 1. 如果是受检异常且在接口方法签名中声明 → 直接抛出
// 2. 如果是 RuntimeException → 直接抛出
// 3. 如果是接口签名中声明的受检异常 → 直接抛出
// 4. 其他异常 → 包装为 RuntimeException 抛出

// 异常序列化
// 异常类名 + 消息 + 堆栈 = 序列化传输
// 调用方反序列化时，如果本地没有对应的异常类
// → 使用 DecodeableRpcResult 中的通用异常包装

// gRPC 服务端
public void getUser(GetUserRequest req, StreamObserver<User> resp) {
    try {
        User user = userService.getUser(req.getId());
        if (user == null) {
            resp.onError(Status.NOT_FOUND
                .withDescription("User not found: " + req.getId())
                .asRuntimeException());
            return;
        }
        resp.onNext(user);
        resp.onCompleted();
    } catch (Exception e) {
        resp.onError(Status.INTERNAL
            .withDescription(e.getMessage())
            .withCause(e)
            .asRuntimeException());
    }
}

// 定义业务异常基类
public class BusinessException extends RuntimeException {
    private final int errorCode;
    private final String errorMessage;

    public BusinessException(int errorCode, String errorMessage) {
        super(errorMessage);
        this.errorCode = errorCode;
        this.errorMessage = errorMessage;
    }
}

// 具体业务异常
public class UserNotFoundException extends BusinessException {
    public UserNotFoundException(long userId) {
        super(10001, "用户不存在: " + userId);
    }
}

// 在 RPC 异常过滤器中处理
public class RpcExceptionFilter {
    public Result handleException(Throwable t) {
        if (t instanceof BusinessException) {
            BusinessException be = (BusinessException) t;
            return Result.error(be.getErrorCode(), be.getErrorMessage());
        }
        if (t instanceof TimeoutException) {
            return Result.error(50001, "调用超时，请稍后重试");
        }
        // 未预期的异常：返回通用错误码，记录日志，触发告警
        log.error("Unexpected exception in RPC call", t);
        alertingService.send("RPC异常", t.getMessage());
        return Result.error(50000, "系统繁忙，请稍后重试");
    }
}

步骤 1: 代理拦截
  userService.getUserById(10086L)
  → RpcProxy.invoke() 被调用

步骤 2: 构造请求
  → RpcInvocation {
      interfaceName: "com.example.UserService",
      methodName: "getUserById",
      parameterTypes: [long.class],
      arguments: [10086L],
      attachments: {traceId: "abc123", timeout: "3000"}
    }

步骤 3: 服务发现与路由
  → 从注册中心获取 Provider 列表
  → 根据路由规则过滤
  → 负载均衡选择一个节点

步骤 4: 序列化
  → RpcInvocation 对象 → 字节流（Hessian2/Protobuf/...）

步骤 5: 协议封装
  → 加上协议头（Magic Number, Request ID, Body Length, ...）

步骤 6: 网络发送
  → Netty channel.writeAndFlush()
  → 注册超时任务（3s）
  → 注册 Future 等待响应

步骤 7: 服务端接收
  → Netty Worker 线程读取数据
  → 协议解码 → 反序列化 → 得到 RpcInvocation

步骤 8: 服务端业务处理
  → 提交到业务线程池
  → 反射调用 UserServiceImpl.getUserById(10086L)
  → 返回 User 对象

步骤 9: 服务端响应
  → 序列化 User 对象 → 协议封装 → 网络发送

步骤 10: 客户端接收响应
  → 反序列化 → 填充 Future → 返回给调用方