Flink RPC原始碼流程-有解無憂

需要了解動態反射和RPC

文章目錄

- - 概述
  - 四個組件
  - 啟動流程
  - Rpc呼叫流程

概述

Akka系統的核心ActorSystem和Actor，若需構建一個Akka系統，首先需要創建ActorSystem，創建完ActorSystem后，可通過其創建Actor（注意：Akka不允許直接new一個Actor，只能通過 Akka 提供的某些 API 才能創建或查找 Actor，一般會通過 ActorSystem#actorOf和ActorContext#actorOf來創建 Actor），另外，我們只能通過ActorRef（Actor的參考，其對原生的 Actor 實體做了良好的封裝，外界不能隨意修改其內部狀態）來與Actor進行通信，
1、ActorSystem 是管理 Actor生命周期的組件， Actor是負責進行通信的組件
2、每個 Actor 都有一個 MailBox，別的 Actor 發送給它的訊息都首先儲存在 MailBox 中，通過這種
方式可以實作異步通信，
3、每個 Actor 是單執行緒的處理方式，不斷的從 MailBox 拉取訊息執行處理，所以對于 Actor 的訊息處
理，不適合呼叫會阻塞的處理方法，
4、Actor 可以改變他自身的狀態，可以接收訊息，也可以發送訊息，還可以生成新的 Actor，
5、如果一個 Actor 要和另外一個 Actor進行通信，則必須先獲取對方 Actor 的ActorRef 物件，然后通過該物件發送訊息即可，
6、通過 tell 發送異步訊息，不接收回應，通過 ask 發送異步訊息，得到 Future 回傳，通過異步回傳處理結果，
7、當在任意地方發現要創建這四個組件的任何一個組件的實體物件的時候，創建成功了之后，都會要去執行他的 onStart() ，在集群啟動的原始碼分析中，其實這些組件的很多的作業流程，都被放在 onStart() 里面，先執行構造方法，后執行onStart方法，
在這里插入圖片描述

四個組件

1、RpcGateway 網關(路由)，各種其他RPC組件，都是 RpcGateWay 的子類
2、RpcServer RpcService 和 RpcEndpoint 之間的粘合層
3、RpcEndpoint 業務邏輯載體，對應的 Actor 的封裝
4、RpcService 對應 ActorSystem 的封裝
在這里插入圖片描述

啟動流程

在RpcEndpoint中還定義了一些方法如runAsync(Runnable)、callAsync(Callable, Time)方法來執行Rpc呼叫，值得注意的是在Flink的設計中，對于同一個Endpoint，所有的呼叫都運行在主執行緒，因此不會有并發問題，當啟動RpcEndpoint/進行Rpc呼叫時，其會委托RcpServer進行處理，進入rpcService.startServer(this)，在RpcService中呼叫connect()方法與對端的RpcEndpoint（RpcServer）建立連接，connect()方法根據給的地址回傳InvocationHandler(AkkaInvocationHandler或FencedAkkaInvocationHandler，也就是對方的代理) xxxRpcGateWay(例如connect(rpcEndpoint.getAddress())) ，連接后回傳一個RpcGateway，即是他的實作類CompletableFuture，AkkaRpcService中封裝了ActorSystem，并保存了ActorRef到RpcEndpoint的映射關系，在構造RpcEndpoint時會啟動指定rpcEndpoint上的RpcServer，其會根據Endpoint型別（FencedRpcEndpoint或其他）來創建不同的Actor（FencedAkkaRpcActor或AkkaRpcActor），并將RpcEndpoint和Actor對應的ActorRef保存起來，然后使用動態代理創建RpcServer，具體代碼如下：

rpcService.startServer(this)

RPCEndpoint：
protected RpcEndpoint(final RpcService rpcService, final String endpointId) {
		this.rpcService = checkNotNull(rpcService, "rpcService");
		this.endpointId = checkNotNull(endpointId, "endpointId");
		//在構造RpcEndpoint時會啟動指定rpcEndpoint上的RpcServer
		this.rpcServer = rpcService.startServer(this);
		this.mainThreadExecutor = new MainThreadExecutor(rpcServer, this::validateRunsInMainThread);
	}

public <C extends RpcEndpoint & RpcGateway> RpcServer startServer(C rpcEndpoint) {
        checkNotNull(rpcEndpoint, "rpc endpoint");

        CompletableFuture<Void> terminationFuture = new CompletableFuture<>();
final Props akkaRpcActorProps;
// 根據RpcEndpoint型別創建不同型別的Props
if (rpcEndpoint instanceof FencedRpcEndpoint) {
            akkaRpcActorProps = Props.create(
                FencedAkkaRpcActor.class,
                rpcEndpoint,
                terminationFuture,
                getVersion(),
                configuration.getMaximumFramesize());
        } else {
            akkaRpcActorProps = Props.create(
                AkkaRpcActor.class,
                rpcEndpoint,
                terminationFuture,
                getVersion(),
                configuration.getMaximumFramesize());
        }

        ActorRef actorRef;
// 同步塊，創建Actor，并獲取對應的ActorRef
synchronized (lock) {
            checkState(!stopped, "RpcService is stopped");
            actorRef = actorSystem.actorOf(akkaRpcActorProps, rpcEndpoint.getEndpointId());
            actors.put(actorRef, rpcEndpoint);
        }

        LOG.info("Starting RPC endpoint for {} at {} .", rpcEndpoint.getClass().getName(), actorRef.path());

// 獲取Actor的路徑
final String akkaAddress = AkkaUtils.getAkkaURL(actorSystem, actorRef);
final String hostname;
        Option<String> host = actorRef.path().address().host();
if (host.isEmpty()) {
            hostname = "localhost";
        } else {
            hostname = host.get();
        }
// 決議該RpcEndpoint實作的所有RpcGateway介面
        Set<Class<?>> implementedRpcGateways = new HashSet<>(RpcUtils.extractImplementedRpcGateways(rpcEndpoint.getClass()));
        
// 額外添加RpcServer和AkkaBasedEnpoint類
        implementedRpcGateways.add(RpcServer.class);
        implementedRpcGateways.add(AkkaBasedEndpoint.class);

final InvocationHandler akkaInvocationHandler;

// 根據不同型別動態創建代理物件
if (rpcEndpoint instanceof FencedRpcEndpoint) {
// a FencedRpcEndpoint needs a FencedAkkaInvocationHandler
            akkaInvocationHandler = new FencedAkkaInvocationHandler<>(
                akkaAddress,
                hostname,
                actorRef,
                configuration.getTimeout(),
                configuration.getMaximumFramesize(),
                terminationFuture,
                ((FencedRpcEndpoint<?>) rpcEndpoint)::getFencingToken);

            implementedRpcGateways.add(FencedMainThreadExecutable.class);
        } else {
            akkaInvocationHandler = new AkkaInvocationHandler(
                akkaAddress,
                hostname,
                actorRef,
                configuration.getTimeout(),
                configuration.getMaximumFramesize(),
                terminationFuture);
        }

// Rather than using the System ClassLoader directly, we derive the ClassLoader
// from this class . That works better in cases where Flink runs embedded and all Flink
// code is loaded dynamically (for example from an OSGI bundle) through a custom ClassLoader
        ClassLoader classLoader = getClass().getClassLoader();

// 生成RpcServer物件，而后對該server的呼叫都會進入Handler的invoke方法處理，handler實作了多個介面的方法
@SuppressWarnings("unchecked")
        RpcServer server = (RpcServer) Proxy.newProxyInstance(
            classLoader,
            implementedRpcGateways.toArray(new Class<?>[implementedRpcGateways.size()]),
            akkaInvocationHandler);

return server;
    }

呼叫RpcEndpoint#start；
委托給RpcServer#start；

呼叫動態代理的AkkaInvocationHandler#invoke；發現呼叫的是StartStoppable#start方法，則直接進行本地方法呼叫；invoke方法的代碼如下：

RPCEndpoint：
public final void start() {
		rpcServer.start();
	}
AkkaInvocationHandler：
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
      Class<?> declaringClass = method.getDeclaringClass();

      Object result;
// 先匹配指定型別（handler已實作介面的方法），若匹配成功則直接進行本地方法呼叫；若匹配為FencedRpcGateway型別，則拋出例外（應該在FencedAkkaInvocationHandler中處理）；其他則進行Rpc呼叫
if (declaringClass.equals(AkkaBasedEndpoint.class) ||
          declaringClass.equals(Object.class) ||
          declaringClass.equals(RpcGateway.class) ||
          declaringClass.equals(StartStoppable.class) ||
          declaringClass.equals(MainThreadExecutable.class) ||
          declaringClass.equals(RpcServer.class)) {
          result = method.invoke(this, args);
      } else if (declaringClass.equals(FencedRpcGateway.class)) {
throw new UnsupportedOperationException("AkkaInvocationHandler does not support the call FencedRpcGateway#" +
              method.getName() + ". This indicates that you retrieved a FencedRpcGateway without specifying a " +
"fencing token. Please use RpcService#connect(RpcService, F, Time) with F being the fencing token to " +
"retrieve a properly FencedRpcGateway.");
      } else {
          result = invokeRpc(method, args);
      }

return result;
  }

呼叫AkkaInvocationHandler#start；
通過ActorRef#tell給對應的Actor發送訊息rpcEndpoint.tell(ControlMessages.START, ActorRef.noSender());
呼叫AkkaRpcActor#handleControlMessage處理控制型別訊息；
在主執行緒中將自身狀態變更為Started狀態；

經過上述步驟就完成了Actor的啟動程序，Actor啟動后便可與Acto通信讓其執行代碼（如runSync/callSync等）和處理Rpc請求了，下面分別介紹處理執行代碼和處理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);

// 回傳型別為空則直接進行invoke
if (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();

// 方法回傳型別為CompletableFuture
if (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；

補充：

AkkaRpcActor，會根據型別的不同，進行不同的處理

protected void handleRpcMessage(Object message) {
// 根據訊息型別不同進行不同的處理
if (message instanceof RunAsync) {
          handleRunAsync((RunAsync) message);
      } else if (message instanceof CallAsync) {
          handleCallAsync((CallAsync) message);
      } else if (message instanceof RpcInvocation) {
          handleRpcInvocation((RpcInvocation) message);
      } else {
          log.warn(
"Received message of unknown type {} with value {}. Dropping this message!",
              message.getClass().getName(),
              message);

          sendErrorIfSender(new AkkaUnknownMessageException("Received unknown message " + message +
" of type " + message.getClass().getSimpleName() + '.'));
      }
  }

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/432169.html

標籤：其他

上一篇：一文搞懂 RabbitMQ 延時佇列（訂單定時取消為例）

下一篇：Spark環境搭建（保姆級教程）