Flink Rpc通信源码级详解:从RpcService到动态代理的完整调用链路

●位置透明：客户端不关心服务端在本地还是远程，统一通过 Gateway 接口调用

●异步非阻塞：所有 RPC 调用都是异步的，返回 CompletableFuture，不阻塞调用线程

●类型安全：通过 Java 接口定义 RPC 方法，编译期类型检查，避免字符串拼接错误

●高可用：支持 Leader 地址动态解析和自动重连，适应 JobManager 主备切换

●可扩展：RPC 框架抽象为接口，底层实现可替换（Akka → Netty）

●管理 RpcEndpoint 的生命周期（启动/停止）

●创建客户端 Gateway 代理（connect）

●管理 ActorSystem 或 Netty 服务

●核心方法：startServer()、connect()、stopServer()、getAddress()

●业务逻辑的实现基类，所有服务端组件继承它

●封装 ActorRef 和 MainThreadExecutor

●所有方法在 Actor 主线程执行，保证线程安全

●核心方法：onStart()、onStop()、runAsync()、callAsync()、getSelfGateway()

●定义 RPC 方法签名的接口，客户端通过它调用服务端

●方法返回 CompletableFuture（有返回值）或 void（fire-and-forget）

●实际运行时是 JDK 动态代理对象

●每个 RpcEndpoint 对应一个 RpcGateway 接口

●封装方法调用信息的可序列化对象

●包含：methodName（方法名）、parameterTypes（参数类型）、args（参数值）

●跨进程传输时被序列化，接收端反序列化后反射调用

●实现 Serializable 接口

RpcService → 启动 → RpcEndpoint（服务端）RpcService → connect → RpcGateway（客户端动态代理）RpcGateway 方法调用 → AkkaInvocationHandler → RpcInvocation → 网络传输 → AkkaRpcActor → 反射调用 RpcEndpoint

●持有 RpcGateway 动态代理对象

●调用 Gateway 方法发起 RPC 调用

●InvocationHandler 拦截方法调用，封装为 RpcInvocation

●序列化后通过网络发送到服务端

●等待服务端响应，CompletableFuture 完成

●启动 RpcEndpoint 监听 RPC 请求

●接收 RpcInvocation 消息，反序列化

●通过 Actor 主线程调度，反射调用 RpcEndpoint 实际方法

●封装响应返回客户端

●所有操作在 Actor 主线程执行，保证线程安全

public interface RpcService {    // 启动 RpcEndpoint，返回 Gateway 代理（服务端自身调用用）    <C extends RpcEndpoint & RpcGateway> C startServer(C rpcEndpoint);    // 连接到远端 RpcEndpoint，返回 Gateway 代理    <C extends RpcGateway> CompletableFuture<C> connect(String address, Class<C> clazz);    // 停止 RpcEndpoint    void stopServer(RpcEndpoint rpcEndpoint);    // 获取 RPC 服务地址    String getAddress();    // 获取调度器    ScheduledExecutorService getScheduledExecutorService();    // 关闭 RPC 服务    CompletableFuture<Void> stopAsync();}

●startServer 泛型参数  要求端点同时继承 RpcEndpoint 和实现 RpcGateway

●connect 返回 CompletableFuture，因为连接建立是异步的（可能需要解析地址、建立 TCP 连接）

●stopServer 停止单个端点，stopAsync 关闭整个 RPC 服务

public abstract class RpcEndpoint implements RpcGateway {    protected final RpcService rpcService;    protected final String endpointId;    private ActorRef actorRef;    private MainThreadExecutor mainThreadExecutor;    protected RpcEndpoint(RpcService rpcService, String endpointId) {        this.rpcService = rpcService;        this.endpointId = endpointId;    }    // 端点启动时调用，子类重写    public void onStart() throws Exception {}    // 端点停止时调用，子类重写    public void onStop() throws Exception {}    // 获取自身 Gateway 代理（可以像调用远程一样调用自身方法）    protected <C extends RpcGateway> C getSelfGateway(Class<C> selfGatewayType) {        return Proxy.newProxyInstance(            selfGatewayType.getClassLoader(),            new Class<?>[]{selfGatewayType},            new AkkaInvocationHandler(actorRef, rpcService.getTimeout())        );    }    // 异步执行 Runnable（在 Actor 主线程）    protected void runAsync(Runnable runnable) {        actorRef.tell(new RunAsync(runnable), ActorRef.noSender());    }    // 异步执行 Callable（在 Actor 主线程）    protected <V> CompletableFuture<V> callAsync(Callable<V> callable, Time timeout) {        final CompletableFuture<V> resultFuture = new CompletableFuture<>();        actorRef.tell(new CallAsync(callable, resultFuture), ActorRef.noSender());        return resultFuture;    }    // 定时调度    protected ScheduledFuture<?> scheduleRunAsync(Runnable runnable, Time delay) {        return mainThreadExecutor.schedule(runnable, delay.getSize(), delay.getUnit());    }    // 校验是否在主线程执行    protected void validateRunsInMainThread() {        if (!mainThreadExecutor.isCurrentThreadMainThread()) {            throw new IllegalStateException("Run method in main thread.");        }    }    // 设置 ActorRef（由 AkkaRpcService 在启动时调用）    void setActorRef(ActorRef actorRef) {        this.actorRef = actorRef;    }    // 设置 MainThreadExecutor（由 AkkaRpcActor 在 preStart 时调用）    void setMainThreadExecutor(MainThreadExecutor mainThreadExecutor) {        this.mainThreadExecutor = mainThreadExecutor;    }}

●主线程执行模型：所有 RpcEndpoint 方法必须在 Actor 主线程执行，validateRunsInMainThread() 可以校验

●runAsync/callAsync：跨线程操作必须通过这两个方法提交到 Actor 主线程，避免线程安全问题

●getSelfGateway：获取自身的 Gateway 代理，可以像调用远程一样调用自身方法，统一编程模型

●setActorRef/setMainThreadExecutor：包级私有方法，由 AkkaRpcService 和 AkkaRpcActor 在启动时调用，不对外暴露

public interface RpcGateway {    // 获取 RPC 端点地址    String getAddress();    // 获取端点 ID    String getEndpointId();}

public interface ResourceManagerGateway extends RpcGateway {    // 请求 Slot    CompletableFuture<Acknowledge> requestSlot(        JobID jobId,        AllocationID allocationId,        ResourceProfile resourceProfile,        String targetAddress,        Time timeout);    // 注册 TaskManager    CompletableFuture<RegistrationResponse> registerTaskManager(        String taskManagerRegistrationId,        TaskManagerLocation taskManagerLocation,        TaskExecutorGateway taskExecutorGateway,        Time timeout);    // 心跳    void heartbeatFromTaskManager(ResourceID resourceID, TaskExecutorHeartbeat heartbeat);}

●有返回值的方法返回 CompletableFuture，使用 ask 模式

●无返回值的方法返回 void，使用 tell 模式（fire-and-forget）

●最后一个参数通常是 Time timeout，指定 RPC 超时时间

public class RpcInvocation implements Serializable {    private static final long serialVersionUID = 1L;    private final String methodName;    private final Class<?>[] parameterTypes;    private final Object[] args;    public RpcInvocation(Method method, Object[] args) {        this.methodName = method.getName();        this.parameterTypes = method.getParameterTypes();        this.args = args;    }    public RpcInvocation(String methodName, Class<?>[] parameterTypes, Object[] args) {        this.methodName = methodName;        this.parameterTypes = parameterTypes;        this.args = args;    }    public String getMethodName() { return methodName; }    public Class<?>[] getParameterTypes() { return parameterTypes; }    public Object[] getArgs() { return args; }    // 根据方法名和参数类型在目标类中查找 Method    public Method getMethod(Class<?> clazz) throws NoSuchMethodException {        return clazz.getMethod(methodName, parameterTypes);    }}

●只存储方法名、参数类型、参数值，不存储 Method 对象（Method 不可序列化）

●接收端通过 getMethod(clazz) 根据方法名和参数类型反射查找 Method

●所有参数必须可序列化，否则序列化失败

●这是典型的命令模式（Command Pattern），将方法调用封装为对象

public class AkkaRpcService implements RpcService {    private final ActorSystem actorSystem;    private final Time timeout;    private final Map<ActorPath, RpcEndpoint> actors = new HashMap<>();    public AkkaRpcService(ActorSystem actorSystem, Time timeout) {        this.actorSystem = actorSystem;        this.timeout = timeout;    }    @Override    public <C extends RpcEndpoint & RpcGateway> C startServer(C rpcEndpoint) {        // 1. 创建 AkkaRpcActor，传入 rpcEndpoint        Props props = Props.create(AkkaRpcActor.class, rpcEndpoint);        ActorRef actorRef = actorSystem.actorOf(props, rpcEndpoint.getEndpointId());        // 2. 设置 ActorRef 到 RpcEndpoint        rpcEndpoint.setActorRef(actorRef);        // 3. 注册到 actors 映射        actors.put(actorRef.path(), rpcEndpoint);        return rpcEndpoint;    }    @Override    public <C extends RpcGateway> CompletableFuture<C> connect(String address, Class<C> clazz) {        // 1. 通过 ActorSelection 选择远程 Actor        ActorSelection actorSelection = actorSystem.actorSelection(address);        // 2. 解析 ActorSelection 为 ActorRef（异步）        return actorSelection.resolveOne(timeout)            .thenApply(actorRef -> {                // 3. 创建 JDK 动态代理                return Proxy.newProxyInstance(                    clazz.getClassLoader(),                    new Class<?>[]{clazz},                    new AkkaInvocationHandler(actorRef, timeout)                );            });    }    @Override    public void stopServer(RpcEndpoint rpcEndpoint) {        ActorRef actorRef = rpcEndpoint.getActorRef();        // 发送 PoisonPill 消息，Actor 处理完当前消息后停止        actorRef.tell(PoisonPill.getInstance(), ActorRef.noSender());        actors.remove(actorRef.path());    }}

●startServer：创建 AkkaRpcActor，将 RpcEndpoint 传入 Actor，设置 ActorRef

●connect：通过 ActorSelection 选择远程 Actor，resolveOne 解析为 ActorRef，创建动态代理

●stopServer：发送 PoisonPill 消息，Actor 处理完当前消息后优雅停止

public class AkkaInvocationHandler implements InvocationHandler {    private final ActorRef rpcEndpoint;    private final Time timeout;    public AkkaInvocationHandler(ActorRef rpcEndpoint, Time timeout) {        this.rpcEndpoint = rpcEndpoint;        this.timeout = timeout;    }    @Override    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {        Class<?> returnType = method.getReturnType();        // 1. 如果是 CompletableFuture 返回类型（ask 模式）        if (returnType == CompletableFuture.class) {            // 2. 创建 RpcInvocation 消息            RpcInvocation invocation = new RpcInvocation(method, args);            // 3. 通过 akka ask 发送消息，返回 Future            Future<Object> future = Patterns.ask(rpcEndpoint, invocation, timeout.toMilliseconds());            // 4. 转换 Akka Future 为 CompletableFuture            return FutureUtils.toJava(future);        } else if (returnType == Void.TYPE) {            // 5. 无返回值（tell 模式，fire-and-forget）            RpcInvocation invocation = new RpcInvocation(method, args);            rpcEndpoint.tell(invocation, ActorRef.noSender());            return null;        } else {            // 6. 其他返回类型不支持            throw new RuntimeException("RpcGateway methods must return CompletableFuture or void.");        }    }}

●返回类型判断：返回 CompletableFuture 走 ask 模式，返回 void 走 tell 模式，其他类型不支持

●ask 模式：通过 Patterns.ask() 发送消息，等待响应，返回 CompletableFuture

●tell 模式：通过 ActorRef.tell() 发送消息，不等待响应，fire-and-forget

●RpcInvocation 封装：将 Method 和 args 封装为可序列化的 RpcInvocation 消息

public class AkkaRpcActor extends UntypedActor {    private final RpcEndpoint rpcEndpoint;    private MainThreadExecutor mainThreadExecutor;    public AkkaRpcActor(RpcEndpoint rpcEndpoint) {        this.rpcEndpoint = rpcEndpoint;    }    @Override    public void preStart() throws Exception {        // 1. 创建 MainThreadExecutor，绑定当前 Actor 线程        mainThreadExecutor = new MainThreadExecutor(getContext().dispatcher(), self());        // 2. 设置到 RpcEndpoint        rpcEndpoint.setMainThreadExecutor(mainThreadExecutor);        // 3. 调用 RpcEndpoint.onStart()        rpcEndpoint.onStart();    }    @Override    public void postStop() throws Exception {        // 调用 RpcEndpoint.onStop()        rpcEndpoint.onStop();    }    @Override    public void onReceive(Object message) throws Exception {        if (message instanceof RpcInvocation) {            handleRpcInvocation((RpcInvocation) message);        } else if (message instanceof RunAsync) {            handleRunAsync((RunAsync) message);        } else if (message instanceof CallAsync) {            handleCallAsync((CallAsync) message);        } else if (message instanceof ControlMessages) {            handleControlMessage((ControlMessages) message);        } else {            unhandled(message);        }    }    // 处理 RPC 调用消息    private void handleRpcInvocation(RpcInvocation invocation) throws Exception {        // 1. 根据方法名和参数类型查找 Method        Method method = invocation.getMethod(rpcEndpoint.getClass());        // 2. 反射调用 RpcEndpoint 方法        Object result = method.invoke(rpcEndpoint, invocation.getArgs());        // 3. 如果有发送者（ask 模式），发送响应        ActorRef sender = getSender();        if (sender != ActorRef.noSender()) {            if (result instanceof CompletableFuture) {                // 4. 如果返回 CompletableFuture，等待完成后发送响应                ((CompletableFuture<?>) result).whenComplete((r, t) -> {                    if (t != null) {                        sender.tell(new Status.Failure(t), self());                    } else {                        sender.tell(new RpcResponse(r), self());                    }                });            } else {                // 5. 直接发送响应                sender.tell(new RpcResponse(result), self());            }        }    }    // 处理异步 Runnable    private void handleRunAsync(RunAsync runAsync) {        runAsync.getRunnable().run();    }    // 处理异步 Callable    private void handleCallAsync(CallAsync callAsync) {        try {            Object result = callAsync.getCallable().call();            callAsync.getResultFuture().complete(result);        } catch (Exception e) {            callAsync.getResultFuture().completeExceptionally(e);        }    }}

●preStart：创建 MainThreadExecutor，设置到 RpcEndpoint，调用 onStart()

●handleRpcInvocation：根据方法名和参数类型反射查找 Method，调用 RpcEndpoint 方法，处理 CompletableFuture 返回

●CompletableFuture 处理：如果方法返回 CompletableFuture，等待 Future 完成后再发送响应，支持异步返回值

●postStop：调用 RpcEndpoint.onStop() 释放资源

●客户端持有 ResourceManagerGateway 动态代理对象

●调用 requestSlot(jobId, allocationId, resourceProfile, targetAddress, timeout)

●进入 JDK 动态代理的 invoke() 方法

●AkkaInvocationHandler.invoke() 拦截方法调用

●获取 Method 对象和 args 参数数组

●检查方法返回类型：CompletableFuture → ask 模式，void → tell 模式

●创建 RpcInvocation 对象：new RpcInvocation(method, args)

●包含 methodName="requestSlot"、parameterTypes、args

●RpcInvocation 实现 Serializable，可跨进程传输

●requestSlot 返回 CompletableFuture，走 ask 模式

●如果是 void 方法（如 heartbeatFromTaskManager），走 tell 模式 fire-and-forget

●ask 模式需要等待响应，tell 模式发送后立即返回

●Akka Remote 层通过 Serialization.serialize() 将 RpcInvocation 序列化为字节数组

●默认使用 Java 序列化（RpcInvocation 实现 Serializable）

●序列化后的字节数组封装为 Akka 远程帧

●通过 RemoteActorRef.tell()/ask() 发送到远端 AkkaRpcActor

●Netty Client 将帧写入 TCP Channel，发送到服务端

●ask 模式返回 Akka Future，等待服务端响应

●Netty Server 接收字节流

●通过 LengthFieldBasedFrameDecoder 解码为完整帧（解决粘包问题）

●帧包含协议版本、消息类型、长度、消息体

●Akka Serialization 将字节数组反序列化为 RpcInvocation 对象

●反序列化需要类加载器能加载 RpcInvocation 类和参数类

●反序列化失败会抛出异常

●通过远端 ActorRef 将 RpcInvocation 放入 AkkaRpcActor 的 Mailbox 队列

●Mailbox 是 FIFO 队列，消息按发送顺序排列

●同一 Actor 的消息在同一个 Mailbox 中

●Dispatcher 从线程池分配一个线程

●执行 Mailbox.run() 方法

●从 Mailbox 取出 RpcInvocation 消息

●调用 AkkaRpcActor.onReceive() 处理

●handleRpcInvocation() 处理 RpcInvocation

●根据 methodName 和 parameterTypes 反射查找 Method 对象

●调用 method.invoke(rpcEndpoint, args) 执行实际业务逻辑

●所有操作在 Actor 主线程执行，保证线程安全

●如果方法返回 CompletableFuture，等待 Future 完成

●返回值封装为 RpcResponse 对象

●通过 Akka 消息返回给客户端的 Future

●客户端的 CompletableFuture 被 complete，调用方获取结果

RpcEndpoint 方法返回值 → 封装 RpcResponse → 序列化 → Netty TCP 回传 → 反序列化 → CompletableFuture.complete()

●如果方法返回 CompletableFuture，AkkaRpcActor 会等待 Future 完成再发送响应

●如果方法执行抛出异常，封装为 Status.Failure 发送，客户端 Future 会 completeExceptionally

●响应也需要序列化，返回值必须可序列化

●akka.actor.ask-timeout：ask 模式默认超时，默认 100s

●akka.client.timeout：客户端连接超时，默认 60s

●方法级超时：通过方法最后一个 Time 参数指定

●基于 ZooKeeper（Standalone/YARN 部署）或 Kubernetes（K8s 部署）

●监听 Leader 节点地址变化

●地址变化时通知 LeaderRetrievalListener

LeaderRetrievalService 监听 → 地址变更 → LeaderRetrievalListener.notifyLeaderAddress() → RpcService.connect(newAddress) → 建立新连接 → 旧连接关闭

●LeaderRetrievalService：检索 Leader 地址，启动/停止监听

●LeaderRetrievalListener：地址变更回调接口

●LeaderConnectionService：管理与 Leader 的 RPC 连接，地址变化时自动重连

●RegisteredRpcConnection：封装 RpcGateway 连接，处理连接生命周期

●Leader 切换时，RPC 请求自动重定向到新 Leader

●旧 Leader 的 RPC 连接自动关闭

●正在进行的 RPC 请求可能失败，客户端需要重试

●Flink 的 ResourceManager/JobMaster 都有 Leader 选举和重连机制

●优点：兼容性最好，无需额外配置，所有 Serializable 对象都支持

●缺点：性能差、体积大、序列化/反序列化慢

●适用：RPC 调用频率相对低，兼容性更重要

●Flink RPC 默认使用 Java 序列化（RpcInvocation 实现 Serializable）

●优点：高性能、体积小、跨语言支持好

●缺点：需定义 .proto 文件、需代码生成、学习成本高

●适用：高性能场景、跨语言通信

●Flink 某些内部通信使用 Protobuf

●优点：高性能、无需定义文件、自动序列化 POJO、体积比 Java 小

●缺点：兼容性稍差、版本升级可能不兼容

●适用：Flink 数据序列化默认使用 Kryo

●RPC 层可配置使用 Kryo

akka {  actor {    serializers {      java = "akka.serialization.JavaSerializer"      proto = "akka.remote.serialization.ProtobufSerializer"    }    serialization-bindings {      "java.io.Serializable" = java    }  }}

●所有 RPC 方法参数和返回值必须可序列化

●大对象序列化开销大，考虑传引用或分片

●序列化异常是常见 RPC 问题，检查参数是否实现 Serializable

●Akka 依赖体积大，增加 Flink 发行包大小

●Akka 配置复杂，学习成本高

●Akka 版本升级可能引入不兼容

●Akka Actor 模型对 Flink RPC 来说过重，Flink 只需要简单的 RPC 功能

●自研 Netty RPC 更轻量、更可控、性能更优

●纯 Netty 实现，依赖少，体积小

●自定义 RPC 协议，更高效

●支持连接多路复用，减少连接数

●配置简洁，易于理解和调优

●性能更优（减少 Akka 抽象层开销）

# Akka 超时配置akka.actor.ask-timeout: 100s          # ask 模式默认超时akka.client.timeout: 60s                # 客户端连接超时akka.remote.netty.tcp.connection-timeout: 120s  # TCP 连接超时# 心跳检测akka.remote.watch-failure-detector.heartbeat-interval: 10s       # 心跳间隔akka.remote.watch-failure-detector.acceptable-heartbeat-pause: 60s  # 可接受心跳暂停akka.remote.transport-failure-detector.heartbeat-interval: 10s    # 传输故障心跳间隔akka.remote.transport-failure-detector.acceptable-heartbeat-pause: 60s# Actor 线程池akka.actor.default-dispatcher.fork-join-executor.parallelism-factor: 2.0  # 并行度因子akka.actor.default-dispatcher.fork-join-executor.parallelism-min: 8       # 最小线程数akka.actor.default-dispatcher.fork-join-executor.parallelism-max: 64      # 最大线程数# 序列化akka.serialization-java.enabled: on     # Java 序列化开关# 远程事件日志akka.remote.log-remote-lifecycle-events: off  # 远程生命周期事件日志# 守护监督策略akka.actor.guardian-supervisor-strategy: "akka.actor.StoppingSupervisorStrategy"

●跨机房部署：调大心跳超时（acceptable-heartbeat-pause），避免网络抖动误判

●高并发场景：调大 Actor 线程池（parallelism-max），避免线程不足

●低延迟场景：调小 ask-timeout，快速失败，但要避免误超时

●生产环境：关闭远程事件日志，减少日志开销

●现象：RPC 调用超时，抛出 AskTimeoutException

●原因：服务端处理慢、网络延迟、Actor 邮箱堆积、服务端挂掉

●排查：检查服务端是否正常、检查网络延迟、检查 Actor 邮箱队列长度、检查服务端是否有阻塞操作

●解决：优化服务端处理逻辑、增大超时时间、检查网络连接

●现象：RPC 调用抛出 IOException，连接断开

●原因：网络故障、服务端重启、防火墙拦截、TCP 连接超时

●排查：检查网络连通性、检查服务端是否正常、检查防火墙配置

●解决：Flink 自动重连机制会重新建立连接，持续失败需排查网络

●现象：RPC 调用抛出 NotSerializableException

●原因：方法参数或返回值不可序列化

●排查：检查所有参数类型是否实现 Serializable、检查返回值类型

●解决：让参数类实现 Serializable，或改用可序列化的类型

●现象：JobManager 切换时 RPC 调用失败

●原因：Leader 地址变更，旧连接失效

●排查：检查 ZooKeeper/K8s Leader 选举状态

●解决：Flink 自动重新解析 Leader 地址并重连，客户端需重试

●现象：RPC 调用延迟越来越大，内存持续增长

●原因：服务端处理速度跟不上请求速度、服务端有阻塞操作

●排查：检查 Actor 邮箱队列长度、检查服务端 CPU 使用率、检查是否有阻塞 IO

●解决：优化服务端处理逻辑、避免阻塞操作、增加服务端并行度

●RPC 方法返回 CompletableFuture，异步非阻塞

●无返回值的通知类方法用 void，fire-and-forget 性能更好

●所有方法参数和返回值实现 Serializable

●设置合理的 RPC 超时时间，避免无限等待

●服务端方法避免阻塞操作，阻塞操作用异步处理

●跨机房部署调大心跳超时，避免误判

●监控 RPC 调用延迟和成功率，及时发现问题

●大对象参数考虑传引用或分片，避免序列化开销大

●不要在 RPC 方法中执行阻塞操作（Thread.sleep、同步 IO、Future.get()）

●不要传递不可序列化的对象（Thread、Connection、Stream）

●不要设置过大的超时时间，故障时无法快速失败

●不要忽略 RPC 异常，需要处理超时和连接断开

●不要在客户端线程中等待 RPC 结果，用 CompletableFuture 链式调用

●不要频繁创建和销毁 RPC 连接，连接创建有开销

●不要混淆 RpcGateway 和 RpcEndpoint，Gateway 是客户端接口，Endpoint 是服务端实现