一个宏大的分布式系统，各个组件间是如何协调工作的？组件是如何解耦的？线程运行如何更高效，缩小阻塞带来的低效问题？本节将对 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 isn't 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_INITpublic 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 tasksReceive JOB_INIT event, scheduling tasksReceive T_KILL event of task id job_20150723_11_task_0Receive T_KILL event of task id job_20150723_11_task_1Receive T_KILL event of task id job_20150723_11_task_2Receive T_KILL event of task id job_20150723_11_task_3Receive T_KILL event of task id job_20150723_11_task_4Receive T_KILL event of task id job_20150723_11_task_5Receive T_KILL event of task id job_20150723_11_task_6Receive T_KILL event of task id job_20150723_11_task_7Receive T_KILL event of task id job_20150723_11_task_8Receive T_KILL event of task id job_20150723_11_task_9Receive T_SCHEDULE event of task id job_20150723_11_task_0Receive T_SCHEDULE event of task id job_20150723_11_task_1Receive T_SCHEDULE event of task id job_20150723_11_task_2Receive T_SCHEDULE event of task id job_20150723_11_task_3Receive T_SCHEDULE event of task id job_20150723_11_task_4Receive T_SCHEDULE event of task id job_20150723_11_task_5Receive T_SCHEDULE event of task id job_20150723_11_task_6Receive T_SCHEDULE event of task id job_20150723_11_task_7Receive T_SCHEDULE event of task id job_20150723_11_task_8Receive T_SCHEDULE event of task id job_20150723_11_task_9
四、总结
本节介绍了 Yarn 的服务和事件库。
服务库标准了生命周期较长的服务型对象，定义了服务的四种状态、启停注册等要实现的办法，给出了繁多类型和组合类型服务的根本实现。
事件库的应用，解决了原始函数型调用的高耦合、阻塞低效等问题。可将一个大工作拆分成多个小工作，小工作变成不同的事件来触发解决。每一个事件处理器解决一种事件，并有一个地方异步调度器治理事件的收集和散发。
最初用一个简化的 MR ApplicationMaster 将事件库和服务库进行联合，更深领会如何在我的项目中将其联合应用。
学习过程中，写一个 demo 能更好的帮忙你了解常识。
参考文章：
《Hadoop 技术底细 - 深刻解析 Yarn 结构设计与实现原理》3.4 节