我要评论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;
}
}
// task事件类型定义
public enum taskeventtype {
t_kill,
t_schedule
}
public class jobevent extends abstractevent<jobeventtype> {
private string jobid;
public jobevent (string jobid, jobeventtype type) {
super(type);
this.jobid = jobid;
}
}
//job事件类型定义
public enum jobeventtype {
job_kill,
job_init,
job_start
}
@suppresswarnings("unchecked")
public class simplemrappmaster extends compositeservice {
private dispatcher dispatcher; //中央异步调度器
private string jobid;
private int tasknumber; //该作业包含的任务数目
private string[] taskids; //该作业内部包含的所有任务
public simplemrappmaster (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 (jobid + "_task_" + i);
}
}
public void serviceinit (final configuration conf) throws exception {
dispatcher = new asyncdispatcher(); //定义一个中央异步调度器
//分别注册job和task事件调度器
dispatcher.register(jobeventtype.class, new jobeventdispatcher());
dispatcher.register(taskeventtype.class, new taskeventdispatcher());
addservice((service)dispatcher);
super.serviceinit(conf);
}
public dispatcher getdispatcher() {
return dispatcher;
}
private class jobeventdispatcher implements eventhandler<jobevent> {
@override
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> {
@override
public void handler (taskevent event) {
if (event.gettype() == taskeventtype.t_kill) {
system.out.println("receive t_kill event of task" + event.gettaskid());
} else if (event.gettype() == taskeventtype.t_schedule) {
system.out.println("receive t_schedule of task" + event.gettaskid());
}
}
}
}
@suppresswarnings("unchecked")
public class simplemrappmastertest {
public static void main (string[] args) throws exception {
string jobid = "job_20131215_12";
simplemrappmaster appmaster = new simplemrappmaster("simple mrappmaster", jobid, 5);
yarnconfiguration conf = new yarnconfiguration(new configuration());
appmaster.serviceinit(conf);
appmaster.servicestart();
appmaster.getdispatcher().geteventhandler().handle(new jobevent(jobid, jobeventtype.job_kill));
appmaster.getdispatcher().geteventhandler().handle(new jobevent(jobid, jobeventtype.job_init));
}
}
如对本文有疑问， 点击进行留言回复！！
HBase Filter 过滤器之FamilyFilter详解
去 HBase，Kylin on Parquet 性能表现如何？
网友评论