public void await(Long pipelineId) throws InterruptedException {
        Assert.notNull(pipelineId);
        PermitMonitor permitMonitor = ArbitrateFactory.getInstance(pipelineId, PermitMonitor.class);
        ChannelStatus status = permitMonitor.getChannelPermit(true);
        boolean isRuning = check(pipelineId);
        if (!status.isStart() && isRuning) {
            // 当前状态不为启动，强制设置为taking，下次授权启动后重新追数据
            MainStemEventData data = new MainStemEventData();
            data.setPipelineId(pipelineId);
            data.setStatus(MainStemEventData.Status.TAKEING);
            single(data);
            permitMonitor.waitForChannelPermit(); // 阻塞等待挂起
            return;
        } else if (status.isStart() && isRuning) {// 正常状态
            return;
        } else if (isRuning == false) {
            if (!status.isStart()) {
                permitMonitor.waitForChannelPermit(); // 阻塞等待挂起
            }
            MainstemMonitor mainstemMonitor = ArbitrateFactory.getInstance(pipelineId, MainstemMonitor.class);
            mainstemMonitor.waitForActive();// 等待自己成为active

            status = permitMonitor.getChannelPermit(false);
            if (status.isStart()) {
                return;
            } else {
                logger.info("pipelineId[{}] mainstem ignore by status[{}]", new Object[] { pipelineId, status });
                await(pipelineId);// 重新进行check一次
            }
        }

    }
public void await ( Long pipelineId ) throws InterruptedException { Assert . notNull ( pipelineId ) ; PermitMonitor permitMonitor = ArbitrateFactory . getInstance ( pipelineId , PermitMonitor . class ) ; ChannelStatus status = permitMonitor . getChannelPermit ( true ) ; boolean isRuning = check ( pipelineId ) ; if ( ! status . isStart ( ) && isRuning ) { MainStemEventData data = new MainStemEventData ( ) ; data . setPipelineId ( pipelineId ) ; data . setStatus ( MainStemEventData . Status . TAKEING ) ; single ( data ) ; permitMonitor . waitForChannelPermit ( ) ; return ; } else if ( status . isStart ( ) && isRuning ) { return ; } else if ( isRuning == false ) { if ( ! status . isStart ( ) ) { permitMonitor . waitForChannelPermit ( ) ; } MainstemMonitor mainstemMonitor = ArbitrateFactory . getInstance ( pipelineId , MainstemMonitor . class ) ; mainstemMonitor . waitForActive ( ) ; status = permitMonitor . getChannelPermit ( false ) ; if ( status . isStart ( ) ) { return ; } else { logger . info ( "pipelineId[{}] mainstem ignore by status[{}]" , new Object [ ] { pipelineId , status } ) ; await ( pipelineId ) ; } } }
<pre>
算法:
1. 检查当前的Permit，阻塞等待其授权(解决Channel的pause状态处理)
2. 尝试创建对应的mainStem节点(非持久化节点)
a. 如果创建成功，则直接构造结果返回
b. 如果创建失败，则关注该节点的exist信号. 继续执行步骤2
</pre>