Yarn的服务库和事件库使用方法

事件类型定义：

package org.apache.hadoop.event;
public enum JobEventType {
        JOB_KILL,
        JOB_INIT,
        JOB_START
}

package org.apache.hadoop.event;

public enum TaskEventType {
    T_KILL,
    T_SCHEDULE
}

事件定义：

package org.apache.hadoop.event;

import org.apache.hadoop.yarn.event.AbstractEvent;
public class JobEvent extends AbstractEvent<JobEventType> {

    private String jobID;

    public JobEvent(JobEventType type,String jobID) {
        super(type);
        this.jobID=jobID;
    }
    public String getJobID() {
        return jobID;
    }

}

package org.apache.hadoop.event;

import org.apache.hadoop.yarn.event.AbstractEvent;

public class TaskEvent extends AbstractEvent<TaskEventType> {

    private String taskID; //TASkID

    public TaskEvent(TaskEventType type,String taskID) {
        super(type);
        this.taskID=taskID;
    }

    public String getTaskID() {
        return taskID;
    }

}

简单服务定义：

package org.apache.hadoop.event;

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;

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;
        this.taskIDS=new String[taskNumber];
        for(int i=0;i<taskNumber;i++){
            this.taskIDS[i]=new String(jobID+"_task_"+i);
        }
    }

    @Override
    protected void serviceInit(Configuration conf) throws Exception {

        dispatcher=new AsyncDispatcher();

        dispatcher.register(JobEventType.class, new JobEventHandller());
        dispatcher.register(TaskEventType.class, new TaskEventHandller());
        addService((Service)dispatcher);

        super.serviceInit(conf);
    }

    public Dispatcher getDispatcher(){
        return dispatcher;
    }

    private class JobEventHandller implements EventHandler<JobEvent>{

        @Override
        public void handle(JobEvent event) {

            //若收到 杀死  作业 事件
            if(event.getType() == JobEventType.JOB_KILL){
                System.out.println("收到 杀死作业事件   ，要 杀掉作业"+event.getJobID()+"下的所有任务");

                for(int i=0;i<=taskNumber;i++){
                    dispatcher.getEventHandler().handle(new TaskEvent(TaskEventType.T_KILL, taskIDS[i]));
                }

            }else if(event.getType()== JobEventType.JOB_INIT){
                System.out.println("收到 启动作业事件   ，要启动 作业"+event.getJobID()+"下的所有任务");
                for(int i=0;i<=taskNumber;i++){
                    dispatcher.getEventHandler().handle(new TaskEvent(TaskEventType.T_SCHEDULE, taskIDS[i]));
                }
            }
        }
    }

    private class TaskEventHandller implements EventHandler<TaskEvent>{

        @Override
        public void handle(TaskEvent event) {
            if(event.getType()==TaskEventType.T_KILL){
                System.out.println("收到杀死任务命令，开始杀死任务"+event.getTaskID());
            }else if(event.getType()==TaskEventType.T_SCHEDULE){
                System.out.println("收到启动任务命令，开始启动任务"+event.getTaskID());
            }
        }
    }

}

测试程序定义：

package org.apache.hadoop.event;

import static org.junit.Assert.*;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.junit.Test;

/**
 * 自己写的 关于事件库 和 服务 库的使用
 * @author joqk
 *
 */
public class SimpleMRAppMasterTest {

    @Test
    public void test() throws Exception {

            String jobID="job_20140912_01";
            SimpleMRAppMaster appMaster=new SimpleMRAppMaster("作业测试", jobID, 10);
            YarnConfiguration conf = new YarnConfiguration(new Configuration());

            appMaster.serviceInit(conf);

            appMaster.start();

            appMaster.getDispatcher().getEventHandler().handle(new JobEvent(JobEventType.JOB_INIT, jobID));

    }

}