docs: private List<NodeTask> parseNodeTask(Channel channel) {
          List<NodeTask> tasks = new ArrayList<NodeTask>();
          List<Pipeline> pipelines = channel.getPipelines();
          Long nid = configClientService.currentNode().getId();
          TaskEvent taksEvent = null;
          if (channel.getStatus().isStart()) {
              taksEvent = TaskEvent.CREATE;
          } else if (channel.getStatus().isStop()) {
              taksEvent = TaskEvent.DELETE;
          } else if (channel.getStatus().isPause()) {
              // modify by ljh at 2013-01-31 , pause状态也需要启动setl线程
              // 因为在发布的时候，restart指令不会推送指令，导致setl线程没有启动
              // return tasks;
              taksEvent = TaskEvent.CREATE;
          }
          // 处理当前最新的状态
          for (Pipeline pipeline : pipelines) {
              List<Node> sNodes = pipeline.getSelectNodes();
              for (Node node : sNodes) {
                  if (nid.equals(node.getId())) {// 判断是否为当前的nid
                      NodeTask task = new NodeTask();
                      task.setPipeline(pipeline);
                      NodeTask matchTask = getMatchTask(tasks, task);
                      if (matchTask == null) {
                          matchTask = task;
                          tasks.add(task);
                      }
  
                      matchTask.setPipeline(pipeline);
                      matchTask.getStage().add(StageType.SELECT);
                      matchTask.getEvent().add(taksEvent);
                  }
              }
  
              List<Node> eNodes = pipeline.getExtractNodes();
              for (Node node : eNodes) {
                  if (nid.equals(node.getId())) {// 判断是否为当前的nid
                      NodeTask task = new NodeTask();
                      task.setPipeline(pipeline);
                      NodeTask matchTask = getMatchTask(tasks, task);
                      if (matchTask == null) {
                          matchTask = task;
                          tasks.add(task);
                      }
  
                      matchTask.getStage().add(StageType.EXTRACT);
                      matchTask.getEvent().add(taksEvent);
                  }
              }
  
              List<Node> tlNodes = pipeline.getLoadNodes();
              for (Node node : tlNodes) {
                  if (nid.equals(node.getId())) {// 判断是否为当前的nid
                      NodeTask task = new NodeTask();
                      task.setPipeline(pipeline);
                      NodeTask matchTask = getMatchTask(tasks, task);
                      if (matchTask == null) {
                          matchTask = task;
                          tasks.add(task);
                      }
                      matchTask.getStage().add(StageType.TRANSFORM);
                      matchTask.getEvent().add(taksEvent);
                      matchTask.getStage().add(StageType.LOAD);
                      matchTask.getEvent().add(taksEvent);
                  }
              }
          }
  
          List<Long> pipelineIds = Lists.transform(channel.getPipelines(), new Function<Pipeline, Long>() {
  
              public Long apply(Pipeline input) {
                  return input.getId();
              }
          });
          // 合并一下target中特有的记录，取一下反操作，表示要关闭
          for (NodeTask task : allTasks) {
              Pipeline pipeline = task.getPipeline();
              if (pipeline.getChannelId().equals(channel.getId()) && !pipelineIds.contains(pipeline.getId())) {
                  // /是同一个channel，但对应的pipeline不在列表里
                  // 处理pipeline删除
                  NodeTask deletePipelineTask = new NodeTask();
                  deletePipelineTask.setPipeline(pipeline);
  
                  List<StageType> stages = task.getStage();
                  List<TaskEvent> events = task.getEvent();
                  for (int i = 0; i < stages.size(); i++) {
                      StageType stage = stages.get(i);
                      TaskEvent event = events.get(i);
                      if (event.isCreate()) {
                          deletePipelineTask.getStage().add(stage);
                          deletePipelineTask.getEvent().add(TaskEvent.DELETE);// 添加为关闭
                      }
                  }
  
                  tasks.add(deletePipelineTask);
              }
  
              if (pipelineIds.contains(pipeline.getId())) {// 在当前的channel列表中
                  boolean needAdd = false;
                  NodeTask matchTask = getMatchTask(tasks, task);// 找到对应的匹配
                  if (matchTask == null) {
                      matchTask = new NodeTask();
                      matchTask.setPipeline(pipeline);
                      needAdd = true;
                  }
                  List<StageType> stages = task.getStage();
                  List<TaskEvent> events = task.getEvent();
                  for (int i = 0; i < stages.size(); i++) {
                      StageType stage = stages.get(i);
                      TaskEvent event = events.get(i);
                      TaskEvent matchEvent = getMatchStage(matchTask, stage);
                      if (matchEvent == null && event.isCreate()) {// 对应的stage已经被移除，触发一个DELETE操作
                          matchTask.getStage().add(stage);
                          matchTask.getEvent().add(TaskEvent.DELETE);
                      }
                  }
  
                  if (needAdd && matchTask.getStage().size() > 0) {
                      tasks.add(matchTask);
                  }
  
              }
          }
  
          // 判断当前的task是否需要全部关闭
          for (NodeTask task : tasks) {
              boolean shutdown = true;
              for (TaskEvent event : task.getEvent()) {// task已为当前最新节点信息
                  shutdown &= event.isDelete();
              }
              task.setShutdown(shutdown);
          }
          return tasks;
      }