otter ArbitrateEventService源码分析

/**
 * 仲裁器事件处理的service，使用者可关注相应的await/single两个方法
 * 
 * @author jianghang 2011-8-9 下午04:39:49
 */
public interface ArbitrateEventService {

    public MainStemArbitrateEvent mainStemEvent();

    public SelectArbitrateEvent selectEvent();

    public ExtractArbitrateEvent extractEvent();

    public TransformArbitrateEvent transformEvent();

    public LoadArbitrateEvent loadEvent();

    public TerminArbitrateEvent terminEvent();

    public ToolArbitrateEvent toolEvent();
}

/**
 * 抽象select模块的调度接口
 * 
 * @author jianghang 2012-9-27 下午09:54:53
 * @version 4.1.0
 */
public interface SelectArbitrateEvent extends ArbitrateEvent {

    public EtlEventData await(Long pipelineId) throws InterruptedException;

    public void single(EtlEventData data);
}

/**
 * select delegate实现
 * 
 * @author jianghang 2012-9-28 上午10:36:38
 * @version 4.1.0
 */
public class SelectDelegateArbitrateEvent extends AbstractDelegateArbitrateEvent implements SelectArbitrateEvent {

    private Map<ArbitrateMode, SelectArbitrateEvent> delegate;

    public EtlEventData await(Long pipelineId) throws InterruptedException {
        return delegate.get(chooseMode(pipelineId)).await(pipelineId);
    }

    public void single(EtlEventData data) {
        delegate.get(chooseMode(data.getPipelineId())).single(data);
    }

    public void setDelegate(Map<ArbitrateMode, SelectArbitrateEvent> delegate) {
        this.delegate = delegate;
    }

}

   public EtlEventData await(Long pipelineId) throws InterruptedException {
       ...
        MemoryStageController stageController = ArbitrateFactory.getInstance(pipelineId, MemoryStageController.class);
        Long processId = stageController.waitForProcess(StageType.SELECT); // 符合条件的processId

        ChannelStatus status = permitMonitor.getChannelPermit();
        if (status.isStart()) {// 即时查询一下当前的状态，状态随时可能会变
            EtlEventData eventData = new EtlEventData();
            eventData.setPipelineId(pipelineId);
            eventData.setProcessId(processId);
            eventData.setStartTime(new Date().getTime());// 返回当前时间
            Long nid = ArbitrateConfigUtils.getCurrentNid();
            eventData.setCurrNid(nid);
            eventData.setNextNid(nid);
            return eventData;// 只有这一条路返回
        } else {
            ...
        }
    }

	public void single(EtlEventData data) {
        Assert.notNull(data);
        MemoryStageController stageController = ArbitrateFactory.getInstance(data.getPipelineId(),
                                                                             MemoryStageController.class);
        stageController.single(StageType.SELECT, data);// 通知下一个节点
    }

public Long waitForProcess(StageType stage) throws InterruptedException {
        if (stage.isSelect() && !replys.containsKey(stage)) {
            initSelect();
        }
        Long processId = replys.get(stage).take();
        if (stage.isSelect()) {// select一旦分出processId，就需要在progress中记录一笔，用于判断谁是最小的一个processId
            progress.put(processId, nullProgress);
        }
        return processId;
    }

    public synchronized boolean single(StageType stage, EtlEventData etlEventData) {
        boolean result = false;
        switch (stage) {
            case SELECT:
                if (progress.containsKey(etlEventData.getProcessId())) {// 可能发生了rollback，对应的progress已经被废弃
                    progress.put(etlEventData.getProcessId(), new StageProgress(stage, etlEventData));
                    replys.get(StageType.EXTRACT).offer(etlEventData.getProcessId());
                    result = true;
                }
                break;
            case EXTRACT:
                if (progress.containsKey(etlEventData.getProcessId())) {
                    progress.put(etlEventData.getProcessId(), new StageProgress(stage, etlEventData));
                    replys.get(StageType.TRANSFORM).offer(etlEventData.getProcessId());
                    result = true;
                }
                break;
            case TRANSFORM:
                if (progress.containsKey(etlEventData.getProcessId())) {
                    progress.put(etlEventData.getProcessId(), new StageProgress(stage, etlEventData));
                    result = true;
                }
                // 并不是立即触发，通知最小的一个process启动
                computeNextLoad();
                break;
            case LOAD:
                Object removed = progress.remove(etlEventData.getProcessId());
                // 并不是立即触发，通知下一个最小的一个process启动
                computeNextLoad();
                // 一个process完成了，自动添加下一个process
                if (removed != null) {
                    replys.get(StageType.SELECT).offer(atomicMaxProcessId.incrementAndGet());
                    result = true;
                }
                break;
            default:
                break;
        }

        return result;
    }

private Map<StageType, ReplyProcessQueue> replys;