状态机与状态机的设计

状态机就是状态转移图。

关于状态机的一个极度确切的描述是它是一个有向图形，由一组节点和一组相应的转移函数组成。状态机通过响应一系列事件而“运行”。每个事件都在属于“当前” 节点的转移函数的控制范围内，其中函数的范围是节点的一个子集。函数返回“下一个”（也许是同一个）节点。这些节点中至少有一个必须是终态。当到达终态， 状态机停止。

状态机可归纳为4个要素，即现态、条件、动作、次态。这样的归纳，主要是出于对状态机的内在因果关系的考虑。“现态”和“条件”是因，“动作”和“次态”是果。

我们定义一个jobEvent

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;
    }

}

JobEventType：

package org.apache.hadoop.state;

public enum JobEventType {
    JOB_KILL,
    JOB_INIT,
    JOB_START,
    JOB_SETUP_COMPLETED,
    JOB_COMPLETED
}

作业内部状态JobStateInternal：

package org.apache.hadoop.state;

/**
 * 定义作业内部状态
 * @author joqk
 *
 */
public enum JobStateInternal {
    NEW,
    SETUP,
    INITED,
    RUNNING,
    SUCCEEDED,
    KILLED,
}

作业状态机JobStateMachine:

package org.apache.hadoop.state;

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.hadoop.event.JobEvent;
import org.apache.hadoop.yarn.event.EventHandler;
import org.apache.hadoop.yarn.state.SingleArcTransition;
import org.apache.hadoop.yarn.state.StateMachine;
import org.apache.hadoop.yarn.state.StateMachineFactory;

public class JobStateMachine implements EventHandler<JobEvent> {

    private String jobID;

    private EventHandler eventHandler;
    private Lock writeLock=null;
    private Lock readLock=null;

    private StateMachine<JobStateInternal,JobEventType,JobEvent> stateMacthie;

    protected static final StateMachineFactory<JobStateMachine,JobStateInternal,JobEventType,JobEvent> stateMachineFactory =
            new StateMachineFactory<JobStateMachine,JobStateInternal,JobEventType,JobEvent>(JobStateInternal.NEW)
            .addTransition(JobStateInternal.NEW, JobStateInternal.INITED, JobEventType.JOB_INIT,new InitTranstion())
            .addTransition(JobStateInternal.INITED, JobStateInternal.SETUP, JobEventType.JOB_START,new StartTransition())
            .installTopology();

    public static class InitTranstion implements SingleArcTransition<JobStateMachine, JobEvent>{

        @Override
        public void transition(JobStateMachine job, JobEvent event) {
            System.out.println("receiving event "+event);

//            job.eventHandler.handle(new JobEvent(JobEventType.JOB_START,job.jobID));
            job.eventHandler.handle(new JobEvent(org.apache.hadoop.event.JobEventType.JOB_START, job.jobID));
        }

    }
    public static class StartTransition implements SingleArcTransition<JobStateMachine, JobEvent>{

        @Override
        public void transition(JobStateMachine job, JobEvent event) {
            System.out.println("receiving event "+event);

//            job.eventHandler.handle(new JobEvent(JobEventType.JOB_START,job.jobID));
//            job.eventHandler.handle(new JobEvent(JobEventType.JOB_SETUP_COMPLETED, job.jobID));
        }

    }

    public JobStateMachine(String jobID,EventHandler eventHandler){

        this.jobID = jobID;
        ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
        this.readLock = readWriteLock.readLock();
        this.writeLock = readWriteLock.writeLock();

        this.eventHandler =eventHandler;
        stateMacthie = stateMachineFactory.make(this);

    }

    /**
     * 得到当前状态机
     * @return
     */
    protected StateMachine<JobStateInternal,JobEventType,JobEvent>  getStateMachine(){
        return stateMacthie;
    }

    /**
     * 获取作业内部状态的变化
     * @return
     */
    public JobStateInternal getInternalState(){
        readLock.lock();
        try{
            return getStateMachine().getCurrentState();
        }finally{
            readLock.unlock();
        }
    }

    @Override
    public void handle(JobEvent event) {
        try{
            writeLock.lock();
            JobStateInternal oldState = getInternalState();
            try{
                getStateMachine().doTransition(event.getType(), event);
            }catch(InvalidStateTranstitonException e){
                System.out.println("Can‘t handle thi event at current state ");
            }
            if(oldState != getInternalState()){
                System.out.println("JOb Transtitioned from "+oldState +"to"+getInternalState());
            }
        }finally{
            writeLock.unlock();
        }
    }

}

这样就完成了一个状态机的定义。