【深入浅出 Yarn 架构与实现】2-4 Yarn 基础库 - 状态机库

当一个服务拥有太多处理逻辑时，会导致代码结构异常的混乱，很难分辨一段逻辑是在哪个阶段发挥作用的。
这时就可以引入状态机模型，帮助代码结构变得清晰。
一、状态机库概述

一）简介

状态机由一组状态组成：
【初始状态 -> 中间状态 -> 最终状态】。
在一个状态机中，每个状态会接收一组特定的事件，根据事件类型进行处理，并转换到下一个状态。当转换到最终状态时则退出。
二）状态转换方式

状态间转换会有下面这三种类型：

三）Yarn 状态机类

在 Yarn 中提供了一个工厂类 StateMachineFactory 来帮助定义状态机。如何使用，我们直接写个 demo。

二、案例 demo

在上一篇文章《Yarn 服务库和事件库》案例基础上进行扩展，增加状态机库的内容。如果还不了解服务库和事件库的同学，建议先学习下上一篇文章。
案例已上传至 github，有帮助可以点个 ⭐️
https://github.com/Simon-Ace/hadoop-yarn-study-demo/tree/master/state-demo
一）状态机实现

状态机实现，可以直接嵌入到上篇文章中的 AsyncDispatcher使用。
这里仅给出状态机JobStateMachine以及各种事件处理的代码。完整的代码项目执行，请到 github demo 中查看。

1. import com.shuofxz.event.JobEvent;
2. import com.shuofxz.event.JobEventType;
3. import org.apache.hadoop.yarn.event.EventHandler;
4. import org.apache.hadoop.yarn.state.*;
6. import java.util.EnumSet;
7. import java.util.concurrent.locks.Lock;
8. import java.util.concurrent.locks.ReadWriteLock;
9. import java.util.concurrent.locks.ReentrantReadWriteLock;
11. /*
12. * 可参考 Yarn 中实现的状态机对象：
13. * ResourceManager 中的 RMAppImpl、RMApp- AttemptImpl、RMContainerImpl 和 RMNodeImpl，
14. * NodeManager 中 的 ApplicationImpl、 ContainerImpl 和 LocalizedResource，
15. * MRAppMaster 中的 JobImpl、TaskImpl 和 TaskAttemptImpl 等
16. * */
17. @SuppressWarnings({"rawtypes", "unchecked"})
18. public class JobStateMachine implements EventHandler<JobEvent> {
19.     private final String jobID;
20.     private EventHandler eventHandler;
21.     private final Lock writeLock;
22.     private final Lock readLock;
24.     // 定义状态机
25.     protected static final StateMachineFactory<JobStateMachine, JobStateInternal,
26.             JobEventType, JobEvent>
27.             stateMachineFactory = new StateMachineFactory<JobStateMachine, JobStateInternal, JobEventType, JobEvent>(JobStateInternal.NEW)
28.             .addTransition(JobStateInternal.NEW, JobStateInternal.INITED, JobEventType.JOB_INIT, new InitTransition())
29.             .addTransition(JobStateInternal.INITED, JobStateInternal.SETUP, JobEventType.JOB_START, new StartTransition())
30.             .addTransition(JobStateInternal.SETUP, JobStateInternal.RUNNING, JobEventType.JOB_SETUP_COMPLETED, new SetupCompletedTransition())
31.             .addTransition(JobStateInternal.RUNNING, EnumSet.of(JobStateInternal.KILLED, JobStateInternal.SUCCEEDED), JobEventType.JOB_COMPLETED, new JobTasksCompletedTransition())
32.             .installTopology();
34.     private final StateMachine<JobStateInternal, JobEventType, JobEvent> stateMachine;
36.     public JobStateMachine(String jobID, EventHandler eventHandler) {
37.         this.jobID = jobID;
39.         // 多线程异步处理，state 有可能被同时读写，使用读写锁来避免竞争
40.         ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
41.         this.readLock = readWriteLock.readLock();
42.         this.writeLock = readWriteLock.writeLock();
44.         this.eventHandler = eventHandler;
45.         stateMachine = stateMachineFactory.make(this);
46.     }
48.     protected StateMachine<JobStateInternal, JobEventType, JobEvent> getStateMachine() {
49.         return stateMachine;
50.     }
52.     public static class InitTransition implements SingleArcTransition<JobStateMachine, JobEvent> {
53.         @Override
54.         public void transition(JobStateMachine jobStateMachine, JobEvent jobEvent) {
55.             System.out.println("Receiving event " + jobEvent);
56.             // do something...
57.             // 完成后发送新的 Event —— JOB_START
58.             jobStateMachine.eventHandler.handle(new JobEvent(jobStateMachine.jobID, JobEventType.JOB_START));
59.         }
60.     }
62.     public static class StartTransition implements SingleArcTransition<JobStateMachine, JobEvent> {
63.         @Override
64.         public void transition(JobStateMachine jobStateMachine, JobEvent jobEvent) {
65.             System.out.println("Receiving event " + jobEvent);
66.             jobStateMachine.eventHandler.handle(new JobEvent(jobStateMachine.jobID, JobEventType.JOB_SETUP_COMPLETED));
67.         }
68.     }
70.     public static class SetupCompletedTransition implements SingleArcTransition<JobStateMachine, JobEvent> {
71.         @Override
72.         public void transition(JobStateMachine jobStateMachine, JobEvent jobEvent) {
73.             System.out.println("Receiving event " + jobEvent);
74.             jobStateMachine.eventHandler.handle(new JobEvent(jobStateMachine.jobID, JobEventType.JOB_COMPLETED));
75.         }
76.     }
78.     public static class JobTasksCompletedTransition implements MultipleArcTransition<JobStateMachine, JobEvent, JobStateInternal> {
79.         @Override
80.         public JobStateInternal transition(JobStateMachine jobStateMachine, JobEvent jobEvent) {
81.             System.out.println("Receiving event " + jobEvent);
83.             // 这是多结果状态部分，因此需要人为制定后续状态
84.             // 这里整个流程结束，设置一下对应的状态
85.             boolean flag = true;
86.             if (flag) {
87.                 return JobStateInternal.SUCCEEDED;
88.             } else {
89.                 return JobStateInternal.KILLED;
90.             }
91.         }
92.     }
94.     @Override
95.     public void handle(JobEvent jobEvent) {
96.         try {
97.             // 注意这里为了避免静态条件，使用了读写锁
98.             writeLock.lock();
99.             JobStateInternal oldState = getInternalState();
100.             try {
101.                 getStateMachine().doTransition(jobEvent.getType(), jobEvent);
102.             } catch (InvalidStateTransitionException e) {
103.                 System.out.println("Can't handle this event at current state!");
104.             }
105.             if (oldState != getInternalState()) {
106.                 System.out.println("Job Transitioned from " + oldState + " to " + getInternalState());
107.             }
109.         } finally {
110.             writeLock.unlock();
111.         }
112.     }
114.     public JobStateInternal getInternalState() {
115.         readLock.lock();
116.         try {
117.             return getStateMachine().getCurrentState();
118.         } finally {
119.             readLock.unlock();
120.         }
121.     }
123.     public enum JobStateInternal {
124.         NEW,
125.         SETUP,
126.         INITED,
127.         RUNNING,
128.         SUCCEEDED,
129.         KILLED
130.     }
131. }

复制代码

二）状态机可视化

hadoop 中提供了状态机可视化的工具类 VisualizeStateMachine.java，可以拷贝到我们的工程中使用。
根据提示，运行需要三个参数：

1. Usage: %s <GraphName> <class[,class[,...]]> <OutputFile>%n

复制代码

运行后会在项目根目录生成图文件 jsm.gv。
需要使用 graphviz工具将 gv 文件转换成 png 文件：

1. # linux 安装
2. yum install graphviz
4. # mac 安装
5. brew install graphviz

复制代码

转换：

1. dot -Tpng jsm.gv > jsm.png

复制代码

可视化状态机展示：

再使用这个工具对 Yarn 中的 Application 状态进行展示：

三）如果不用状态机库

【思考】
如果不用状态机，代码结构会是什么样呢？
下面这样的代码，如果要增加或修改逻辑可能就是很痛苦的一件事情了。

1. // 一堆的函数调用
3. // 一堆的 if 嵌套
5. // 或者 switch case

复制代码

三、总结

本节对 Yarn 状态机库进行了介绍。实际使用时会结合事件库、服务库一同使用。
状态机库的使用帮助代码结构更加的清晰，新增状态处理逻辑只需要增加一个状态类别，或者增加一个方法处理对应类型的事件即可。将整个处理逻辑进行了拆分，便于编写和维护。
参考文章：
源码|Yarn的事件驱动模型与状态机

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！