docs: private void initStage(List<String> currentProcesses) {
          // 3. 循环处理每个process
          List<Long> processIds = new ArrayList<Long>();
          for (String process : currentProcesses) {
              processIds.add(StagePathUtils.getProcessId(process));
          }
          Collections.sort(processIds); // 排序一下
          List<Long> needSyncProcessIds = new ArrayList<Long>();
  
          synchronized (currentProcessIds) { // 需要同步处理
              for (Long processId : processIds) {
                  if (!currentProcessIds.contains(processId)) {
                      needSyncProcessIds.add(processId);
                  }
              }
  
              for (Long currentProcessId : currentProcessIds) {
                  if (!processIds.contains(currentProcessId)) {
                      currentStages.remove(currentProcessId);// 如果process已删除,删除本地的stage信息
                  }
              }
  
              if (logger.isDebugEnabled()) {
                  logger.debug("pipeline[{}] old processIds{},current processIds{}", new Object[] { getPipelineId(),
                          currentProcessIds, processIds });
              }
  
              currentProcessIds = processIds; // 切换引用，需设置为volatile保证线程安全&可见性
          }
  
          for (Long syncProcessId : needSyncProcessIds) {
              syncStage(syncProcessId);// 开始同步单个的process
              // boolean reply = initProcessStage(processId);// 通知下节点变化
              // if (!currentProcessIds.contains(processId) && reply == false) {
              // syncStage(processId);// 开始同步单个的process
              // } else {
              // // 如果上一次循环中已经添加了监控，则忽略掉本次处理, 避免重复监听
              // }
          }
  
          if (processIds.size() > 0) {// 尝试触发一下第一个process进行load
              initProcessStage(processIds.get(0));
          }
          processChanged(currentProcessIds);// 通知变化
      }