yarn中的事件分发与状态机框架

// 1) 创建事件异步分发处理对象
Dispatcher dispatcher = new AsyncDispatcher("RM Event dispatcher");
// 2) 注册事件以及对应的处理类对象
dispatcher.register(RMFatalEventType.class,new ResourceManager.RMFatalEventDispatcher());
// 3）初始化及启动事件分发器
dispatcher.init();
dispatcher.start();
// 4) 向事件分发器投递消息
dispatcher.getEventHandler().handle(new RMFatalEvent(RMFatalEventType.STATE_STORE_OP_FAILED, failureCause));

public interface EventHandler<T extends Event> {
    void handle(T event);
}

private class RMFatalEventDispatcher implements EventHandler<RMFatalEvent> {
    public void handle(RMFatalEvent event) {
        ...
    }
}

// 1. 构造状态机工厂, 并添加状态装换
StateMachineFactory<RMAppImpl, RMAppState, RMAppEventType, RMAppEvent> stateMachineFactory =
    new StateMachineFactory<RMAppImpl, RMAppState, RMAppEventType, RMAppEvent>(RMAppState.NEW).
    addTransition(
        RMAppState.NEW, // 当前状态
        RMAppState.NEW_SAVING, // 处理后的状态
        RMAppEventType.START, // 事件类型
        new RMAppNewlySavingTransition()). // 处理类对象
    addTransition(
        RMAppState.NEW, // 当前状态
        EnumSet.of( // 处理后可能的状态集
            RMAppState.SUBMITTED,
            RMAppState.ACCEPTED,
            RMAppState.FINISHED, 
            RMAppState.FAILED,
            RMAppState.KILLED,
            RMAppState.FINAL_SAVING),
        RMAppEventType.RECOVER, // 事件类型
        new RMAppRecoveredTransition()). // 处理类对象
    addTransition(...).
    installTopology();
// 2. 构造状态机实例队列
StateMachine<RMAppState, RMAppEventType, RMAppEvent> stateMachine = stateMachineFactory.make(this);
// 3. 事件处理与状态转换
stateMachine.doTransition(event.getType(), event);

final public class StateMachineFactory<OPERAND, STATE extends Enum<STATE>, EVENTTYPE extends Enum<EVENTTYPE>, EVENT> {
    private final TransitionsListNode transitionsListNode;
    private Map<STATE, Map<EVENTTYPE, Transition<OPERAND, STATE, EVENTTYPE, EVENT>>> stateMachineTable;
    private STATE defaultInitialState;
}

private class TransitionsListNode {
    final ApplicableTransition<OPERAND, STATE, EVENTTYPE, EVENT> transition;
    final TransitionsListNode next;

    TransitionsListNode(
        ApplicableTransition<OPERAND, STATE, EVENTTYPE, EVENT> transition,
        TransitionsListNode next) {
        this.transition = transition;
        this.next = next;
    }
}

private static class RMAppTransition implements SingleArcTransition<RMAppImpl, RMAppEvent> {
    public void transition(RMAppImpl app, RMAppEvent event) {
    };
}

private static final class RMAppNewlySavingTransition extends RMAppTransition {
    @Override
    public void transition(RMAppImpl app, RMAppEvent event) {
        ...
    }
}

private static final class RMAppRecoveredTransition implements MultipleArcTransition<RMAppImpl, RMAppEvent, RMAppState> {
    @Override
    public RMAppState transition(RMAppImpl app, RMAppEvent event) {
        ...
        return RMAppState.ACCEPTED;
    }
}

private class InternalStateMachine implements StateMachine<STATE, EVENTTYPE, EVENT> {
    @Override
    public synchronized STATE doTransition(EVENTTYPE eventType, EVENT event)
        throws InvalidStateTransitionException {
        listener.preTransition(operand, currentState, event);
        STATE oldState = currentState;
        currentState = 
            StateMachineFactory.this.doTransition(operand, currentState, eventType, event);
        listener.postTransition(operand, oldState, currentState, event);
        return currentState;
    }
}

final public class StateMachineFactory<OPERAND, STATE extends Enum<STATE>, EVENTTYPE extends Enum<EVENTTYPE>, EVENT> {
    private STATE doTransition(
        OPERAND operand, 
        STATE oldState, 
        EVENTTYPE eventType, 
        EVENT event)
        throws InvalidStateTransitionException {
        Map<EVENTTYPE, Transition<OPERAND, STATE, EVENTTYPE, EVENT>> transitionMap = stateMachineTable.get(oldState);
        if (transitionMap != null) {
            Transition<OPERAND, STATE, EVENTTYPE, EVENT> transition = transitionMap.get(eventType);
            if (transition != null) {
                return transition.doTransition(operand, oldState, event, eventType);
            }
        }
        throw new InvalidStateTransitionException(oldState, eventType);
    }
    
    private class SingleInternalArc implements Transition<OPERAND, STATE, EVENTTYPE, EVENT> {
        public STATE doTransition(OPERAND operand, STATE oldState, EVENT event, EVENTTYPE eventType) {
            if (hook != null) {
                hook.transition(operand, event);
            }
            return postState;
        }
    }
    
    private class MultipleInternalArc implements Transition<OPERAND, STATE, EVENTTYPE, EVENT> {
        public STATE doTransition(OPERAND operand, STATE oldState, EVENT event, EVENTTYPE eventType) throws InvalidStateTransitionException {
            STATE postState = hook.transition(operand, event);
            if (!validPostStates.contains(postState)) {
                throw new InvalidStateTransitionException(oldState, eventType);
            }
            return postState;
        }
    }
}

rmDispatcher.register(RMAppEventType.class, new ApplicationEventDispatcher(rmContext));

@Private
public static final class ApplicationEventDispatcher implements EventHandler<RMAppEvent> {

    private final RMContext rmContext;

    public ApplicationEventDispatcher(RMContext rmContext) {
        this.rmContext = rmContext;
    }

    @Override
    public void handle(RMAppEvent event) {
        ApplicationId appID = event.getApplicationId();
        RMApp rmApp = this.rmContext.getRMApps().get(appID);
        if (rmApp != null) {
            try {
                rmApp.handle(event);
            } catch (Throwable t) {
                LOG.error(
                    "Error in handling event type " + event.getType() +
                    " for application " + appID, t);
            }
        }
    }
}

// RMAppImpl.java
public void handle(RMAppEvent event) {
    ...
    this.stateMachine.doTransition(event.getType(), event);
    ...
}