Hadoop服务库:
&">nbsp; YARN采用了基于服务的对象管理模型,主要特点有:
被服务化的对象分4个状态:NOTINITED,INITED,STARTED,STOPED 任何服务状态变化都可以触发另外一些动作 可通过组合方式对任意服务进行组合,统一管理
具体类请参见 org.apache.hadoop.service包下.核心接口是Service,抽象实现是AbstractService
YARN中,ResourceManager和NodeManager属于组合服务,内部包含多个单一和组合服务.以实现对内部多种服务的统一管理.
Hadoop事件库:
YARN采用事件驱动并发模型, 把各种逻辑抽象成事件,进入事件队列,然后由中央异步调度器负责传递给相应的事件调度器处理,或者调度器之间再传递,直至完成任务.
具体参见org.apache.hadoop.yarn.event.主要类和接口是:Event, AsyncDispatcher,EventHandler
按照惯例, 先给出一个Demo,然后顺着Demo研究代码实现.
示例我是直接抄<hadoop技术内幕>:
例子涉及如下几个模块:
Task
TaskType
Job
JobType
Dispatcher
package com.demo1;
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;
/**
* Created by yang on 2014/8/25.
*/
public class SimpleService extends CompositeService {
private Dispatcher dispatcher;
private String jobID;
private int taskNum;
private String[] taskIDs;
public SimpleService(String name, String jobID, int taskNum) {
super(name);
this.jobID = jobID;
this.taskNum = taskNum;
this.taskIDs = new String[taskNum];
for (int i = 0; i < taskNum; i++) {
taskIDs[i] = new String(jobID + "_task_" + i);
}
}
public Dispatcher getDispatcher() {
return dispatcher;
}
public void serviceInit(Configuration conf) throws Exception {
dispatcher = new AsyncDispatcher();
dispatcher.register(JobEventType.class, new JobEventDIspatcher());
dispatcher.register(TaskEventType.class, new TaskEventDIspatcher());
addService((Service)dispatcher);
super.serviceInit(conf);
}
private class JobEventDIspatcher implements EventHandler<JobEvent> {
@Override
public void handle(JobEvent jobEvent) {
if (jobEvent.getType() == JobEventType.JOB_KILL) {
System.out.println("JOB KILL EVENT");
for (int i = 0; i < taskNum; i++) {
dispatcher.getEventHandler().handle(new TaskEvent(taskIDs[i], TaskEventType.T_KILL));
}
} else if (jobEvent.getType() == JobEventType.JOB_INIT) {
System.out.println("JOB INIT EVENT");
for (int i = 0; i < taskNum; i++) {
dispatcher.getEventHandler().handle(new TaskEvent(taskIDs[i], TaskEventType.T_SCHEDULE));
}
}
}
}
private class TaskEventDIspatcher implements EventHandler<TaskEvent> {
@Override
public void handle(TaskEvent taskEvent) {
if (taskEvent.getType() == TaskEventType.T_KILL) {
System.out.println("TASK KILL EVENT" + taskEvent.getTaskID());
} else if (taskEvent.getType() == TaskEventType.T_SCHEDULE) {
System.out.println("TASK INIT EVENT" + taskEvent.getTaskID());
}
}
}
}
测试程序
package com.demo1;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
/**
* Created by yang on 2014/8/25.
*/
public class Test {
public static void main(String[] args) throws Exception {
String jobID="job_1";
SimpleService ss = new SimpleService("test",jobID,5);
YarnConfiguration config = new YarnConfiguration(new Configuration());
ss.serviceInit(config);
ss.init(config);
ss.start();
ss.getDispatcher().getEventHandler().handle(new JobEvent(jobID,JobEventType.JOB_KILL));
ss.getDispatcher().getEventHandler().handle(new JobEvent(jobID,JobEventType.JOB_KILL));
}
}
不出意外的话,运行结果应该类似:
14/08/25 16:02:20 INFO event.AsyncDispatcher: Registering class com.yws.demo1.JobEventType for class com.yws.demo1.SimpleService$JobEventDIspatcher
14/08/25 16:02:42 INFO event.AsyncDispatcher: Registering class com.yws.demo1.TaskEventType for class com.yws.demo1.SimpleService$TaskEventDIspatcher
14/08/25 16:02:54 INFO event.AsyncDispatcher: Registering class com.yws.demo1.JobEventType for class com.yws.demo1.SimpleService$JobEventDIspatcher
14/08/25 16:03:03 INFO event.AsyncDispatcher: Registering class com.yws.demo1.TaskEventType for class com.yws.demo1.SimpleService$TaskEventDIspatcher
JOB KILL EVENT
JOB KILL EVENT
TASK KILL EVENTjob_1_task_0
TASK KILL EVENTjob_1_task_1
TASK KILL EVENTjob_1_task_2
TASK KILL EVENTjob_1_task_3
TASK KILL EVENTjob_1_task_4
TASK KILL EVENTjob_1_task_0
TASK KILL EVENTjob_1_task_1
TASK KILL EVENTjob_1_task_2
TASK KILL EVENTjob_1_task_3
TASK KILL EVENTjob_1_task_4
我们开始分析:
所谓的Task,Job,其实是按业务逻辑划分的, 他们都继承AbstractEvent类.
SimpleService是一个组合服务,里面放了EventHandler和Dispatcher
从Test开始,看看Service是如何创建的
构造函数比较简单,就是将一个job拆分成taskNum个Task
ss.serviceInit(config);做了什么呢:
创建一个中央事件调度器: AsyncDispatcher(具体实现我们在后文分析)
并把Job和Task的Event及2者对应的EventHandler注册到调度器中.
这里就是初始化和启动服务了.最后2行就是模拟2个事件的JOB_KILL事件.
我们进到ss.getDispatcher().getEventHandler(),发现他其实是创建一个GenericEventHandler
这个handler干什么是呢?
就是把
塞到BlockingQueue<Event> eventQueue; 中.
不知道你发现没有, 这个方法仅仅是一个入队操作啊. 那具体调用JobEventDIspatcher.handler是在什么地方呢?
这时联想到之前不是有个中央调度器嘛, AsyncDispatcher, Line 80行, 他创建了一个线程,并不断的从之前说的EventQueue中不断的取Event,然后执行,这里的执行也就是调用了具体的handler了
就这样一个基于事件驱动的程序这么完成了.
按照hadoop 早起版本中, 业务逻辑之间是通过函数调用方式实现的,也就是串行的. 现在基于事件驱动后,大大提高了并发性.很值得我们学习.
来张全家福:
HandlerThread就是前文说的那个隐藏线程. EventHandler会产生一些新的Event,然后又重新进入队列.循环.