本文主要讲解WebMonitorEndpoint启动过程源码分析，通过创建WebMonitorEndpoint和启动WebMonitorEndpoint启动，在启动过程中初始化了好多handler,这里handler可以理解为web中的sevlet。
1.WebMonitorEndpoint的创建入口
接收客户端接收的各种rest的请求，内部初始化了很多handler
restEndpointFactory变量的对象是：SessionRestEndpointFactory
//创建webMonitorEndpointwebMonitorEndpoint =restEndpointFactory.createRestEndpoint(configuration,dispatcherGatewayRetriever,resourceManagerGatewayRetriever,blobServer,executor,metricFetcher,highAvailabilityServices.getClusterRestEndpointLeaderElectionService(),fatalErrorHandler);log.debug("Starting Dispatcher REST endpoint.");//启动服务webMonitorEndpoint，监控客户端提供的请求webMonitorEndpoint.start();
2. 创建DispatcherRestEndpoint
这里就是new创建的对象DispatcherRestEndpoint
/** {@link RestEndpointFactory} which creates a {@link DispatcherRestEndpoint}. */public enum SessionRestEndpointFactory implements RestEndpointFactory<DispatcherGateway> {INSTANCE;@Overridepublic WebMonitorEndpoint<DispatcherGateway> createRestEndpoint(Configuration configuration,LeaderGatewayRetriever<DispatcherGateway> dispatcherGatewayRetriever,LeaderGatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever,TransientBlobService transientBlobService,ScheduledExecutorService executor,MetricFetcher metricFetcher,LeaderElectionService leaderElectionService,FatalErrorHandler fatalErrorHandler)throws Exception {final RestHandlerConfiguration restHandlerConfiguration =RestHandlerConfiguration.fromConfiguration(configuration);return new DispatcherRestEndpoint(dispatcherGatewayRetriever,configuration,restHandlerConfiguration,resourceManagerGatewayRetriever,transientBlobService,executor,metricFetcher,leaderElectionService,RestEndpointFactory.createExecutionGraphCache(restHandlerConfiguration),fatalErrorHandler);}}
3. 启动服务webMonitorEndpoint
初始化好多Handler，包括JobSubmitHandler，咱们后面提交任务的时候会调用
4. DispatcherRestEndpoint类的初始化initializeHandlers
@Overrideprotected List<Tuple2<RestHandlerSpecification, ChannelInboundHandler>> initializeHandlers(final CompletableFuture<String> localAddressFuture) {//DispatcherRestEndpoint extends WebMonitorEndpoint<DispatcherGateway>//这里实际调用的是WebMonitorEndpoint的initializeHandlers方法List<Tuple2<RestHandlerSpecification, ChannelInboundHandler>> handlers =super.initializeHandlers(localAddressFuture);// Add the Dispatcher specific handlersfinal Time timeout = restConfiguration.getTimeout();//客户端提交应用程序后，由JobManager 中的Netty服务端的JobSubmitHandler来执行处理JobSubmitHandler jobSubmitHandler =new JobSubmitHandler(leaderRetriever, timeout, responseHeaders, executor, clusterConfiguration);//这里在handlers中添加了一个handler是jobSubmitHandlerhandlers.add(Tuple2.of(jobSubmitHandler.getMessageHeaders(), jobSubmitHandler));return handlers;}
5. WebMonitorEndpoint类的initializeHandlers方法
这里主要是创建各种handler,类似web服务中的servlet方法
无论是哪个Handler都有一个handleRequest方法，channelRrerad0的底层，最终调用的就是Handler.handleRequest()方法，将来提交Job时，由WebMonitorEndPoint接收到，调转到JobSubmitHandler来执行，最终执行handleRequest()
/*** Handler的作用，对应到Flink web页面的一个rest服务，Handler = Servlet 总共有60多个* @param localAddressFuture future rest address of the RestServerEndpoint* @return*/@Overrideprotected List<Tuple2<RestHandlerSpecification, ChannelInboundHandler>> initializeHandlers(final CompletableFuture<String> localAddressFuture) {//初始化容器30个ArrayList<Tuple2<RestHandlerSpecification, ChannelInboundHandler>> handlers =new ArrayList<>(30);//value 是：ChannelInboundHandler有一个channelRead0()方法，会自动调用卑Netty//无论是哪个Handler都有有一个handleRequest方法//将来提交Job的时候，由WebMonitorEndPoint接收到，调转到JobSubmitHandler来执行，最终执行handleRequestfinal Collection<Tuple2<RestHandlerSpecification, ChannelInboundHandler>>webSubmissionHandlers = initializeWebSubmissionHandlers(localAddressFuture);handlers.addAll(webSubmissionHandlers);
6.启动 webMonitorEndpoint.start();
主节点上的WebMonitorEndpoint启动完成，Netty的服务端进行启动完成
提交任务的时候，会看到Netty的客户端启动
public final void start() throws Exception {synchronized (lock) {Preconditions.checkState(state == State.CREATED, "The RestServerEndpoint cannot be restarted.");log.info("Starting rest endpoint.");final Router router = new Router();final CompletableFuture<String> restAddressFuture = new CompletableFuture<>();//初始化各种Handler,包括JobSubmitHandlerhandlers = initializeHandlers(restAddressFuture);/* sort the handlers such that they are ordered the following:* jobs* jobs/overview* jobs/:jobid* jobs/:jobid/config* :**/Collections.sort(handlers, RestHandlerUrlComparator.INSTANCE);checkAllEndpointsAndHandlersAreUnique(handlers);//注册服务handlers.forEach(handler -> registerHandler(router, handler, log));//启动Netty服务端ChannelInitializer<SocketChannel> initializer =new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws ConfigurationException {RouterHandler handler = new RouterHandler(router, responseHeaders);// SSL should be the first handler in the pipelineif (isHttpsEnabled()) {ch.pipeline().addLast("ssl",new RedirectingSslHandler(restAddress,restAddressFuture,sslHandlerFactory));}ch.pipeline().addLast(new HttpServerCodec()).addLast(new FileUploadHandler(uploadDir)).addLast(new FlinkHttpObjectAggregator(maxContentLength, responseHeaders));for (InboundChannelHandlerFactory factory :inboundChannelHandlerFactories) {Optional<ChannelHandler> channelHandler =factory.createHandler(configuration, responseHeaders);if (channelHandler.isPresent()) {ch.pipeline().addLast(channelHandler.get());}}ch.pipeline().addLast(new ChunkedWriteHandler()).addLast(handler.getName(), handler).addLast(new PipelineErrorHandler(log, responseHeaders));}};//初始化两个工作组NioEventLoopGroup bossGroup =new NioEventLoopGroup(1, new ExecutorThreadFactory("flink-rest-server-netty-boss"));NioEventLoopGroup workerGroup =new NioEventLoopGroup(0, new ExecutorThreadFactory("flink-rest-server-netty-worker"));bootstrap = new ServerBootstrap();bootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(initializer);Iterator<Integer> portsIterator;try {portsIterator = NetUtils.getPortRangeFromString(restBindPortRange);} catch (IllegalConfigurationException e) {throw e;} catch (Exception e) {throw new IllegalArgumentException("Invalid port range definition: " + restBindPortRange);}int chosenPort = 0;while (portsIterator.hasNext()) {try {chosenPort = portsIterator.next();final ChannelFuture channel;if (restBindAddress == null) {channel = bootstrap.bind(chosenPort);} else {channel = bootstrap.bind(restBindAddress, chosenPort);}serverChannel = channel.syncUninterruptibly().channel();break;} catch (final Exception e) {// syncUninterruptibly() throws checked exceptions via Unsafe// continue if the exception is due to the port being in use, fail early// otherwiseif (!(e instanceof java.net.BindException)) {throw e;}}}if (serverChannel == null) {throw new BindException("Could not start rest endpoint on any port in port range "+ restBindPortRange);}log.debug("Binding rest endpoint to {}:{}.", restBindAddress, chosenPort);final InetSocketAddress bindAddress = (InetSocketAddress) serverChannel.localAddress();final String advertisedAddress;if (bindAddress.getAddress().isAnyLocalAddress()) {advertisedAddress = this.restAddress;} else {advertisedAddress = bindAddress.getAddress().getHostAddress();}port = bindAddress.getPort();log.info("Rest endpoint listening at {}:{}", advertisedAddress, port);restBaseUrl = new URL(determineProtocol(), advertisedAddress, port, "").toString();restAddressFuture.complete(restBaseUrl);state = State.RUNNING;//主节点上的WebMonitorEndpoint启动完成，//任务提交的时候：启动Netty的客户端//启动startInternal();}}
7. startInternal();执行选举WebMonitorEndPoint的Leader
/*** 主节点，三件重要的事情* 1.ResourceManager启动* 2.Dispatch启动* 3.WebMonitorEndPoint* 都会进行选举* @throws Exception*/@Overridepublic void startInternal() throws Exception {//执行选举：ZooKeeperMultipleComponentLeaderElectionDriver// 这里的this,leaderElectionService.start(this)，最终参与选举的某个获胜的角色会调用：isLeader ==> this.grantLeaderShip(),//如果失败就调用notLeader()leaderElectionService.start(this);//启动一个定时任务，定期删除不用的缓存文件startExecutionGraphCacheCleanupTask();if (hasWebUI) {log.info("Web frontend listening at {}.", getRestBaseUrl());}}
8. startExecutionGraphCacheCleanupTask();删除缓存定时任务
private void startExecutionGraphCacheCleanupTask() {final long cleanupInterval = 2 * restConfiguration.getRefreshInterval();//定时执行一次清理任务executionGraphCache.cleanup//对应缓存文件删除executionGraphCleanupTask =executor.scheduleWithFixedDelay(executionGraphCache::cleanup,cleanupInterval,cleanupInterval,TimeUnit.MILLISECONDS);}
9. executionGraphCache::cleanup满足条件删除缓存
@Overridepublic void cleanup() {long currentTime = System.currentTimeMillis();// remove entries which have exceeded their time to live//如果满足条件就进行删除cachedExecutionGraphs.values().removeIf((ExecutionGraphEntry entry) -> currentTime >= entry.getTTL());}
10.咱们回头看第7步中的选举过程执行
//执行选举：MultipleComponentLeaderElectionDriver
// 这里的this,leaderElectionService.start(this)，最终参与选举的某个获胜的角色会调用：isLeader ==> this.grantLeaderShip(),
检查leader状态，准备创建leader
/** Callback from leader contenders when they start their service. */private void addContender(EmbeddedLeaderElectionService service, LeaderContender contender) {synchronized (lock) {checkState(!shutdown, "leader election service is shut down");checkState(!service.running, "leader election service is already started");try {if (!allLeaderContenders.add(service)) {throw new IllegalStateException("leader election service was added to this service multiple times");}service.contender = contender;service.running = true;updateLeader().whenComplete((aVoid, throwable) -> {if (throwable != null) {fatalError(throwable);}});} catch (Throwable t) {fatalError(t);}}}
11.进行选举Leader
@GuardedBy("lock")private CompletableFuture<Void> updateLeader() {// this must be called under the lockassert Thread.holdsLock(lock);if (currentLeaderConfirmed == null && currentLeaderProposed == null) {// we need a new leaderif (allLeaderContenders.isEmpty()) {// no new leader available, tell everyone that there is no leader currentlyreturn notifyAllListeners(null, null);} else {// propose a leader and ask itfinal UUID leaderSessionId = UUID.randomUUID();EmbeddedLeaderElectionService leaderService = allLeaderContenders.iterator().next();currentLeaderSessionId = leaderSessionId;currentLeaderProposed = leaderService;currentLeaderProposed.isLeader = true;LOG.info("Proposing leadership to contender {}",leaderService.contender.getDescription());return execute(new GrantLeadershipCall(leaderService.contender, leaderSessionId, LOG));}} else {return CompletableFuture.completedFuture(null);}}
12.创建线程，执行选举
这里创建线程的方式是用实现Runnable接口来实现的
private static class GrantLeadershipCall implements Runnable {private final LeaderContender contender;private final UUID leaderSessionId;private final Logger logger;GrantLeadershipCall(LeaderContender contender, UUID leaderSessionId, Logger logger) {this.contender = checkNotNull(contender);this.leaderSessionId = checkNotNull(leaderSessionId);this.logger = checkNotNull(logger);}@Overridepublic void run() {try {contender.grantLeadership(leaderSessionId);} catch (Throwable t) {logger.warn("Error granting leadership to contender", t);contender.handleError(t instanceof Exception ? (Exception) t : new Exception(t));}}}
13.确认Leader,到此整个WebMonitorEndpoint启动过程执行完
org.apache.flink.runtime.webmonitor.WebMonitorEndpoint类的grantLeadership()方法进行确认Leader选举完成
// -------------------------------------------------------------------------// LeaderContender// -------------------------------------------------------------------------@Overridepublic void grantLeadership(final UUID leaderSessionID) {log.info("{} was granted leadership with leaderSessionID={}",getRestBaseUrl(),leaderSessionID);leaderElectionService.confirmLeadership(leaderSessionID, getRestBaseUrl());}
14.总结WebMonitorEndpoint的启动过程
The shortest answer is doing!
最简短的回答就是行动！
感谢阅读。期待点赞、分享、关注。
文章转载自beenrun，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。
