Jotaro动手教你实现rpc

调用方只写这一行：

User user = userService.getUserById(1);

userService 是接口，实现在另一个进程里。JRpc 把这次方法调用收成一帧二进制，用 Netty 送过去，对端按接口名找到实现、过令牌桶、反射执行，再把 User 送回来。连接用完即关。

仓库在 jotaro-rpc，版本 1.0.6-SNAPSHOT。下面按动手的顺序写：报文、代理、协议、注册、发现、服务端执行。每段先讲这段代码在链路里干什么，再给出仓库里的实现。

sequenceDiagram participant App as 业务代码 participant Proxy as NettyClientProxy participant Disc as ZKServiceCenter participant Cli as NettyRpcClient participant Srv as NettyRpcServerHandler participant Impl as 接口实现 App->>Proxy: userService.getUserById(1) Proxy->>Proxy: 组装 RpcRequest Proxy->>Cli: sendRequest Cli->>Disc: serviceDiscovery(接口全名) Disc-->>Cli: InetSocketAddress Cli->>Srv: connect + writeAndFlush Srv->>Srv: 查实现表，取令牌 Srv->>Impl: Method.invoke Impl-->>Srv: 返回值 Srv-->>Cli: RpcResponse，写出后关闭连接 Cli-->>Proxy: channel 属性 rpcResponse Proxy-->>App: response.getData()

四个模块够用。common 放报文、编解码和限流，register 管 ZooKeeper，server 负责监听和反射，client 负责代理和连接。

flowchart LR subgraph clientMod["client"] Proxy["NettyClientProxy"] RpcCli["NettyRpcClient"] end subgraph registerMod["register"] Reg["ZKServiceRegister"] Center["ZKServiceCenter"] end subgraph serverMod["server"] Provider["ServerProvider"] RpcSrv["NettyRpcServer"] end subgraph commonMod["common"] Proto["RpcRequest / RpcResponse"] Codec["JotaroEncoder / JotaroDecoder"] end Proxy --> RpcCli RpcCli --> Center RpcCli --> Codec Provider --> Reg RpcSrv --> Provider RpcSrv --> Codec classDef c fill:#dbeafe,stroke:#2563eb,color:#1e3a8a classDef r fill:#ffedd5,stroke:#ea580c,color:#7c2d12 classDef s fill:#dcfce7,stroke:#16a34a,color:#14532d classDef m fill:#ede9fe,stroke:#7c3aed,color:#4c1d95 class Proxy,RpcCli c class Reg,Center r class Provider,RpcSrv s class Proto,Codec m

目录

1. 一次调用长什么样
2. 动态代理把接口变成请求
3. 自己写一帧协议
4. 挂上 Netty Pipeline
5. JSON 过网后把类型捞回来
6. 实现类挂进 ZooKeeper
7. 找到机器，再选一台
8. 服务端先限流，再反射
9. 从这条连接上拿回结果
10. 跑通 UserService

1. 一次调用长什么样

远程调用要能在对端还原成一次 Java 方法调用，所以请求里带齐四样东西：哪个接口、哪个方法、参数值、参数类型。类型必须单独带，JSON 过网之后数字和对象的运行时类会对不上方法签名。

@Data
@Builder
public class RpcRequest implements Serializable {
    private String interfaceName;
    private String methodName;
    private Object[] params;
    private Class<?>[] paramsType;
}

回来的包同样简单。code = 0 是成功，code = -1 是失败，原因写在 message。dataType 留给客户端把 JSON 变回原来的返回类型。

@Data
@Builder
public class RpcResponse implements Serializable {
    private Integer code;
    private String message;
    private Object data;
    private Class<?> dataType;

    public static RpcResponse success(Object data, Class<?> dataType) {
        return RpcResponse.builder()
                .code(RespCodeEnum.SUCCESS.getCode())
                .data(data)
                .dataType(dataType)
                .build();
    }

    public static RpcResponse fail(String message) {
        return RpcResponse.builder()
                .code(RespCodeEnum.UNKNOWN_ERROR.getCode())
                .message(message)
                .build();
    }
}

RespCodeEnum 里成功是 0，未知错误是 -1。

2. 动态代理把接口变成请求

业务代码拿不到实现类。getProxy 用 JDK 动态代理生成一个实现，方法调用全部进 invoke。这里把 Method 拆开，填进上一节的 RpcRequest，然后交给 NettyRpcClient。

public <T> T getProxy(Class<T> clazz) {
    Object obj = Proxy.newProxyInstance(clazz.getClassLoader(), new Class[]{clazz}, this);
    return (T) obj;
}

@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
    RpcRequest rpcRequest = RpcRequest.builder()
            .interfaceName(method.getDeclaringClass().getName())
            .methodName(method.getName())
            .params(args)
            .paramsType(method.getParameterTypes())
            .build();

    RpcResponse rpcResponse = this.ioClient.sendRequest(rpcRequest);
    if (rpcResponse == null) {
        throw new IllegalStateException("rpc response is null, request=" + method.getName());
    }
    return rpcResponse.getData();
}

代理只把 data 交回去。响应对象本身是 null，抛 IllegalStateException。限流、找不到服务、发送失败走的是 RpcResponse.fail，data 为空，调用方拿到 null。

3. 自己写一帧协议

TCP 是字节流，两条消息粘在一起时对端不知道从哪切开。JRpc 用两层长度。外层 4 字节由 Netty 的 LengthFieldPrepender 负责，解决粘包。内层是编码器自己写的业务头。

flowchart LR A["frameLen
4 字节
Prepender 添加"] --> B["msgType
2 字节
0 请求 / 1 响应"] B --> C["serialize
2 字节
0 Java / 1 JSON"] C --> D["payloadLen
4 字节
Encoder 写入"] D --> E["payload
N 字节"] classDef len fill:#ffedd5,stroke:#ea580c,color:#7c2d12 classDef head fill:#dbeafe,stroke:#2563eb,color:#1e3a8a classDef body fill:#ede9fe,stroke:#7c3aed,color:#4c1d95 class A,D len class B,C head class E body

JotaroEncoder 只写内层。请求写 0，响应写 1，接着写序列化类型、包体长度和包体。当前 pipeline 塞进去的是 JsonSerializer，类型值是 1。

@Override
protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception {
    if (msg instanceof RpcRequest) {
        out.writeShort(MsgTypeEnum.REQ.getCode());
    } else if (msg instanceof RpcResponse) {
        out.writeShort(MsgTypeEnum.RESP.getCode());
    }
    out.writeShort(serializer.getSerializeType());
    byte[] bytes = serializer.serialize(msg);
    out.writeInt(bytes.length);
    out.writeBytes(bytes);
}

解码时外层 4 字节已经被 LengthFieldBasedFrameDecoder 剥掉，缓冲区从 msgType 开始。按类型找到序列化器，读出包体，还原成 RpcRequest 或 RpcResponse。

@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
    short msgTypeCode = in.readShort();
    MsgTypeEnum msgType = MsgTypeEnum.getTypeById((int) msgTypeCode);
    if (Objects.isNull(msgType)) {
        log.error("[decoder] not support msg type, msgTypeCode:{}", msgTypeCode);
        return;
    }
    short serializerType = in.readShort();
    Serializer serializer = Type2SSerializerEnum.getSerializerByType(serializerType);
    if (Objects.isNull(serializer)) {
        log.error("[decoder] not support serializer, type:{}", serializerType);
        return;
    }
    int msgLength = in.readInt();
    byte[] msg = new byte[msgLength];
    in.readBytes(msg);
    out.add(serializer.deserialize(msg, msgTypeCode));
}

Type2SSerializerEnum 把帧头里的数字映射到类：0 是 ObjectSerializer，1 是 JsonSerializer，用 newInstance() 创建。两端现在都发送 JSON。对端把类型写成 0 时，这段解码会走 Java 原生序列化。

4. 挂上 Netty Pipeline

客户端和服务端的 pipeline 一样，最后一个 handler 不同。addLast 的顺序决定加工顺序：入站从头到尾，出站从尾到头。所以写出时先经过 JotaroEncoder，再由 LengthFieldPrepender 在最前面补 4 字节帧长。

@Override
protected void initChannel(SocketChannel ch) {
    ChannelPipeline pipeline = ch.pipeline();
    pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
    pipeline.addLast(new LengthFieldPrepender(4));
    pipeline.addLast(new JotaroEncoder(new JsonSerializer()));
    pipeline.addLast(new JotaroDecoder());
    pipeline.addLast(new NettyRpcServerHandler(serverProvider));
}

LengthFieldBasedFrameDecoder 的五个参数：最大帧长、长度字段偏移、长度字段占 4 字节、长度调整值 0、解码后跳过 4 字节。跳过之后，JotaroDecoder 读到的第一个 short 就是消息类型。

flowchart TB subgraph outbound["出站 writeAndFlush"] direction LR Enc["JotaroEncoder
类型 + 序列化方式 + 包体长度 + 包体"] Pre["LengthFieldPrepender
前面补 4 字节帧长"] Enc --> Pre end subgraph inbound["入站 channelRead"] direction LR Frame["LengthFieldBasedFrameDecoder
按帧长切开并丢掉帧长"] Dec["JotaroDecoder
还原 RpcRequest 或 RpcResponse"] H["业务 Handler"] Frame --> Dec --> H end Pre -.-> Frame classDef out fill:#dbeafe,stroke:#2563eb,color:#1e3a8a classDef in fill:#dcfce7,stroke:#16a34a,color:#14532d class Enc,Pre out class Frame,Dec,H in

服务端在这里启动。boss 接连接，worker 读包。bind 只绑端口，写进 ZooKeeper 的 host 是构造 ServerProvider 时传入的那个地址。

@Override
public void start(int port) {
    NioEventLoopGroup bossGroup = new NioEventLoopGroup();
    NioEventLoopGroup workGroup = new NioEventLoopGroup();
    ServerBootstrap serverBootstrap = new ServerBootstrap();
    serverBootstrap.group(bossGroup, workGroup)
            .channel(NioServerSocketChannel.class)
            .childHandler(new NettyServerInitializer(serverProvider));
    ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
    channelFuture.channel().closeFuture().sync();
}

5. JSON 过网后把类型捞回来

JsonSerializer 用 Fastjson2。反序列化打开 SupportClassForName，请求里的 Class[] 和响应里的 dataType 才能变回类。

自定义对象从 JSON 回来经常是 JSONObject，直接拿去 Method.invoke 会对不上参数类型。所以每个参数先做一次兼容检查，对不上就按声明类型再转一次。

private Object parseReqObject(byte[] bytes) {
    RpcRequest rpcRequest = JSON.parseObject(bytes, RpcRequest.class, JSONReader.Feature.SupportClassForName);
    Object[] objs = new Object[rpcRequest.getParamsType().length];
    for (int i = 0; i < objs.length; i++) {
        Class<?> paramType = rpcRequest.getParamsType()[i];
        if (!paramType.isAssignableFrom(rpcRequest.getParams()[i].getClass())) {
            objs[i] = JSON.to(paramType, rpcRequest.getParams()[i]);
        } else {
            objs[i] = rpcRequest.getParams()[i];
        }
    }
    rpcRequest.setParams(objs);
    return rpcRequest;
}

返回值同样处理。服务端成功时把 method.getReturnType() 放进 dataType，客户端解码时如果 data 的类对不上，再 JSON.to(dataType, data)。User 能变回 User，靠的就是这个字段。

Java 原生序列化在 ObjectSerializer 里，用 ObjectOutputStream 写、ObjectInputStream 读，类型留在字节流里。RpcRequest、RpcResponse、User 都实现了 Serializable。当前 pipeline 发送的是 JSON。

6. 实现类挂进 ZooKeeper

服务端做两件事：本地留一张「接口名 → 实现」的表，供反射查找；ZooKeeper 上留一个临时节点，供客户端发现。

public void provideServiceInterface(Object service) {
    Class<?>[] interfaceNames = service.getClass().getInterfaces();
    InetSocketAddress inetSocketAddress = new InetSocketAddress(this.connAddress.getHost(), this.connAddress.getPort());
    for (Class clazz : interfaceNames) {
        this.interfaceProvider.put(clazz.getName(), service);
        this.servicerRegister.serviceRegister(clazz.getName(), inetSocketAddress);
    }
}

一个实现类有几个接口，就注册几条服务。

ZooKeeper 分两级。服务路径是永久节点，进程退出后还在，避免服务名本身消失。实例是临时节点，会话断了 ZooKeeper 会把它删掉。地址写在路径最后一段，节点 data 是空的。

String servicePath = ZKUtils.getZkPathByServiceName(serviceName);
if (Objects.isNull(this.zkClient.checkExists().forPath(servicePath))) {
    this.zkClient.create()
            .creatingParentsIfNeeded()
            .withMode(CreateMode.PERSISTENT)
            .forPath(servicePath);
}
this.zkClient.create()
        .creatingParentsIfNeeded()
        .withMode(CreateMode.EPHEMERAL)
        .forPath(String.join("/", servicePath, ZKUtils.getServiceInetAddress(address)));

Curator 客户端带了命名空间 JotaroRpc，所以磁盘上的路径是：

flowchart TB NS["命名空间 JotaroRpc"] Svc["/com.jotaro.api.UserService
PERSISTENT，服务下线后还在"] A["/127.0.0.1:9999
EPHEMERAL"] B["/127.0.0.1:10000
EPHEMERAL"] NS --> Svc Svc --> A Svc --> B classDef ns fill:#ede9fe,stroke:#7c3aed,color:#4c1d95 classDef persistent fill:#dbeafe,stroke:#2563eb,color:#1e3a8a classDef ephemeral fill:#ffedd5,stroke:#ea580c,color:#7c2d12 class NS ns class Svc persistent class A,B ephemeral

连接参数在 ZKClientFactory：会话 40 秒，连接 15 秒，ExponentialBackoffRetry(1000, 3)。地址从 classpath 的 zk.propertis 读 jrpc.zk.address，文件名就是这个拼写。读不到就用 localhost:2181。

7. 找到机器，再选一台

客户端每次调用都问一次 ZKServiceCenter。先看本地 Map，空了再 getChildren。子节点名就是注册时写的 host:port，交给负载均衡挑一个，再拆成 InetSocketAddress。

@Override
public InetSocketAddress serviceDiscovery(String serviceName) {
    List<String> serviceNodesStr = cacheService.getServiceFromCache(serviceName);
    if (CollectionUtil.isEmpty(serviceNodesStr)) {
        serviceNodesStr = this.zkClient.getChildren().forPath(ZKUtils.getZkPathByServiceName(serviceName));
    }
    if (CollectionUtil.isEmpty(serviceNodesStr)) {
        return null;
    }
    return parseAddress(this.loadBalancer.balance(serviceNodesStr));
}

flowchart TD Start["serviceDiscovery(接口全名)"] Cache{"本地 Map 里有地址?"} ZK["getChildren /接口全名"] Empty{"子节点为空?"} Null["返回 null"] Pick["RoundLoadBalancer"] Addr["拆成 InetSocketAddress"] Start --> Cache Cache -->|有| Pick Cache -->|没有| ZK ZK --> Empty Empty -->|是| Null Empty -->|否| Pick Pick --> Addr classDef step fill:#dbeafe,stroke:#2563eb,color:#1e3a8a classDef decision fill:#fef9c3,stroke:#ca8a04,color:#713f12 classDef bad fill:#fee2e2,stroke:#dc2626,color:#7f1d1d classDef ok fill:#dcfce7,stroke:#16a34a,color:#14532d class Start,ZK,Pick step class Cache,Empty decision class Null bad class Addr ok

本地 Map 的写入在 ZKWatcher。它监听命名空间根，本意是节点创建就放进缓存。路径解析用的是节点 data：

private String[] parsePath(ChildData childData) {
    String path = new String(childData.getData());
    return path.split("/");
}

注册临时节点时没有写入 data，切出来的段数到不了「服务名 + 地址」。缓存因此一直是空的，serviceDiscovery 每次走到 getChildren。真正选出地址的是下面这段。

默认均衡是轮询。choose 从 -1 开始，第一次取到下标 0。

public String balance(List<String> addressList) {
    choose++;
    choose = choose % addressList.size();
    return addressList.get(choose);
}

随机均衡是 new Random().nextInt(addressList.size())。换实现只改 ZKServiceCenter 构造里的 this.loadBalancer。

一致性哈希给每个真实地址挂 10 个虚拟节点，名字是 地址#序号，放进 TreeMap。查找时对一个随机整数做哈希，顺着环取下一个节点，到尾部就回到第一个，再截掉 # 后面的序号。

String randomNumStr = String.valueOf(RandomUtil.randomInt());
String address = getServer(getHash.apply(randomNumStr), addressList);

private String getServer(Integer nodeHashNum, List<String> addressList) {
    if (MapUtil.isEmpty(HASH_2_VIRTUAL_NODES)) {
        init(addressList);
    } else {
        refresh(addressList);
    }
    TreeMap<Integer, String> subMap = (TreeMap<Integer, String>) HASH_2_VIRTUAL_NODES.tailMap(nodeHashNum);
    Integer key = subMap.isEmpty() ? HASH_2_VIRTUAL_NODES.firstKey() : subMap.firstKey();
    String selectedAddress = HASH_2_VIRTUAL_NODES.get(key);
    return selectedAddress.substring(0, selectedAddress.indexOf("#"));
}

哈希输入是随机数，同一次调用的参数不会把流量粘在固定节点上。10 个虚拟节点把随机点摊到各真实地址。构造方法必须传入 Function<String, Integer>，它会覆盖类上那份 FNV-1a 初值。

8. 服务端先限流，再反射

包到了 NettyRpcServerHandler。先按接口名查本地表，再按方法名和参数类型找到 Method，取到令牌才 invoke。写完响应立刻关掉连接，关闭放在写出完成之后，避免对端还没读到字节连接已经断了。

@Override
protected void channelRead0(ChannelHandlerContext ctx, RpcRequest rpcRequest) {
    RpcResponse rpcResponse = getResponse(rpcRequest);
    ctx.writeAndFlush(rpcResponse).addListener(ChannelFutureListener.CLOSE);
}

private RpcResponse getResponse(RpcRequest rpcRequest) {
    Object service = serverProvider.getService(rpcRequest.getInterfaceName());
    if (service == null) {
        return RpcResponse.fail("服务未找到: " + rpcRequest.getInterfaceName());
    }
    try {
        Method method = service.getClass().getMethod(rpcRequest.getMethodName(), rpcRequest.getParamsType());
        RateLimiter rateLimiter = serverProvider.getRateLimiter()
                .getRateLimit(rpcRequest.getInterfaceName(), method.getName());
        if (!rateLimiter.getToken()) {
            return RpcResponse.fail("服务限流");
        }
        Object invoke = method.invoke(service, rpcRequest.getParams());
        return RpcResponse.success(invoke, method.getReturnType());
    } catch (Exception e) {
        return RpcResponse.fail("服务端异常");
    }
}

令牌桶按 接口名#方法名 懒创建，参数是 (补充间隔毫秒, 容量)，也就是 100 毫秒、容量 10。一开始桶是满的，连续 10 次直接减库存放行。

@Override
public boolean getToken() {
    if (curCapacity > 0) {
        curCapacity--;
        return true;
    }
    long current = System.currentTimeMillis();
    if (current - lastTimeStamp >= RATE) {
        if ((current - lastTimeStamp) / RATE >= 2) {
            curCapacity += (int) ((current - lastTimeStamp) / RATE) - 1;
        }
        if (curCapacity > CAPACITY) {
            curCapacity = CAPACITY;
        }
        lastTimeStamp = current;
        return true;
    }
    return false;
}

flowchart TD Req["getToken"] Has{"curCapacity > 0 ?"} Dec["减 1，放行"] Wait{"距离上次放行 >= 100ms ?"} Gap{"间隔数 >= 2 ?"} Add["补上 间隔数减 1，封顶 10"] Pass["本次放行，不扣刚补进的令牌"] Deny["返回 false，文案是服务限流"] Req --> Has Has -->|是| Dec Has -->|否| Wait Wait -->|否| Deny Wait -->|是| Gap Gap -->|是| Add --> Pass Gap -->|否| Pass classDef ok fill:#dcfce7,stroke:#16a34a,color:#14532d classDef no fill:#fee2e2,stroke:#dc2626,color:#7f1d1d classDef q fill:#fef9c3,stroke:#ca8a04,color:#713f12 class Dec,Add,Pass ok class Deny no class Has,Wait,Gap q

库存到 0 之后，不满 100 毫秒就拒绝。满 1 个间隔，本次放行，库存仍是 0。满 2 个及以上间隔，把「间隔数 − 1」补进桶里并封顶 10，本次另外放行。getToken 没有加锁，同一方法上的并发请求会一起改 curCapacity。

9. 从这条连接上拿回结果

客户端的 Bootstrap 和 EventLoopGroup 是静态的，channel 每次新建。连上、写出、等到连接关闭，再从 channel 的 attribute 里取响应。一次调用一条连接。

@Override
public RpcResponse sendRequest(RpcRequest rpcRequest) {
    try {
        ChannelFuture channelFuture = bootstrap
                .connect(this.serviceCenter.serviceDiscovery(rpcRequest.getInterfaceName()))
                .sync();
        Channel channel = channelFuture.channel();
        channel.writeAndFlush(rpcRequest).addListener((ChannelFutureListener) future -> {
            if (!future.isSuccess()) {
                channel.close();
            }
        });
        channel.closeFuture().sync();
        AttributeKey<RpcResponse> key = AttributeKey.valueOf("rpcResponse");
        return channel.attr(key).get();
    } catch (Exception e) {
        return RpcResponse.fail("发送异常");
    }
}

服务端把响应写进这条连接之后，客户端 handler 只做两件事：塞进同一个 attribute，然后关闭 channel。sendRequest 堵在 closeFuture 上，关完就能 attr.get()。

@Override
protected void channelRead0(ChannelHandlerContext ctx, RpcResponse msg) {
    AttributeKey<RpcResponse> key = AttributeKey.valueOf("rpcResponse");
    ctx.channel().attr(key).set(msg);
    ctx.channel().close();
}

重试单独放在 GuavaRetryHandler。异常就重试；结果是 null，或者 code 和成功枚举对不上，也重试。固定等 2 秒，一共调用 3 次。

Retryer<RpcResponse> retryer = RetryerBuilder.<RpcResponse>newBuilder()
        .retryIfException()
        .retryIfResult(rpcResponse -> Objects.isNull(rpcResponse)
                || ObjectUtil.notEqual(RespCodeEnum.SUCCESS, rpcResponse.getCode()))
        .withWaitStrategy(WaitStrategies.fixedWait(2, TimeUnit.SECONDS))
        .withStopStrategy(StopStrategies.stopAfterAttempt(3))
        .build();
return retryer.call(() -> this.rpcClient.sendRequest(rpcRequest));

notEqual 比较的是枚举 RespCodeEnum.SUCCESS 和响应里的 Integer code，两者对不上，非空响应也会进入重试。默认代理调用的是 sendRequest，这条重试要自己接上 RpcClient 才会生效。

10. 跑通 UserService

接口和实现都很小。实现不读入参，固定返回一个用户，方便你确认字节确实走了一个来回。

public interface UserService {
    User getUserById(Integer id);
}

public class UserServiceImpl implements UserService {
    @Override
    public User getUserById(Integer id) {
        return User.builder().id(1).name("test").build();
    }
}

先起 ZooKeeper，再起服务端。start 会阻塞，这个测试方法不会自己结束。

UserService userService = new UserServiceImpl();
ServerProvider serverProvider = new ServerProvider("127.0.0.1", 9999);
serverProvider.provideServiceInterface(userService);

RpcServer rpcServer = new NettyRpcServer(serverProvider);
rpcServer.start(9999);

这时 ZooKeeper 上会出现：

/JotaroRpc/com.jotaro.api.UserService/127.0.0.1:9999

前半段是永久节点，最后一段是临时节点。客户端再拿代理调用：

NettyClientProxy proxy = new NettyClientProxy();
UserService userService = proxy.getProxy(UserService.class);
User user = userService.getUserById(1);

这一次会向 ZooKeeper 要 com.jotaro.api.UserService 的子节点，轮询选中 127.0.0.1:9999，发出一条 JSON 帧。服务端反射进 getUserById，把 User 写回后关闭连接。user 就是 id = 1、name = test。

ZooKeeper 不在本机 2181 时，在 classpath 放 zk.propertis：

jrpc.zk.address=127.0.0.1:2181

换均衡器，改 ZKServiceCenter 里的 this.loadBalancer。换序列化，改两端 pipeline 里 JotaroEncoder 的参数。帧头上的类型会跟着变，解码器按这个类型选实现。