1. 程式人生 > >【Flink】Flink 底層RPC框架分析

【Flink】Flink 底層RPC框架分析

1. 前言

對於Flink中各個元件(JobMaster、TaskManager、Dispatcher等),其底層RPC框架基於Akka實現,本文著重分析Flink中的Rpc框架實現機制及梳理其通訊流程。

2. Akka介紹

由於Flink底層Rpc是基於Akka實現,我們先了解下Akka的基本使用。

Akka是一個開發併發、容錯和可伸縮應用的框架。它是Actor Model的一個實現,和Erlang的併發模型很像。在Actor模型中,所有的實體被認為是獨立的actors。actors和其他actors通過傳送非同步訊息通訊。Actor模型的強大來自於非同步。它也可以顯式等待響應,這使得可以執行同步操作。但是,強烈不建議同步訊息,因為它們限制了系統的伸縮性。每個actor有一個郵箱(mailbox),它收到的訊息儲存在裡面。另外,每一個actor維護自身單獨的狀態。一個Actors網路如下所示:

每個actor是一個單一的執行緒,它不斷地從其郵箱中poll(拉取)訊息,並且連續不斷地處理。對於已經處理過的訊息的結果,actor可以改變它自身的內部狀態或者傳送一個新訊息或者孵化一個新的actor。儘管單個的actor是自然有序的,但一個包含若干個actor的系統卻是高度併發的並且極具擴充套件性的。因為那些處理執行緒是所有actor之間共享的。這也是我們為什麼不該在actor執行緒裡呼叫可能導致阻塞的“呼叫”。因為這樣的呼叫可能會阻塞該執行緒使得他們無法替其他actor處理訊息。

2.1. 建立Akka系統

Akka系統的核心ActorSystem和Actor,若需構建一個Akka系統,首先需要建立ActorSystem,建立完ActorSystem後,可通過其建立Actor(注意:Akka不允許直接new一個Actor,只能通過 Akka 提供的某些 API 才能建立或查詢 Actor,一般會通過 ActorSystem#actorOf和ActorContext#actorOf來建立 Actor),另外,我們只能通過ActorRef(Actor的引用, 其對原生的 Actor 例項做了良好的封裝,外界不能隨意修改其內部狀態)來與Actor進行通訊。如下程式碼展示瞭如何配置一個Akka系統。

// 1. 構建ActorSystem
// 使用預設配置
ActorSystem system = ActorSystem.create("sys");
// 也可顯示指定appsys配置
// ActorSystem system1 = ActorSystem.create("helloakka", ConfigFactory.load("appsys"));

// 2. 構建Actor,獲取該Actor的引用,即ActorRef
ActorRef helloActor = system.actorOf(Props.create(HelloActor.class), "helloActor");

// 3. 給helloActor傳送訊息
helloActor.tell("hello helloActor", ActorRef.noSender());

// 4. 關閉ActorSystem
system.terminate();

在Akka中,建立的每個Actor都有自己的路徑,該路徑遵循 ActorSystem 的層級結構,大致如下:

本地:akka://sys/user/helloActor
遠端:akka.tcp://[email protected]:2020/user/remoteActor 

其中本地路徑含義如下:

  • sys,建立的ActorSystem的名字;
  • user,通過ActorSystem#actorOf和ActorContext#actorOf 方法建立的 Actor 都屬於/user下,與/user對應的是/system, 其是系統層面建立的,與系統整體行為有關,在開發階段並不需要對其過多關注;
  • helloActor,我們建立的HelloActor。

其中遠端部分路徑含義如下:

  • akka.tcp,遠端通訊方式為tcp;
  • [email protected]:2020,ActorSystem名字及遠端主機ip和埠號。

2.2. 根據path獲取Actor

若提供了Actor的路徑,可以通過路徑獲取到ActorRef,然後與之通訊,程式碼如下所示:

ActorSystem system = ActorSystem.create("sys");
ActorSelection as = system.actorSelection("/path/to/actor");

Timeout timeout = new Timeout(Duration.create(2, "seconds"));
Future<ActorRef> fu = as.resolveOne(timeout);

fu.onSuccess(new OnSuccess<ActorRef>() {
    @Override
    public void onSuccess(ActorRef actor) {
        System.out.println("actor:" + actor);
        actor.tell("hello actor", ActorRef.noSender());
    }
}, system.dispatcher());

fu.onFailure(new OnFailure() {
    @Override
    public void onFailure(Throwable failure) {
        System.out.println("failure:" + failure);
    }
}, system.dispatcher());

由上面可知,若需要與遠端Actor通訊,路徑中必須提供ip:port。

2.3. 與Actor通訊

2.3.1. tell方式

當使用tell方式時,表示僅僅使用非同步方式給某個Actor傳送訊息,無需等待Actor的響應結果,並且也不會阻塞後續程式碼的執行,如:

helloActor.tell("hello helloActor", ActorRef.noSender());

其中:第一個引數為訊息,它可以是任何可序列化的資料或物件,第二個引數表示傳送者,通常來講是另外一個 Actor 的引用, ActorRef.noSender()表示無傳送者((實際上是一個 叫做deadLetters的Actor)。

2.3.2. ask方式

當我們需要從Actor獲取響應結果時,可使用ask方法,ask方法會將返回結果包裝在scala.concurrent.Future中,然後通過非同步回撥獲取返回結果。 如呼叫方:

// 非同步傳送訊息給Actor,並獲取響應結果
Future<Object> fu = Patterns.ask(printerActor, "hello helloActor", timeout);
fu.onComplete(new OnComplete<Object>() {
    @Override
    public void onComplete(Throwable failure, String success) throws Throwable {
         if (failure != null) {
             System.out.println("failure is " + failure);
         } else {
             System.out.println("success is " + success);
         }
    }
}, system.dispatcher());

HelloActor處理訊息方法的程式碼大致如下:

private void handleMessage(Object object) {
    if (object instanceof String) {
      String str = (String) object;
      log.info("[HelloActor] message is {}, sender is {}", str,     getSender().path().toString());
      // 給傳送者傳送訊息
      getSender().tell(str, getSelf());
    }
  }

上面主要介紹了Akka中的ActorSystem、Actor,及與Actor的通訊;Flink藉此構建了其底層通訊系統。

3. RPC類圖結構

下圖展示了Flink中RPC框架中涉及的主要類。

3.1. RpcGateway

Flink的RPC協議通過RpcGateway來定義;由前面可知,若想與遠端Actor通訊,則必須提供地址(ip和port),如在Flink-on-Yarn模式下,JobMaster會先啟動ActorSystem,此時TaskExecutor的Container還未分配,後面與TaskExecutor通訊時,必須讓其提供對應地址,從類繼承圖可以看到基本上所有元件都實現了RpcGateway介面,其程式碼如下:

public interface RpcGateway {

    /**
     * Returns the fully qualified address under which the associated rpc endpoint is reachable.
     *
     * @return Fully qualified (RPC) address under which the associated rpc endpoint is reachable
     */
    String getAddress();

    /**
     * Returns the fully qualified hostname under which the associated rpc endpoint is reachable.
     *
     * @return Fully qualified hostname under which the associated rpc endpoint is reachable
     */
    String getHostname();
}

3.2. RpcEndpoint

每個RpcEndpoint對應了一個路徑(endpointId和actorSystem共同確定),每個路徑對應一個Actor,其實現了RpcGateway介面,其建構函式如下:

protected RpcEndpoint(final RpcService rpcService, final String endpointId) {
    // 儲存rpcService和endpointId
    this.rpcService = checkNotNull(rpcService, "rpcService");
    this.endpointId = checkNotNull(endpointId, "endpointId");
    // 通過RpcService啟動RpcServer
    this.rpcServer = rpcService.startServer(this);
    // 主執行緒執行器,所有呼叫在主執行緒中序列執行
    this.mainThreadExecutor = new MainThreadExecutor(rpcServer,     this::validateRunsInMainThread);
}

在RpcEndpoint中還定義了一些方法如runAsync(Runnable)callAsync(Callable, Time)方法來執行Rpc呼叫,值得注意的是在Flink的設計中,對於同一個Endpoint,所有的呼叫都執行在主執行緒,因此不會有併發問題,當啟動RpcEndpoint/進行Rpc呼叫時,其會委託RcpServer進行處理。

3.3. RpcService

Rpc服務的介面,其主要作用如下:

  • 根據提供的RpcEndpoint來啟動RpcServer(Actor);
  • 根據提供的地址連線到RpcServer,並返回一個RpcGateway;
  • 延遲/立刻排程Runnable、Callable;
  • 停止RpcServer(Actor)或自身服務;

在Flink中其實現類為AkkaRpcService。

3.3.1. AkkaRpcService

AkkaRpcService中封裝了ActorSystem,並儲存了ActorRef到RpcEndpoint的對映關係,在構造RpcEndpoint時會啟動指定rpcEndpoint上的RpcServer,其會根據Endpoint型別(FencedRpcEndpoint或其他)來建立不同的Actor(FencedAkkaRpcActor或AkkaRpcActor),並將RpcEndpoint和Actor對應的ActorRef儲存起來,然後使用動態代理建立RpcServer,具體程式碼如下:

    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;
    }

當啟動RpcServer後,即建立了相應的Actor(注意此時Actor的處於停止狀態)和動態代理物件,需要呼叫RpcEndpoint#start啟動啟動Actor,此時啟動RpcEndpoint流程如下(以非FencedRpcEndpoint為例):

  • 呼叫RpcEndpoint#start;

  • 委託給RpcServer#start;

  • 呼叫動態代理的AkkaInvocationHandler#invoke;發現呼叫的是StartStoppable#start方法,則直接進行本地方法呼叫;invoke方法的程式碼如下:

    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請求;

3.3.1.1. 執行程式碼

與Actor通訊,通過呼叫runSync/callSync等方法其直接執行程式碼。

下面以scheduleRunAsync方法為例分析請求Actor執行程式碼流程,方法程式碼如下:

public void scheduleRunAsync(Runnable runnable, long delayMillis) {
        checkNotNull(runnable, "runnable");
        checkArgument(delayMillis >= 0, "delay must be zero or greater");
        // 判斷是否為本地Actor
        if (isLocal) {
            long atTimeNanos = delayMillis == 0 ? 0 : System.nanoTime() + (delayMillis * 1_000_000);
            // 向Actor傳送訊息runnable
            tell(new RunAsync(runnable, atTimeNanos));
        } else {
            // 丟擲異常,不支援遠端傳送Runnable訊息
            throw new RuntimeException("Trying to send a Runnable to a remote actor at " +
                rpcEndpoint.path() + ". This is not supported.");
        }
    }
  • AkkaInvocationHandler#invoke -> AkkaInvocation#scheduleRunAsync;

  • AkkaRpcActor#handleMessage -> AkkaRpcActor#handleRpcMessage,其中handleRpcMessage方法如下:

      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() + '.'));
          }
      }
    
  • AkkaRpcActor#handleRunAsync,其程式碼如下:

      private void handleRunAsync(RunAsync runAsync) {
          // 獲取延遲排程時間
          final long timeToRun = runAsync.getTimeNanos();
          final long delayNanos;
    
            // 若為0或已經到了排程時間,則立刻進行排程
          if (timeToRun == 0 || (delayNanos = timeToRun - System.nanoTime()) <= 0) {
              // run immediately
              try {
                  runAsync.getRunnable().run();
              } catch (Throwable t) {
                  log.error("Caught exception while executing runnable in main thread.", t);
                  ExceptionUtils.rethrowIfFatalErrorOrOOM(t);
              }
          }
          else {
              // schedule for later. send a new message after the delay, which will then be immediately executed
                // 計算出延遲時間
              FiniteDuration delay = new FiniteDuration(delayNanos, TimeUnit.NANOSECONDS);
    
                // 重新封裝訊息
              RunAsync message = new RunAsync(runAsync.getRunnable(), timeToRun);
    
              final Object envelopedSelfMessage = envelopeSelfMessage(message);
    
                // 等待指定延遲時間後給自己再發送一個訊息
              getContext().system().scheduler().scheduleOnce(delay, getSelf(), envelopedSelfMessage,
                      getContext().dispatcher(), ActorRef.noSender());
          }
      }
    

    注意:當還未到排程時間時,該Actor會延遲一段時間後再次給自己傳送訊息;

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

經過上述步驟就完成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