Hadoop3.2.1 【 YARN 】源码分析 : ContainersLauncher源码浅析

向右看齐 2023-01-10 09:43 202阅读 0赞

# 一. 前言 #

Container启动是由ContainersLauncher服务完成的， 维护了一个线程池以并行完成Container相关操作， 比如启动或者杀死Container， 其中启动Container请求是由ApplicationMaster发起的， 而杀死Container请求则可能来自ApplicationMaster或者ResourceManager。

ContainersLauncher是Containers 的启动程序, 必须ResourceLocalizationService启动之后,才能启动此服务. 因为它依赖于本地文件系统上创建系统目录.

# 二. 属性 #

没啥特殊的属性,

*  线程池 containerLauncher , 用于Container的启动 .
 *  ContainerExecutor这个需要关注一下. YARN提供了两种ContainerExecutor实现， 一种是DefaultContainerExecutor， 另一种是LinuxContainerExecutor。

// 上下文内容
      private Context context;
    
      // ContainerExecutor : 一种是DefaultContainerExecutor， 另一种是LinuxContainerExecutor
      private ContainerExecutor exec;
    
      // 
      private Dispatcher dispatcher;
    
      
      private ContainerManagerImpl containerManager;
    
      
      private LocalDirsHandlerService dirsHandler;
    
      // 启动 container的线程池
      @VisibleForTesting
      public ExecutorService containerLauncher =
          HadoopExecutors.newCachedThreadPool(
            new ThreadFactoryBuilder()
              .setNameFormat("ContainersLauncher #%d")
              .build());
    
      // 记录Container 的集合
      @VisibleForTesting
      public final Map<ContainerId, ContainerLaunch> running =  Collections.synchronizedMap(new HashMap<ContainerId, ContainerLaunch>());

# 三. 构造方法 #

没啥可说的,就是执行初始化赋值操作…

@VisibleForTesting
      public ContainersLauncher(Context context, Dispatcher dispatcher,
          ContainerExecutor exec, LocalDirsHandlerService dirsHandler,
          ContainerManagerImpl containerManager) { 
        this();
    
        // 执行初始化操作...
        init(context, dispatcher, exec, dirsHandler, containerManager);
      }
    
      @Override
      public void init(Context nmContext, Dispatcher nmDispatcher,
          ContainerExecutor containerExec, LocalDirsHandlerService nmDirsHandler,
          ContainerManagerImpl nmContainerManager) { 
        this.exec = containerExec;
        this.context = nmContext;
        this.dispatcher = nmDispatcher;
        this.dirsHandler = nmDirsHandler;
        this.containerManager = nmContainerManager;
      }

# 四.serviceInit #

这里FileContext.getLocalFSFileContext(conf); 这句没看懂,为啥还要重新加载一下文件系统的配置 ??

@Override
      protected void serviceInit(Configuration conf) throws Exception { 
        try { 
          // 这个是干啥啊的 ???
    
          //TODO Is this required?
          FileContext.getLocalFSFileContext(conf);
        } catch (UnsupportedFileSystemException e) { 
          throw new YarnRuntimeException("Failed to start ContainersLauncher", e);
        }
        super.serviceInit(conf);
      }

# 五. handle #

对Container 执行各种操作

*  LAUNCH\_CONTAINER : 启动Container
 *  RELAUNCH\_CONTAINER : 重新启动Container
 *  RECOVER\_CONTAINER : 异常恢复Container/ 重新提交.
 *  RECOVER\_PAUSED\_CONTAINER : 恢复暂停的容器
 *  CLEANUP\_CONTAINER : 清理容器
 *  CLEANUP\_CONTAINER\_FOR\_REINIT : 清理容器,重新进行初始化
 *  SIGNAL\_CONTAINER : 向Container发送指令

@Override
      public void handle(ContainersLauncherEvent event) { 
        // TODO: ContainersLauncher launches containers one by one!!
    
        // 一个一个的启动Container
        Container container = event.getContainer();
    
        // 获取ContainerId
        ContainerId containerId = container.getContainerId();
    
        switch (event.getType()) { 
          // 启动Container
          case LAUNCH_CONTAINER:
            Application app =
              context.getApplications().get(
                  containerId.getApplicationAttemptId().getApplicationId());
    		// 构建ContainerLaunch 对象, 交由线程池执行.
            ContainerLaunch launch =
                new ContainerLaunch(context, getConfig(), dispatcher, exec, app,
                  event.getContainer(), dirsHandler, containerManager);
            containerLauncher.submit(launch);
            running.put(containerId, launch);
            break;
          
          case RELAUNCH_CONTAINER:
          // 重新启动Container
            app = context.getApplications().get(
                    containerId.getApplicationAttemptId().getApplicationId());
    
            ContainerRelaunch relaunch =
                new ContainerRelaunch(context, getConfig(), dispatcher, exec, app,
                    event.getContainer(), dirsHandler, containerManager);
            containerLauncher.submit(relaunch);
            running.put(containerId, relaunch);
            break;
    
          case RECOVER_CONTAINER:
            // 异常恢复Container/ 重新提交.
            app = context.getApplications().get(
                containerId.getApplicationAttemptId().getApplicationId());
            launch = new RecoveredContainerLaunch(context, getConfig(), dispatcher,
                exec, app, event.getContainer(), dirsHandler, containerManager);
            containerLauncher.submit(launch);
            running.put(containerId, launch);
            break;
          case RECOVER_PAUSED_CONTAINER:
            // 恢复暂停的容器
            app = context.getApplications().get(
                containerId.getApplicationAttemptId().getApplicationId());
            launch = new RecoverPausedContainerLaunch(context, getConfig(),
                dispatcher, exec, app, event.getContainer(), dirsHandler,
                containerManager);
            containerLauncher.submit(launch);
            break;
          case CLEANUP_CONTAINER:
            // 清理容器
            cleanup(event, containerId, true);
            break;
          case CLEANUP_CONTAINER_FOR_REINIT:
           // 清理容器,重新进行初始化
            cleanup(event, containerId, false);
            break;
          case SIGNAL_CONTAINER:
            // 向Container发送信息
            SignalContainersLauncherEvent signalEvent =
                (SignalContainersLauncherEvent) event;
            ContainerLaunch runningContainer = running.get(containerId);
            if (runningContainer == null) { 
              // Container not launched. So nothing needs to be done.
              LOG.info("Container " + containerId + " not running, nothing to signal.");
              return;
            }
    
            try { 
              runningContainer.signalContainer(signalEvent.getCommand());
            } catch (IOException e) { 
              LOG.warn("Got exception while signaling container " + containerId
                  + " with command " + signalEvent.getCommand());
            }
            break;
          case PAUSE_CONTAINER:
            // 暂停Container
            ContainerLaunch launchedContainer = running.get(containerId);
            if (launchedContainer == null) { 
              // Container not launched. So nothing needs to be done.
              return;
            }
    
            // Pause the container
            try { 
              launchedContainer.pauseContainer();
            } catch (Exception e) { 
              LOG.info("Got exception while pausing container: " +
                StringUtils.stringifyException(e));
            }
            break;
          case RESUME_CONTAINER:
            // 恢复Container
            ContainerLaunch launchCont = running.get(containerId);
            if (launchCont == null) { 
              // Container not launched. So nothing needs to be done.
              return;
            }
    
            // Resume the container.
            try { 
              launchCont.resumeContainer();
            } catch (Exception e) { 
              LOG.info("Got exception while resuming container: " +
                StringUtils.stringifyException(e));
            }
            break;
        }
      }

# 六.启动Container #

启动Container其实就是构建一个ContainerLaunch对象,然后交由containerLauncher线程池启动.

Application app =
              context.getApplications().get(
                  containerId.getApplicationAttemptId().getApplicationId());
    
            // 构建ContainerLaunch 对象, 交由线程池执行.
            ContainerLaunch launch =
                new ContainerLaunch(context, getConfig(), dispatcher, exec, app,
                  event.getContainer(), dirsHandler, containerManager);
            containerLauncher.submit(launch);
            running.put(containerId, launch);