「Flink」工具人Flink之组件通信

private void runCluster(Configuration configuration) throws Exception {
synchronized (lock) {
      //各项服务初始化，其中就包涵了AkkaService的初始化
      initializeServices(configuration);
      .....
}
}

protected void initializeServices(Configuration configuration) throws Exception {


LOG.info("Initializing cluster services.");


synchronized (lock) {
final String bindAddress = configuration.getString(JobManagerOptions.ADDRESS);
final String portRange = getRPCPortRange(configuration);
//创建RpcService
commOnRpcService= createRpcService(configuration, bindAddress, portRange);
......
}
}


@Nonnull
private RpcService createRpcService(Configuration configuration, String bindAddress, String portRange) throws Exception {
    //创建了AkkaRpcService
return AkkaRpcServiceUtils.createRpcService(bindAddress, portRange, configuration);
}

protected RpcEndpoint(final RpcService rpcService, final String endpointId) {
this.rpcService = checkNotNull(rpcService, "rpcService");
this.endpointId = checkNotNull(endpointId, "endpointId");
    //调用AkkaRpcService创建本服务的ActorRef,并绑定远端Gateway的动态代理
this.rpcServer = rpcService.startServer(this);


this.mainThreadExecutor = new MainThreadExecutor(rpcServer, this::validateRunsInMainThread);
}

@Override
public RpcServer startServer(C rpcEndpoint) {
checkNotNull(rpcEndpoint, "rpc endpoint");
.......
ActorRef actorRef;
    //在这里创建了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());


final String akkaAddress = AkkaUtils.getAkkaURL(actorSystem, actorRef);
final String hostname;
Option host = actorRef.path().address().host();
if (host.isEmpty()) {
hostname = "localhost";
} else {
hostname = host.get();
}


Set> implementedRpcGateways = new HashSet<>(RpcUtils.extractImplementedRpcGateways(rpcEndpoint.getClass()));


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();

//在这里创建了动态代理
@SuppressWarnings("unchecked")
RpcServer server = (RpcServer) Proxy.newProxyInstance(
classLoader,
implementedRpcGateways.toArray(new Class[implementedRpcGateways.size()]),
akkaInvocationHandler);


return server;
}

/**
* Start the rpc service and begin to run the job.
*
* @param newJobMasterId The necessary fencing token to run the job
* @return Future acknowledge if the job could be started. Otherwise the future contains an exception
*/
public CompletableFuture start(final JobMasterId newJobMasterId) throws Exception {
// make sure we receive RPC and async calls
super.start();


return callAsyncWithoutFencing(() -> startJobExecution(newJobMasterId), RpcUtils.INF_TIMEOUT);
}

@Override
public void start() {
rpcEndpoint.tell(Processing.START, ActorRef.noSender());
}

/**
* Retrying registration for the job manager <--> task manager connection.
*/
private static final class JobManagerRetryingRegistration
extends RetryingRegistration<JobMasterId, JobMasterGateway, JMTMRegistrationSuccess> {


private final String taskManagerRpcAddress;


private final TaskManagerLocation taskManagerLocation;


JobManagerRetryingRegistration(
Logger log,
RpcService rpcService,
String targetName,
Class targetType,
String targetAddress,
JobMasterId jobMasterId,
RetryingRegistrationConfiguration retryingRegistrationConfiguration,
String taskManagerRpcAddress,
TaskManagerLocation taskManagerLocation) {
super(
log,
rpcService,
targetName,
targetType,
targetAddress,
jobMasterId,
retryingRegistrationConfiguration);


this.taskManagerRpcAddress = taskManagerRpcAddress;
this.taskManagerLocation = Preconditions.checkNotNull(taskManagerLocation);
}


@Override
protected CompletableFuture invokeRegistration(
JobMasterGateway gateway,
JobMasterId jobMasterId,
long timeoutMillis) throws Exception {

      //通知JobMaster，来连接TaskManager
return gateway.registerTaskManager(taskManagerRpcAddress, taskManagerLocation, Time.milliseconds(timeoutMillis));
}
}

@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
Class declaringClass = method.getDeclaringClass();


Object result;


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

protected RpcInvocation createRpcInvocationMessage(
final String methodName,
final Class[] parameterTypes,
final Object[] args) throws IOException {
final RpcInvocation rpcInvocation;


if (isLocal) {
rpcInvocation = new LocalRpcInvocation(
methodName,
parameterTypes,
args);
} else {
try {
RemoteRpcInvocation remoteRpcInvocation = new RemoteRpcInvocation(
methodName,
parameterTypes,
args);


if (remoteRpcInvocation.getSize() > maximumFramesize) {
throw new IOException("The rpc invocation size exceeds the maximum akka framesize.");
} else {
rpcInvocation = remoteRpcInvocation;
}
} catch (IOException e) {
LOG.warn("Could not create remote rpc invocation message. Failing rpc invocation because...", e);
throw e;
}
}


return rpcInvocation;
}

@Override
public CompletableFuture registerTaskManager(
final String taskManagerRpcAddress,
final TaskManagerLocation taskManagerLocation,
final Time timeout) {


final ResourceID taskManagerId = taskManagerLocation.getResourceID();


if (registeredTaskManagers.containsKey(taskManagerId)) {
final RegistrationResponse respOnse= new JMTMRegistrationSuccess(resourceId);
return CompletableFuture.completedFuture(response);
} else {
      // 连接TaskManager 并返回TaskExecutor
return getRpcService()
.connect(taskManagerRpcAddress, TaskExecutorGateway.class)
.handleAsync(
(TaskExecutorGateway taskExecutorGateway, Throwable throwable) -> {
......
},
getMainThreadExecutor());
}
}