1.概述
在《Hadoop2源码分析－YARN RPC 示例介绍》一文当中，给大家介绍了YARN 的 RPC 机制，以及相关代码的演示，今天我们继续去学习 YARN 的服务库和事件库，分享目录如下所示：
- 服务库和事件库介绍
- 使用示例
- 截图预览
下面开始今天的内容分享。
2.服务库和事件库介绍
2.1服务库
YARN对于生命周期较长的对象使用服务的对象模型进行管理，主要特点如下：
- 用于被服务化的对象包含4个状态，他们分别是：被创建、已初始化、已启动和已停止。源代码地址在 org.apache.hadoop.service 的 Service 接口中，内容如下所示：
public enum STATE {
/** Constructed but not initialized */
NOTINITED(0, "NOTINITED"),
/** Initialized but not started or stopped */
INITED(1, "INITED"),
/** started and not stopped */
STARTED(2, "STARTED"),
/** stopped. No further state transitions are permitted */
STOPPED(3, "STOPPED");
/**
* An integer value for use in array lookup and JMX interfaces.
* Although {@link Enum#ordinal()} could do this, explicitly
* identify the numbers gives more stability guarantees over time.
*/
private final int value;
/**
* A name of the state that can be used in messages
*/
private final String statename;
private STATE(int value, String name) {
this.value = value;
this.statename = name;
}
/**
* Get the integer value of a state
* @return the numeric value of the state
*/
public int getValue() {
return value;
}
/**
* Get the name of a state
* @return the state's name
*/
@Override
public String toString() {
return statename;
}
}
public abstract class AbstractService implements Service {
// ......
}
通过阅读代码，我们可以看出，服务的对象它实现了接口Service，并定义了最基本的服务状态：创建、初始化、启动以及停止。对于 AbstractService 类来说，它实现了 Service 接口。
- 任何服务状态的变化都可以触发其他的动作，例如：
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);
}
}
}
}
这里，我们会去触发一个监听动作，全局监听状态的改变，异常的捕捉监听等。
- 可以通过组合的方式进行服务组合，这样做的好处是便于统一去管理：在 YARN 中，如果是非组合服务，可以直接继承 AbstractService 类，否则需继承 CompositeService。
2.2事件库
在 YARN 中，核心服务其本质就是一个*异步调度器，包含有ResourceManager、 NodeManager、MRAppMaster等内容，YARN 事件与事件处理器的关系在
org.apache.hadoop.yarn.event 中。在使用 YARN 事件库的时候，需要先定义一个*异步调度器 AsyncDispatcher，它负责事件的处理与转发，然后我们根据实际业务需求定义一系列事件 Event 与事件处理器 EventHandler，并将事件注册到*异步调度器中用于完成事件统一管理和应用调度。流程如下图所示：
3.使用示例
接下来，我们编写示例代码，去代码中理解这部分流程。
- 首先是 JMRAppMaster 类：
package cn.hadoop.task.exec; import org.apache.hadoop.conf.Configuration;
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; import cn.hadoop.task.CompositeService;
import cn.hadoop.task.JobEvent;
import cn.hadoop.task.JobEventType;
import cn.hadoop.task.TaskEvent;
import cn.hadoop.task.TaskEventType; /**
* @Date Jul 22, 2015
*
* @Author dengjie
*
* @Note TODO
*/
public class JMRAppMaster extends CompositeService {
private Dispatcher dispatcher; // AsyncDispatcher
private String jobID;
private int taskNumber; // include numbers
private String[] taskIDs; // include all task public JMRAppMaster(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] = new String(this.jobID + "_task_" + i);
}
} public void serviceInit(Configuration conf) throws Exception {
dispatcher = new AsyncDispatcher();// default a 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 Dispatcher getDispatcher() {
return dispatcher;
} private class JobEventDispatcher implements EventHandler<JobEvent> { @SuppressWarnings("unchecked")
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());
}
}
}
}
另外，还需要添加一些其他类，这些类以来可以在 Hadoop 源码工程中找到，这里就不贴代码了，大家可以到 Hadoop 工程的源码中找到对应的类，相关类名如下图所示：
接下来是一个测试类，去测试一下我们所编写的示例流程。
JMRAppMasterTest类：
package cn.hadoop.rpc.test.yarn.task; import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.yarn.conf.YarnConfiguration; import cn.hadoop.task.JobEvent;
import cn.hadoop.task.JobEventType;
import cn.hadoop.task.exec.JMRAppMaster; /**
* @Date Jul 22, 2015
*
* @Author dengjie
*
* @Note TODO
*/
public class JMRAppMasterTest {
@SuppressWarnings({ "unchecked", "resource" })
public static void main(String[] args) {
String jobID = "job_20150723_11";
JMRAppMaster appMaster = new JMRAppMaster("Simple 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));
}
}
4.截图预览
在编写完成相关流程代码后，我们运行代码来观察整个流程，截图如下所示：
5.总结
在编写这部分流程代码时，可以参考 Hadoop YARN 部分的工程源码，通过运行调试代码，掌握对事件库和服务库的流程，以及它们的工作机制。另外，在编写的过程当中，最好将源码的文件引入到自己的工程，不要单独使用 JAR 包的方式导入，由于我们是独立运行某个模块，需要改动源代码的函数访问权限，若是直接引入 JAR 包地址，会导致函数修饰权限问题而不能运行，这里大家在运行调试的时候注意即可。
6.结束语
这篇博客就和大家分享到这里，如果大家在研究学习的过程当中有什么问题，可以加群进行讨论或发送邮件给我，我会尽我所能为您解答，与君共勉！