yarn框架第一版是( )年发布（【深入浅出 Yarn 架构与实现】2-3 Yarn 基础库 – 服务库与事件库）

一个庞大的分布式系统
  
  
  
  
  
 ，各个组件间是如何协调工作的？组件是如何解耦的？线程运行如何更高效  
  
  
  
  
  ，减少阻塞带来的低效问题？本节将对 Yarn 的服务库和事件库进行介绍
 


 


 


 


 


 

，看看 Yarn 是如何解决这些问题的            。

一
  
  
  
  
  
 、服务库

一）简介

对于生命周期较长的对象
 
 
 
 
 
，Yarn 采用基于服务的模型对其进行管理   
   
   
   
   
   ，有以下几个特点：

基于状态管理：分为 4 个状态：NOTINITED(被创建)  
  
  
  
  
  、INITED(已初始化)
 


 


 


 


 


 

、 STARTED(已启动)
 
 
 
 
 
、STOPPED(已停止)
  
 
  
 
  
 
  
 
  
 
  
。 服务状态的变化会触发其他的操作










。 可通过组合的方式对服务进行组合 
  
  
  
  
  
 。

二）源码简析

源代码地址在 hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/service 的 Service 接口中
  
  
  
  
  
 。

其中定义了服务的四个状态
 

 

 

 

 

 
，以及需要实现的状态转换   
   
   
   
   
   、获取信息
 

 

 

 

 

 
、注册等方法
 


 


 


 


 


 
。 public interface Service extends Closeable { public enum STATE { NOTINITED(0, "NOTINITED"), INITED(1, "INITED"), STARTED(2, "STARTED"), STOPPED(3, "STOPPED"); } void init(Configuration config); void start(); void stop(); void close() throws IOException; void registerServiceListener(ServiceStateChangeListener listener); // ......

抽象类 AbstractService 实现了 Service 接口

 

 

 

 

 

，提供了基础的 Service 实现            ，非组合服务直接继承这个抽象类再开发即可 
 
 
 
 
 
。

public abstract class AbstractService implements Service { // 以 start 实现为例
 
 

 
 

 
 

 
 

 
 

 
 ，执行后会触发其他的操作 public void start() { if (isInState(STATE.STARTED)) { return; } //enter the started state synchronized (stateChangeLock) { if (stateModel.enterState(STATE.STARTED) != STATE.STARTED) { try { startTime = System.currentTimeMillis(); serviceStart(); if (isInState(STATE.STARTED)) { //if the service started (and isnt now in a later state), notify if (LOG.isDebugEnabled()) { LOG.debug("Service " + getName() + " is started"); } notifyListeners(); } } catch (Exception e) { noteFailure(e); ServiceOperations.stopQuietly(LOG, this); throw ServiceStateException.convert(e); } } } } // ......

对于组合类的服务如 ResourceManager

 

 

 

 

 

、NodeManager 等











，需要继承 CompositeService
   
   
   
   
   
  。其中会有对组合服务的逻辑处理
 

 

 

 

 

 
。

public List<Service> getServices() { synchronized (serviceList) { return new ArrayList<Service>(serviceList); } } protected void addService(Service service) { if (LOG.isDebugEnabled()) { LOG.debug("Adding service " + service.getName()); } synchronized (serviceList) { serviceList.add(service); } }

二            、事件库

传统函数式调用的问题：

整个执行过程是串行
 
 

 
 

 
 

 
 

 
 

 
 、同步进行的 

 

 

 

 

 
。调用另一个函数的时候                 ，需要等待函数执行完毕 
 
 
 
 
 
 
 
 
 
 
 ，才会继续往下走            。示意图如下：

为了解决函数式调用的问题

















，可使用「事件驱动」的编程模型

 
 

 
 

 
 

 
 

 
 

 
。

所有对象都被抽象成事件处理器 事件处理器之间通过事件相关联 每种事件处理器处理一种事件 根据需要会触发另一种事件 每类事件的处理可分割为多个步骤           ，用有限状态机表示 重要的是有一个「中央异步调度器（AsyncDispatcher）」  
 
  
 
  
 
  
 
  
 
  
 ，负责对待处理事件的收取和分发

示意图如下：

通过以上的方式











，可以使程序有低耦合高内聚的特点
  
  
  
  
  
 ，各个模块仅需完成各自的功能  
  
  
  
  
  ，同时提高了执行效率
 


 


 


 


 


 

，把拆分的操作通过事件的方式发送出去即可










。

三











、服务库和事件库使用案例

本节将实现一个简化版的 MapReduce ApplicationMaster
 
 
 
 
 
，帮助了解 service 和 event 的使用方法                  。

与 MR 类似   
   
   
   
   
   ，一个 job 将被分为多个 task 执行
 
 
 
 
 
 
 
 
 
 
 
。因此涉及 job 和 task 两种对象的事件
















。并有一个 AsyncDispatcher 处理调度            。

案例已上传至 github
 

 

 

 

 

 
，有帮助可以点个 ⭐️

https://github.com/Simon-Ace/hadoop-yarn-study-demo/tree/master/service-event-demo

一）事件部分

参考 hadoop 源码中 Task 和 Job Event 的实现

 

 

 

 

 

，进行一些简化
  
 
  
 
  
 
  
 
  
 
  
。

1                 、task public enum TaskEventType { //Producer:Client, Job T_KILL, //Producer:Job T_SCHEDULE } public class TaskEvent extends AbstractEvent<TaskEventType> { private String taskID; public TaskEvent(String taskID, TaskEventType type) { super(type); this.taskID = taskID; } public String getTaskID() { return taskID; } }

2 
 
 
 
 
 
 
 
 
 
 
 、job

public enum JobEventType { //Producer:Client JOB_KILL, //Producer:MRAppMaster JOB_INIT } public class JobEvent extends AbstractEvent<JobEventType> { private String jobID; public JobEvent(String jobID, JobEventType type) { super(type); this.jobID = jobID; } public String getJobId() { return jobID; } }

二）事件调度器

定义和注册 EventDispatcher service 初始化和启动方法 import com.shuofxz.event.JobEvent; import com.shuofxz.event.JobEventType; import com.shuofxz.event.TaskEvent; import com.shuofxz.event.TaskEventType; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.service.CompositeService; import org.apache.hadoop.service.Service; import org.apache.hadoop.yarn.event.AsyncDispatcher; import org.apache.hadoop.yarn.event.Dispatcher; import org.apache.hadoop.yarn.event.EventHandler; @SuppressWarnings("unchecked") public class MyMRAppMaster extends CompositeService { private Dispatcher dispatcher; // AsyncDispatcher private String jobID; private int taskNumber; // 一个 job 包含的 task 数 private String[] taskIDs; public MyMRAppMaster(String name, String jobID, int taskNumber) { super(name); this.jobID = jobID; this.taskNumber = taskNumber; taskIDs = new String[taskNumber]; for (int i = 0; i < taskNumber; i++) { taskIDs[i] = this.jobID + "_task_" + i; } } public void serviceInit(Configuration conf) throws Exception { dispatcher = new AsyncDispatcher(); dispatcher.register(JobEventType.class, new JobEventDispatcher()); // register a job dispatcher.register(TaskEventType.class, new TaskEventDispatcher()); // register a task addService((Service) dispatcher); super.serviceInit(conf); } public void serviceStart() throws Exception { super.serviceStart(); } public Dispatcher getDispatcher() { return dispatcher; } private class JobEventDispatcher implements EventHandler<JobEvent> { public void handle(JobEvent event) { if (event.getType() == JobEventType.JOB_KILL) { System.out.println("Receive JOB_KILL event, killing all the tasks"); for (int i = 0; i < taskNumber; i++) { dispatcher.getEventHandler().handle(new TaskEvent(taskIDs[i], TaskEventType.T_KILL)); } } else if (event.getType() == JobEventType.JOB_INIT) { System.out.println("Receive JOB_INIT event, scheduling tasks"); for (int i = 0; i < taskNumber; i++) { dispatcher.getEventHandler().handle(new TaskEvent(taskIDs[i], TaskEventType.T_SCHEDULE)); } } } } private class TaskEventDispatcher implements EventHandler<TaskEvent> { public void handle(TaskEvent event) { if (event.getType() == TaskEventType.T_KILL) { System.out.println("Receive T_KILL event of task id " + event.getTaskID()); } else if (event.getType() == TaskEventType.T_SCHEDULE) { System.out.println("Receive T_SCHEDULE event of task id " + event.getTaskID()); } } } }

三）测试程序

生成一个新的 job 触发事件 JOB_KILL 和 JOB_INIT public class MyMRAppMasterTest { public static void main(String[] args) { String jobID = "job_20221011_99"; MyMRAppMaster appMaster = new MyMRAppMaster("My MRAppMaster Test", jobID, 10); YarnConfiguration conf = new YarnConfiguration(new Configuration()); try { appMaster.serviceInit(conf); appMaster.serviceStart(); } catch (Exception e) { e.printStackTrace(); } appMaster.getDispatcher().getEventHandler().handle(new JobEvent(jobID, JobEventType.JOB_KILL)); appMaster.getDispatcher().getEventHandler().handle(new JobEvent(jobID, JobEventType.JOB_INIT)); } }

输出结果：

Receive JOB_KILL event, killing all the tasks Receive JOB_INIT event, scheduling tasks Receive T_KILL event of task id job_20150723_11_task_0 Receive T_KILL event of task id job_20150723_11_task_1 Receive T_KILL event of task id job_20150723_11_task_2 Receive T_KILL event of task id job_20150723_11_task_3 Receive T_KILL event of task id job_20150723_11_task_4 Receive T_KILL event of task id job_20150723_11_task_5 Receive T_KILL event of task id job_20150723_11_task_6 Receive T_KILL event of task id job_20150723_11_task_7 Receive T_KILL event of task id job_20150723_11_task_8 Receive T_KILL event of task id job_20150723_11_task_9 Receive T_SCHEDULE event of task id job_20150723_11_task_0 Receive T_SCHEDULE event of task id job_20150723_11_task_1 Receive T_SCHEDULE event of task id job_20150723_11_task_2 Receive T_SCHEDULE event of task id job_20150723_11_task_3 Receive T_SCHEDULE event of task id job_20150723_11_task_4 Receive T_SCHEDULE event of task id job_20150723_11_task_5 Receive T_SCHEDULE event of task id job_20150723_11_task_6 Receive T_SCHEDULE event of task id job_20150723_11_task_7 Receive T_SCHEDULE event of task id job_20150723_11_task_8 Receive T_SCHEDULE event of task id job_20150723_11_task_9

四

















、总结

本节介绍了 Yarn 的服务和事件库










。

服务库规范了生命周期较长的服务型对象            ，定义了服务的四种状态           、启停注册等要实现的方法
 
 

 
 

 
 

 
 

 
 

 
 ，给出了单一类型和组合类型服务的基本实现 
  
  
  
  
  
 。

事件库的使用











，解决了原始函数型调用的高耦合  
 
  
 
  
 
  
 
  
 
  
 、阻塞低效等问题
  
  
  
  
  
 。可将一个大任务拆分成多个小任务                 ，小任务变成不同的事件来触发处理
 


 


 


 


 


 
。每一个事件处理器处理一种事件 
 
 
 
 
 
 
 
 
 
 
 ，并有一个中央异步调度器管理事件的收集和分发 
 
 
 
 
 
。

最后用一个简化的 MR ApplicationMaster 将事件库和服务库进行结合

















，更深体会如何在项目中将其结合使用
   
   
   
   
   
  。

学习过程中           ，写一个 demo 能更好的帮助你理解知识
 

 

 

 

 

 
。

参考文章：

《Hadoop 技术内幕 - 深入解析 Yarn 结构设计与实现原理》3.4 节