marpc v6.0 — RpcContext 隐式传参 & Netty 传输

版本概述

v6.0 目标是增强跨调用链的上下文传递能力，并引入高性能的 Netty 长连接传输：

RpcContext 隐式传参：基于 ThreadLocal 的上下文透传，无需修改业务接口签名
Netty 传输：自定义二进制协议，长连接复用，可与 OkHttp 传输按配置切换

代码结构变更

marpc-core/src/main/java/com/malinghan/marpc/
├── context/                             # v6.0 新增
│   └── RpcContext.java                  # ThreadLocal 上下文容器
└── transport/
    ├── RpcTransport.java                # v6.0 新增：传输层接口
    ├── OkHttpTransport.java             # v6.0 新增：OkHttp 实现
    ├── MarpcTransport.java              # v6.0 修改：读取 context，finally 清理
    └── netty/                           # v6.0 新增
        ├── MarpcProtocol.java           # 协议常量（魔数、版本、类型）
        ├── MarpcFrame.java              # 帧对象（type, sequenceId, payload）
        ├── MarpcFrameEncoder.java       # 出站编码器
        ├── MarpcFrameDecoder.java       # 入站解码器（处理粘包）
        ├── NettyClientHandler.java      # 客户端响应处理器
        ├── NettyRpcClient.java          # Netty 传输实现
        ├── NettyServerHandler.java      # 服务端请求处理器
        └── NettyRpcServer.java          # Netty 服务端（独立端口）

修改文件：

marpc-core/src/main/java/com/malinghan/marpc/
├── core/RpcRequest.java                 # 新增 context 字段
├── consumer/MarpcInvocationHandler.java # 填充 context，替换为 RpcTransport
├── consumer/ConsumerBootstrap.java      # 接收并传递 RpcTransport
└── config/MarpcConfig.java              # 注册 RpcTransport Bean，条件装配 NettyRpcServer

核心功能

1. RpcContext 隐式传参

业务代码无需修改接口签名，通过 ThreadLocal 在调用链中透传上下文参数（如灰度标记、链路追踪 ID 等）。

RpcContext API：

// 设置/获取任意 key-value
RpcContext.set("traceId", "abc-123");
String traceId = RpcContext.get("traceId");

// 便捷方法：灰度标记
RpcContext.setGrayId("gray-user-001");
String grayId = RpcContext.getGrayId();

// 清理（框架自动调用，防止线程池复用时内存泄漏）
RpcContext.clear();

透传流程：

Consumer 业务代码
    │  RpcContext.setGrayId("gray-001")
    ▼
MarpcInvocationHandler.invoke()
    │  request.setContext(RpcContext.getAll())
    │  finally: RpcContext.clear()
    ▼
── 网络传输（context 序列化在 RpcRequest.context 字段）──
    ▼
MarpcTransport.invoke() / NettyServerHandler
    │  RpcContext.setAll(request.getContext())
    │  finally: RpcContext.clear()
    ▼
Provider 业务代码
    │  RpcContext.getGrayId()  // 读取到 "gray-001"
    ▼
  返回结果

RpcRequest 新增字段：

@Data
public class RpcRequest {
    private String service;
    private String method;
    private String methodSign;
    private Object[] args;
    private Map<String, String> context = new HashMap<>();  // v6.0 新增
}

特性：

零侵入：业务接口无需添加参数
自动清理：Consumer 和 Provider 两侧均在 finally 块清理，防止内存泄漏
线程安全：每个线程独立的 ThreadLocal 存储

2. Netty 传输

自定义二进制协议

帧格式（12 字节固定头 + 变长 body）：

+----------+----------+----------+----------+----------+----------+
| 魔数(2B) | 版本(1B) | 类型(1B) | 序列号(4B)| 长度(4B) | Body(NB) |
+----------+----------+----------+----------+----------+----------+
  0xAA 0xBB   0x01      0x01/0x02  int32      int32     JSON bytes

字段	长度	说明
魔数	2B	`0xAA 0xBB`，快速识别合法帧
版本	1B	`0x01`，预留升级空间
类型	1B	`0x01` = REQUEST，`0x02` = RESPONSE
序列号	4B	请求/响应匹配，原子递增
长度	4B	body 字节数
Body	NB	fastjson2 序列化的 RpcRequest / RpcResponse

客户端：连接池 + CompletableFuture

// NettyRpcClient 核心逻辑
public RpcResponse send(String instance, RpcRequest request) {
    Channel channel = getOrCreateChannel(instance);  // 复用长连接
    int sequenceId = sequenceIdGenerator.incrementAndGet();
    CompletableFuture<RpcResponse> future = new CompletableFuture<>();
    pendingRequests.put(sequenceId, future);

    byte[] payload = JSON.toJSONBytes(request);
    channel.writeAndFlush(new MarpcFrame(TYPE_REQUEST, sequenceId, payload));

    return future.get(timeoutMs, TimeUnit.MILLISECONDS);  // 同步等待响应
}

每个 host:port 维护一个长连接（ConcurrentHashMap<String, Channel>）
用 ConcurrentHashMap<Integer, CompletableFuture<RpcResponse>> 匹配异步响应
序列号原子递增，保证请求/响应一一对应

服务端：独立端口监听

// NettyRpcServer 启动
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
        .channel(NioServerSocketChannel.class)
        .childHandler(new ChannelInitializer<SocketChannel>() {
            protected void initChannel(SocketChannel ch) {
                ch.pipeline()
                        .addLast(new MarpcFrameDecoder())   // 解码（处理粘包）
                        .addLast(new MarpcFrameEncoder())   // 编码
                        .addLast(new NettyServerHandler(providerBootstrap));
            }
        });
bootstrap.bind(port).sync();
log.info("[NettyRpcServer] 启动，监听端口: {}", port);

独立监听 Netty 端口（默认 9090），与 HTTP 端口（8080）并存
实现 InitializingBean / DisposableBean，随 Spring 容器生命周期启停

传输层接口抽象

public interface RpcTransport {
    RpcResponse send(String instance, RpcRequest request);
}

两种实现均实现此接口，通过配置切换，MarpcInvocationHandler 无感知：

实现	协议	连接方式
`OkHttpTransport`	HTTP/JSON	短连接
`NettyRpcClient`	自定义二进制	长连接复用

配置

传输方式切换：

marpc:
  transport: okhttp    # okhttp（默认）或 netty
  netty:
    port: 9090         # Netty 服务端监听端口

完整配置示例（Netty 模式）：

marpc:
  zk:
    address: localhost:2181
  app: marpc-app
  env: dev
  transport: netty
  netty:
    port: 9090
  retry:
    maxRetries: 2
    timeout: 3000
    switchInstanceOnRetry: true
  circuitbreaker:
    enabled: false
    faultLimit: 5
    halfOpenInitialDelay: 10000
    halfOpenDelay: 5000
    windowSize: 10
  router:
    gray:
      enabled: false
      ratio: 0

Provider 和 Consumer 均需配置 marpc.transport: netty，且 Consumer 的 marpc.netty.port 需与 Provider 监听端口一致。

调用流程

Consumer 业务代码
    │  RpcContext.set(key, value)
    ▼
MarpcInvocationHandler.invoke()
    │  填充 request.context = RpcContext.getAll()
    │  preFilter → 熔断器 → 路由 → 负载均衡
    ▼
RpcTransport.send(instance, request)
    │
    ├─ OkHttpTransport ──> HTTP POST /marpc ──> MarpcTransport（Spring MVC）
    │                                               │ RpcContext.setAll(request.context)
    │                                               │ providerBootstrap.invoke(request)
    │                                               └ finally: RpcContext.clear()
    │
    └─ NettyRpcClient ──> TCP（自定义二进制）──> NettyServerHandler
                                                    │ RpcContext.setAll(request.context)
                                                    │ providerBootstrap.invoke(request)
                                                    └ finally: RpcContext.clear()
    │
    ▼
postFilter → 返回结果
    │
    └ finally: RpcContext.clear()（Consumer 侧）

Demo 测试场景

Scene10：RpcContext 隐式传参

测试目标： 验证上下文参数在 Consumer → Provider 调用链中透传成功

=== Scene10: RpcContext 隐式传参 ===
  说明：通过 RpcContext 在调用链中透传上下文参数，无需修改接口签名

  [测试1] 设置 grayId
    Consumer 设置: grayId=gray-user-001
    调用成功: hello, context-test-1
    提示：Provider 可通过 RpcContext.getGrayId() 读取到 'gray-user-001'

  [测试2] 设置多个上下文参数
    Consumer 设置: traceId=trace-abc-123, userId=user-456, grayId=gray-user-002
    调用成功: hello, context-test-2
    提示：Provider 可通过 RpcContext.get(key) 读取所有参数

  [测试3] 验证 ThreadLocal 自动清理
    第一次调用前设置 grayId=gray-user-003
    第一次调用完成
    第二次调用前不设置任何参数
    第二次调用成功: hello, context-test-4
    提示：Provider 应读取不到上次的 grayId（已自动清理）
=== Scene10 完成 ===

Provider 日志输出：

[RpcContext] method=hello, grayId=gray-user-001, traceId=null, userId=null
[RpcContext] method=hello, grayId=gray-user-002, traceId=trace-abc-123, userId=user-456
[RpcContext] method=hello, grayId=gray-user-003, traceId=null, userId=null
（第二次调用无日志输出，证明 ThreadLocal 已清理）

Scene11：Netty 传输验证

测试目标： 验证 Netty 传输的基本功能、并发正确性、与 RpcContext 联合使用

前置条件： 需配置 marpc.transport=netty（Provider 和 Consumer 均需配置）

=== Scene11: Netty 传输验证 ===
  说明：需配置 marpc.transport=netty 启用 Netty 传输
  当前传输方式由 marpc.transport 配置决定，本场景逻辑与传输方式无关

  [测试1] 基本类型调用
    hello("netty") = hello, netty
    add(10, 32) = 42
    hello("ha", 3) = ha ha ha

  [测试2] 复杂对象调用
    getUser(42) = id=42, name=user-42
    list("item-", 4) = [item-0, item-1, item-2, item-3]

  [测试3] 并发调用（10 线程，验证序列号匹配正确）
    并发结果: 成功=10, 失败=0

  [测试4] Netty 传输 + RpcContext 联合验证
    Consumer 设置: grayId=netty-gray-001, traceId=netty-trace-xyz
    调用成功: hello, netty-context
    提示：Provider 日志应输出 grayId=netty-gray-001, traceId=netty-trace-xyz
=== Scene11 完成 ===

Provider 日志输出（Netty 模式）：

[NettyRpcServer] 启动，监听端口: 9090
[RpcContext] method=hello, grayId=netty-gray-001, traceId=netty-trace-xyz, userId=null

测试流程

1. OkHttp 传输模式测试（默认）

启动 Provider：

cd marpc-demo-provider
mvn spring-boot:run

启动 Consumer：

cd marpc-demo-consumer
mvn spring-boot:run

预期输出：

Scene1-9：v5.0 已有场景正常运行
Scene10：Consumer 设置的 context 参数在 Provider 日志中输出
Scene11：使用 OkHttp 传输，所有测试通过

2. Netty 传输模式测试

修改 Provider 配置（marpc-demo-provider/src/main/resources/application.yaml）：

marpc:
  transport: netty
  netty:
    port: 9090

修改 Consumer 配置（marpc-demo-consumer/src/main/resources/application.yaml）：

marpc:
  transport: netty
  netty:
    port: 9090

启动 Provider：

cd marpc-demo-provider
mvn spring-boot:run

验证 Provider 日志：

[NettyRpcServer] 启动，监听端口: 9090

启动 Consumer：

cd marpc-demo-consumer
mvn spring-boot:run

验证 Consumer 日志：

[MarpcConfig] 使用 Netty 传输
[NettyRpcClient] 连接成功: localhost:9090

预期输出：

Scene1-9：v5.0 已有场景正常运行（传输层透明切换）
Scene10：RpcContext 在 Netty 传输下正常工作
Scene11：并发测试成功=10，失败=0（验证序列号匹配正确）

3. 传输方式对比测试

测试项	OkHttp	Netty
基本调用	✅	✅
复杂对象	✅	✅
并发调用	✅	✅
RpcContext	✅	✅
连接方式	短连接	长连接复用
协议	HTTP/JSON	自定义二进制
Provider 端口	8080 (HTTP)	9090 (Netty)

v5.0 → v6.0 升级要点

变更点	v5.0	v6.0
上下文传参	无，需修改接口签名	`RpcContext` ThreadLocal 透传
传输层	OkHttp 硬编码	`RpcTransport` 接口，OkHttp / Netty 可切换
协议	HTTP + JSON	OkHttp: HTTP/JSON；Netty: 自定义二进制
连接方式	短连接	Netty 模式下长连接复用
`RpcRequest`	无 context	新增 `Map<String, String> context`
`MarpcInvocationHandler`	持有 `OkHttpClient`	持有 `RpcTransport`，填充 context，finally 清理
`ConsumerBootstrap`	无传输层参数	接收 `RpcTransport` 并传入 Handler
`MarpcConfig`	无传输配置	`rpcTransport()` Bean，条件装配 `NettyRpcServer`
Provider 端口	仅 HTTP 8080	HTTP 8080 + 可选 Netty 9090

后续规划（v7.0）

ma-registry 集成：接入自研 HTTP 注册中心，替代 Zookeeper
监控指标：调用耗时、成功率、熔断次数统计上报
连接健康检测：Netty 心跳机制，自动重连断开的长连接
序列化优化：支持 Protobuf 等更高效的序列化方式

Navigation Menu

Repository navigation

v6.0.md

v6.0.md

marpc v6.0 — RpcContext 隐式传参 & Netty 传输

版本概述

代码结构变更

核心功能

1. RpcContext 隐式传参

2. Netty 传输

自定义二进制协议

客户端：连接池 + CompletableFuture

服务端：独立端口监听

传输层接口抽象

配置

调用流程

Demo 测试场景

Scene10：RpcContext 隐式传参

Scene11：Netty 传输验证

测试流程

1. OkHttp 传输模式测试（默认）

2. Netty 传输模式测试

3. 传输方式对比测试

v5.0 → v6.0 升级要点

后续规划（v7.0）

FilesExpand file tree

v6.0.md

Latest commit

History

v6.0.md

File metadata and controls

marpc v6.0 — RpcContext 隐式传参 & Netty 传输

版本概述

代码结构变更

核心功能

1. RpcContext 隐式传参

2. Netty 传输

自定义二进制协议

客户端：连接池 + CompletableFuture

服务端：独立端口监听

传输层接口抽象

配置

调用流程

Demo 测试场景

Scene10：RpcContext 隐式传参

Scene11：Netty 传输验证

测试流程

1. OkHttp 传输模式测试（默认）

2. Netty 传输模式测试

3. 传输方式对比测试

v5.0 → v6.0 升级要点

后续规划（v7.0）