Java架构-还不了解Flink底层RPC使用的框架和原理？那就认真看完( 四 ) _Java架构

3.3.1.2. 处理Rpc请求
当调用非AkkaInvocationHandler实现的方法时，则进行Rpc请求。
下面分析处理Rpc调用的流程。
AkkaInvocationHandler#invokeRpc，其方法如下：private Object invokeRpc(Method method, Object[] args) throws Exception {// 获取方法相应的信息String methodName = method.getName(); Class<?>[] parameterTypes = method.getParameterTypes(); Annotation[][] parameterAnnotations = method.getParameterAnnotations(); Time futureTimeout = extractRpcTimeout(parameterAnnotations, args, timeout);// 创建RpcInvocationMessage(可分为LocalRpcInvocation/RemoteRpcInvocation) final RpcInvocation rpcInvocation = createRpcInvocationMessage(methodName, parameterTypes, args); Class<?> returnType = method.getReturnType(); final Object result;// 无返回，则使用tell方法if (Objects.equals(returnType, Void.TYPE)) { tell(rpcInvocation); result = null; } else {// execute an asynchronous call// 有返回，则使用ask方法 CompletableFuture<?> resultFuture = ask(rpcInvocation, futureTimeout); CompletableFuture<?> completableFuture = resultFuture.thenApply((Object o) -> {// 调用返回后进行反序列化if (o instanceof SerializedValue) {try {return ((SerializedValue<?>) o).deserializeValue(getClass().getClassLoader()); } catch (IOException | ClassNotFoundException e) {throw new CompletionException(new RpcException("Could not deserialize the serialized payload of RPC method : " + methodName, e)); } } else {// 直接返回return o; } });// 若返回类型为CompletableFuture则直接赋值if (Objects.equals(returnType, CompletableFuture.class)) { result = completableFuture; } else {try {// 从CompletableFuture获取 result = completableFuture.get(futureTimeout.getSize(), futureTimeout.getUnit()); } catch (ExecutionException ee) {throw new RpcException("Failure while obtaining synchronous RPC result.", ExceptionUtils.stripExecutionException(ee)); } } }return result; }

AkkaRpcActor#handleRpcInvocation，其代码如下：

private void handleRpcInvocation(RpcInvocation rpcInvocation) { Method rpcMethod = null;try {// 获取方法的信息 String methodName = rpcInvocation.getMethodName(); Class<?>[] parameterTypes = rpcInvocation.getParameterTypes();// 在RpcEndpoint中找指定方法 rpcMethod = lookupRpcMethod(methodName, parameterTypes); } catch (ClassNotFoundException e) { log.error("Could not load method arguments.", e);// 异常处理 RpcConnectionException rpcException = new RpcConnectionException("Could not load method arguments.", e); getSender().tell(new Status.Failure(rpcException), getSelf()); } catch (IOException e) { log.error("Could not deserialize rpc invocation message.", e);// 异常处理 RpcConnectionException rpcException = new RpcConnectionException("Could not deserialize rpc invocation message.", e); getSender().tell(new Status.Failure(rpcException), getSelf()); } catch (final NoSuchMethodException e) { log.error("Could not find rpc method for rpc invocation.", e);// 异常处理 RpcConnectionException rpcException = new RpcConnectionException("Could not find rpc method for rpc invocation.", e); getSender().tell(new Status.Failure(rpcException), getSelf()); }if (rpcMethod != null) {try {// this supports declaration of anonymous classes rpcMethod.setAccessible(true);// 返回类型为空则直接进行invokeif (rpcMethod.getReturnType().equals(Void.TYPE)) {// No return value to send back rpcMethod.invoke(rpcEndpoint, rpcInvocation.getArgs()); }else {final Object result;try { result = rpcMethod.invoke(rpcEndpoint, rpcInvocation.getArgs()); }catch (InvocationTargetException e) { log.debug("Reporting back error thrown in remote procedure {}", rpcMethod, e);// tell the sender about the failure getSender().tell(new Status.Failure(e.getTargetException()), getSelf());return; }final String methodName = rpcMethod.getName();// 方法返回类型为CompletableFutureif (result instanceof CompletableFuture) {final CompletableFuture<?> responseFuture = (CompletableFuture<?>) result;// 发送结果（使用Patterns发送结果给调用者，并会进行序列化并验证结果大小） sendAsyncResponse(responseFuture, methodName); } else {// 类型非CompletableFuture，发送结果（使用Patterns发送结果给调用者，并会进行序列化并验证结果大小） sendSyncResponse(result, methodName); } } } catch (Throwable e) { log.error("Error while executing remote procedure call {}.", rpcMethod, e);// tell the sender about the failure getSender().tell(new Status.Failure(e), getSelf()); } } }

将结果返回给调用者AkkaInvocationHandler#ask；

经过上述步骤就完成Rpc（本地/远程）调用，可以看到底层也是通过Akka提供的tell/ask方法进行通信；经过上述步骤就完成Rpc（本地/远程）调用，可以看到底层也是通过Akka提供的tell/ask方法进行通信；
4. 总结
RPC框架是Flink任务运行的基础，Flink整个RPC框架基于Akka实现，并对Akka中的ActorSystem、Actor进行了封装和使用，文章主要分析了Flink底层RPC通信框架的实现和相关流程，Flink整个通信框架的组件主要由RpcEndpoint、RpcService、RpcServer、AkkaInvocationHandler、AkkaRpcActor等构成。RpcEndpoint定义了一个Actor的路径；RpcService提供了启动RpcServer、执行代码体等方法；RpcServer/AkkaInvocationHandler提供了与Actor通信的接口；AkkaRpcActor为Flink封装的Actor 。