Apache Flink源码解析（七）Flink RPC的底层

2018-11-13 本文已影响247人铛铛铛clark

Prerequisites

Flink的RPC服务是基于Akka Remote实现的。一个简单的Akka Remoting ActorSystem的配置如下:
```
akka {
  actor {
    provider = remote
  }
  remote {
    enabled-transports = ["akka.remote.netty.tcp"]
    netty.tcp {
      hostname = "127.0.0.1"
      port = 2552
    }
 }
}
```
- 从这份配置文件可以看出，要建立一个ActorSystem，首先需要提供ActorSystem运行的机器的地址和端口。
参考Akka Remoting的文档，获取远程节点的Actor有两条途径。
- 第一条，通过actorSelection(path)，在这儿需要知道远程节点的地址。获取到了ActorSelection就已经可以发送消息过去，也可以通过回信获取这个Actor的ActorRef。
- 第二条，通过配置，配置文件如下。通过这种方式，远程系统的daemon会被请求建立这个Actor，ActorRef可以直接通过system.actorOf(new Props(...)获取。
```
akka {
  actor {
    deployment {
      /sampleActor {
        remote = "akka.tcp://sampleActorSystem@127.0.0.1:2553"
      }
    }
  }
}
```

定义RPC协议

RPC协议是客户端和服务端的通信接口。如下所示定义了一个BaseGateway的通信接口。
```
  public interface BaseGateway extends RpcGateway {
      CompletableFuture<Integer> foobar();
  }
```

在Flink中，RPC协议的定义通过实现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();
}

这个接口需要实现两个方法，分别是getAddress和getHostname。原因如下：
- 如上文所述，想要通过ActorSystem获取远程Actor，必须要有地址。而在Flink中，例如Yarn这种模式下，JobMaster会先建立ActorSystem，这时TaskExecutor的Container都还没有分配，自然无法在配置中指定远程Actor的地址，所以一个远程节点提供自己的地址是必须的。

实现RPC协议

Flink的RPC协议一般定义为一个Java接口，服务端需要实现这个接口。如下是上面定义的BaseGateway的实现。

  public static class BaseEndpoint extends RpcEndpoint implements BaseGateway {

      private final int foobarValue;

      protected BaseEndpoint(RpcService rpcService, int foobarValue) {
          super(rpcService);

          this.foobarValue = foobarValue;
      }

      @Override
      public CompletableFuture<Integer> foobar() {
          return CompletableFuture.completedFuture(foobarValue);
      }

      @Override
      public CompletableFuture<Void> postStop() {
          return CompletableFuture.completedFuture(null);
      }
  }

RpcEndpoint是rpc请求的接收端的基类。RpcEndpoint是通过RpcService来启动的。

构造并启动RpcService

RpcService会在每一个ClusterEntrypoint(JobMaster)和TaskManagerRunner(TaskExecutor)启动的过程中被初始化并启动。
RpcService主要负责启动RpcEndpoint(也就是服务端)，连接到远程的RpcEndpoint并提供一个代理(也就是客户端)。
此外，为了防止状态的concurrent modification，RpcEndpoint上所有的Rpc调用都只会运行在主线程上，RpcService提供了运行在其它线程的方法。

构造并启动RpcEndpoint(服务端)

每一个RpcEndpoint在初始化阶段会通过该节点的RpcService的startServer方法来初始化服务。

在该方法中创建了一个Akka的Actor，这个Actor也是Rpc调用的实际接收者，Rpc的请求会在客户端被封装成RpcInvocation对象以Akka消息的形式发送。

  @Override
  public <C extends RpcEndpoint & RpcGateway> RpcServer startServer(C rpcEndpoint) {
      ...
      // 新建一个包含了rpcEndpoint的AkkaRpcActor，负责接收封装成Akka消息的rpc请求
          akkaRpcActorProps = Props.create(AkkaRpcActor.class, rpcEndpoint, terminationFuture, getVersion());
      ...
          actorRef = actorSystem.actorOf(akkaRpcActorProps, rpcEndpoint.getEndpointId());
      ...
  }

接下来生成一个本地的InvocationHandler，用于将调用转换成消息发送到相应的RpcEndpoint(具体细节在下一节发送Rpc请求会详细介绍)

      ...
      // 获取这个Endpoint的所有Gateway，也就是所有RPC协议的接口
      Set<Class<?>> implementedRpcGateways = new HashSet<>(RpcUtils.extractImplementedRpcGateways(rpcEndpoint.getClass()));
      ...
      // 新建一个InvocationHandler，用于将rpc请求包装成LocalRpcInvocation消息并发送给RpcServer(本地)

      final InvocationHandler akkaInvocationHandler;
      ...
          akkaInvocationHandler = new AkkaInvocationHandler(akkaAddress, hostname, actorRef, timeout, maximumFramesize, terminationFuture);
      ...

通过Rpc接口和InvocationHandler构造一个代理对象，这个代理对象存在RpcEndpoint的RpcServer变量中，是给RpcEndpoint所在的JVM本地调用使用

      // 生成一个包含这些接口的代理，将调用转发到InvocationHandler
      @SuppressWarnings("unchecked")
      RpcServer server = (RpcServer) Proxy.newProxyInstance(classLoader, implementedRpcGateways.toArray(new Class<?>[implementedRpcGateways.size()]), akkaInvocationHandler);

      return server;
  }

启动RpcEndpoint
- 实际上就是启动构造阶段生成的RpcServer的start方法，这个方法由AkkaInvocationHandler实现，实际上就是向绑定的RpcEndpoint的Actor发送一条START消息，通知它服务已启动。

构造Rpc客户端

Rpc的客户端实际上是一个代理对象，构造这个代理对象，需要提供实现的接口和InvocationHandler，在Flink中有AkkaInvocationHandler的实现。
在构造RpcEndpoint的过程中实际上已经生成了一个供本地使用的Rpc客户端。并且通过RpcEndpoint的getSelfGateway方法可以直接获取这个代理对象。

而在远程调用时，则通过RpcService的connect方法获取远程RpcEndpoint的客户端(也是一个代理)。connect方法需要提供Actor的地址。(至于地址是如何获得的，可以通过LeaderRetrievalService，在这个部分不多做介绍。)

<C extends RpcGateway> CompletableFuture<C> connect(
    String address,
    Class<C> clazz);

首先通过地址获取ActorSelection，在Prerequisite中也介绍了这是连接远程Actor的方法之一

private <C extends RpcGateway> CompletableFuture<C> connectInternal(
        final String address,
        final Class<C> clazz,
        Function<ActorRef, InvocationHandler> invocationHandlerFactory) {
    ...
    // 通过地址获取ActorSelection, 并获取ActorRef引用
    final ActorSelection actorSel = actorSystem.actorSelection(address);

通过ActorSelection获取ActorRef并发送握手消息

    final Future<ActorIdentity> identify = Patterns
        .ask(actorSel, new Identify(42), timeout.toMilliseconds())
        .<ActorIdentity>mapTo(ClassTag$.MODULE$.<ActorIdentity>apply(ActorIdentity.class));
    final CompletableFuture<ActorIdentity> identifyFuture = FutureUtils.toJava(identify);
    final CompletableFuture<ActorRef> actorRefFuture = identifyFuture.thenApply(
    ...
    // 发送handshake消息
    final CompletableFuture<HandshakeSuccessMessage> handshakeFuture = actorRefFuture.thenCompose(
        (ActorRef actorRef) -> FutureUtils.toJava(
            Patterns
                .ask(actorRef, new RemoteHandshakeMessage(clazz, getVersion()), timeout.toMilliseconds())
                .<HandshakeSuccessMessage>mapTo(ClassTag$.MODULE$.<HandshakeSuccessMessage>apply(HandshakeSuccessMessage.class))));

最后根据ActorRef，通过InvocationHandlerFactory生成AkkaInvocationHandler并构造代理

    // 根据ActorRef引用生成InvocationHandler
    return actorRefFuture.thenCombineAsync(
        handshakeFuture,
        (ActorRef actorRef, HandshakeSuccessMessage ignored) -> {
            InvocationHandler invocationHandler = invocationHandlerFactory.apply(actorRef);
            ClassLoader classLoader = getClass().getClassLoader();
            @SuppressWarnings("unchecked")
            C proxy = (C) Proxy.newProxyInstance(
                classLoader,
                new Class<?>[]{clazz},
                invocationHandler);

            return proxy;
        },
        actorSystem.dispatcher());
}

而invocationHandlerFactory.apply方法如下

Tuple2<String, String> addressHostname = extractAddressHostname(actorRef);
return new AkkaInvocationHandler(addressHostname.f0, addressHostname.f1, actorRef, timeout, maximumFramesize, null);

发送Rpc请求

上文中客户端会提供代理对象，而代理对象会调用AkkaInvocationHandler的invoke方法并传入Rpc调用的方法和参数信息。

    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {

而在AkkaInvocationHandler对该方法的实现中，会判断方法属于哪个类，如果是Rpc方法的话就会调用invokeRpc方法。

首先将方法封装成一个RpcInvocation，它有两种实现，一种是本地的LocalRpcInvocation，不需要序列化，另一种是远程的RemoteRpcInvocation。根据当前AkkaInvocationHandler和对应的RpcEndpoint是否在同一个JVM中来判断生成哪一个。
```
private Object invokeRpc(Method method, Object[] args) throws Exception {
    ...
    final RpcInvocation rpcInvocation = createRpcInvocationMessage(methodName, parameterTypes, args);
```

根据返回类型判断使用tell还是ask的形式发送akka消息

    Class<?> returnType = method.getReturnType();
    final Object result;
    if (Objects.equals(returnType, Void.TYPE)) {
        tell(rpcInvocation);
        result = null;
    } else if (Objects.equals(returnType, CompletableFuture.class)) {
        // execute an asynchronous call
        result = ask(rpcInvocation, futureTimeout);
    } else {
        // execute a synchronous call
        CompletableFuture<?> futureResult = ask(rpcInvocation, futureTimeout);
        result = futureResult.get(futureTimeout.getSize(), futureTimeout.getUnit());
    }
    return result;
}

Rpc请求的处理

首先Rpc消息是通过RpcEndpoint所绑定的Actor的ActorRef发送的，所以接收到消息的就是RpcEndpoint构造期间生成的AkkRpcActor

akkaRpcActorProps = Props.create(AkkaRpcActor.class, rpcEndpoint, terminationFuture, getVersion());
actorRef = actorSystem.actorOf(akkaRpcActorProps, rpcEndpoint.getEndpointId());

AkkaRpcActor接收到的消息总共有三种
- 一种是握手消息，如上文所述，在客户端构造时会通过ActorSelection发送过来。收到消息后会检查接口，版本，如果一致就返回成功
- 第二种是启停消息。例如在RpcEndpoint调用start方法后，就会向自身发送一条Processing.START消息，来转换当前Actor的状态为STARTED。STOP也类似。并且只有在Actor状态为STARTED时才会处理Rpc请求
- 第三种就是Rpc请求消息，通过解析RpcInvocation获取方法名和参数类型，并从RpcEndpoint类中找到Method对象，并通过反射调用该方法。如果有返回结果，会以Akka消息的形式发送回sender。

Apache Flink源码解析（七）Flink RPC的底层

Prerequisites

定义RPC协议

实现RPC协议

构造并启动RpcService

构造并启动RpcEndpoint(服务端)

构造Rpc客户端

发送Rpc请求

Rpc请求的处理

猜你喜欢

热点阅读

Apache Flink源码解析 （七）Flink RPC的底层

Prerequisites

定义RPC协议

实现RPC协议

构造并启动RpcService

构造并启动RpcEndpoint(服务端)

构造Rpc客户端

发送Rpc请求

Rpc请求的处理

猜你喜欢

热点阅读

Apache Flink源码解析（七）Flink RPC的底层