TaskManagerRunner 启动 Akka Actor System 源码关键流程

private void startTaskManagerRunnerServices() throws Exception {synchronized (lock) {rpcSystem = RpcSystem.load(configuration);this.executor =Executors.newScheduledThreadPool(Hardware.getNumberCPUCores(),new ExecutorThreadFactory("taskmanager-future"));highAvailabilityServices =HighAvailabilityServicesUtils.createHighAvailabilityServices(configuration,executor,AddressResolution.NO_ADDRESS_RESOLUTION,rpcSystem,this);JMXService.startInstance(configuration.getString(JMXServerOptions.JMX_SERVER_PORT));rpcService = createRpcService(configuration, highAvailabilityServices, rpcSystem);// ...
}// TaskManagerRunner
static RpcService createRpcService(final Configuration configuration,final HighAvailabilityServices haServices,final RpcSystem rpcSystem)throws Exception {checkNotNull(configuration);checkNotNull(haServices);
return RpcUtils.createRemoteRpcService(rpcSystem,configuration,determineTaskManagerBindAddress(configuration, haServices, rpcSystem),//...);}// RpcUtils
public static RpcService createRemoteRpcService(RpcSystem rpcSystem,Configuration configuration,@Nullable String externalAddress,String externalPortRange,@Nullable String bindAddress,@SuppressWarnings("OptionalUsedAsFieldOrParameterType") Optional<Integer> bindPort)throws Exception {RpcSystem.RpcServiceBuilder rpcServiceBuilder =rpcSystem.remoteServiceBuilder(configuration, externalAddress, externalPortRange);if (bindAddress != null) {rpcServiceBuilder = rpcServiceBuilder.withBindAddress(bindAddress);}if (bindPort.isPresent()) {rpcServiceBuilder = rpcServiceBuilder.withBindPort(bindPort.get());}return rpcServiceBuilder.createAndStart();
}// AkkaRpcServiceUtils
public AkkaRpcService createAndStart(TriFunction<ActorSystem, AkkaRpcServiceConfiguration, ClassLoader, AkkaRpcService>constructor)throws Exception {if (actorSystemExecutorConfiguration == null) {actorSystemExecutorConfiguration =AkkaUtils.getForkJoinExecutorConfig(AkkaBootstrapTools.getForkJoinExecutorConfiguration(configuration));}final ActorSystem actorSystem;// akka internally caches the context class loader// make sure it uses the plugin class loadertry (TemporaryClassLoaderContext ignored =TemporaryClassLoaderContext.of(getClass().getClassLoader())) {if (externalAddress == null) {// create local actor systemactorSystem =AkkaBootstrapTools.startLocalActorSystem(configuration,actorSystemName,logger,actorSystemExecutorConfiguration,customConfig);} else {// create remote actor systemactorSystem =AkkaBootstrapTools.startRemoteActorSystem(configuration,actorSystemName,externalAddress,externalPortRange,bindAddress,Optional.ofNullable(bindPort),logger,actorSystemExecutorConfiguration,customConfig);}}return constructor.apply(actorSystem,AkkaRpcServiceConfiguration.fromConfiguration(configuration),RpcService.class.getClassLoader());}
}// AkkaBootstrapTools
private static ActorSystem startRemoteActorSystem(Configuration configuration,String actorSystemName,String externalAddress,int externalPort,String bindAddress,int bindPort,Logger logger,Config actorSystemExecutorConfiguration,Config customConfig)throws Exception {String externalHostPortUrl =NetUtils.unresolvedHostAndPortToNormalizedString(externalAddress, externalPort);String bindHostPortUrl =NetUtils.unresolvedHostAndPortToNormalizedString(bindAddress, bindPort);logger.info("Trying to start actor system, external address {}, bind address {}.",externalHostPortUrl,bindHostPortUrl);try {Config akkaConfig =AkkaUtils.getAkkaConfig(configuration,new HostAndPort(externalAddress, externalPort),new HostAndPort(bindAddress, bindPort),actorSystemExecutorConfiguration);if (customConfig != null) {akkaConfig = customConfig.withFallback(akkaConfig);}return startActorSystem(akkaConfig, actorSystemName, logger);} catch (Throwable t) {if (t instanceof ChannelException) {Throwable cause = t.getCause();if (cause != null && t.getCause() instanceof BindException) {throw new IOException("Unable to create ActorSystem at address "+ bindHostPortUrl+ " : "+ cause.getMessage(),t);}}throw new Exception("Could not create actor system", t);}
}private static ActorSystem startActorSystem(Config akkaConfig, String actorSystemName, Logger logger) {logger.debug("Using akka configuration\n {}", akkaConfig);ActorSystem actorSystem = AkkaUtils.createActorSystem(actorSystemName, akkaConfig);logger.info("Actor system started at {}", AkkaUtils.getAddress(actorSystem));return actorSystem;
}// AkkaUtils
public static ActorSystem createActorSystem(String actorSystemName, Config akkaConfig) {// Initialize slf4j as logger of Akka's Netty instead of java.util.logging (FLINK-1650)InternalLoggerFactory.setDefaultFactory(new Slf4JLoggerFactory());return RobustActorSystem.create(actorSystemName, akkaConfig);
}public static RobustActorSystem create(String name, Config applicationConfig) {return create(name, applicationConfig, FatalExitExceptionHandler.INSTANCE);
}@VisibleForTesting
static RobustActorSystem create(String name,Config applicationConfig,Thread.UncaughtExceptionHandler uncaughtExceptionHandler) {return create(name,ActorSystemSetup.create(BootstrapSetup.create(Optional.empty(),Optional.of(applicationConfig),Optional.empty())),Option.apply(uncaughtExceptionHandler));
}private static RobustActorSystem create(String name,ActorSystemSetup setup,Option<Thread.UncaughtExceptionHandler> uncaughtExceptionHandler) {final Optional<BootstrapSetup> bootstrapSettings = setup.get(BootstrapSetup.class);final ClassLoader classLoader = RobustActorSystem.class.getClassLoader();final Config appConfig =bootstrapSettings.map(BootstrapSetup::config).flatMap(RobustActorSystem::toJavaOptional).orElseGet(() -> ConfigFactory.load(classLoader));final Option<ExecutionContext> defaultEC =toScalaOption(bootstrapSettings.map(BootstrapSetup::defaultExecutionContext).flatMap(RobustActorSystem::toJavaOptional));final RobustActorSystem robustActorSystem =new RobustActorSystem(name, appConfig, classLoader, defaultEC, setup) {@Overridepublic Thread.UncaughtExceptionHandler uncaughtExceptionHandler() {return uncaughtExceptionHandler.getOrElse(super::uncaughtExceptionHandler);}};robustActorSystem.start();return robustActorSystem;
}private static RobustActorSystem create(String name,ActorSystemSetup setup,Option<Thread.UncaughtExceptionHandler> uncaughtExceptionHandler) {final Optional<BootstrapSetup> bootstrapSettings = setup.get(BootstrapSetup.class);final ClassLoader classLoader = RobustActorSystem.class.getClassLoader();final Config appConfig =bootstrapSettings.map(BootstrapSetup::config).flatMap(RobustActorSystem::toJavaOptional).orElseGet(() -> ConfigFactory.load(classLoader));final Option<ExecutionContext> defaultEC =toScalaOption(bootstrapSettings.map(BootstrapSetup::defaultExecutionContext).flatMap(RobustActorSystem::toJavaOptional));final RobustActorSystem robustActorSystem =new RobustActorSystem(name, appConfig, classLoader, defaultEC, setup) {@Overridepublic Thread.UncaughtExceptionHandler uncaughtExceptionHandler() {return uncaughtExceptionHandler.getOrElse(super::uncaughtExceptionHandler);}};robustActorSystem.start();return robustActorSystem;
}