docs: public synchronized void termin(TerminType type) {
          // 构建termin信号
          List<Long> processIds = new ArrayList<Long>(progress.keySet());
          Collections.sort(processIds);// 做一下排序
          for (Long processId : processIds) {
              EtlEventData eventData = progress.get(processId).getData();
  
              TerminEventData data = new TerminEventData();
              data.setPipelineId(getPipelineId());
              data.setType(type);
              data.setCode("channel");
              data.setDesc(type.toString());
              data.setProcessId(processId);
              if (eventData != null) {
                  data.setBatchId(eventData.getBatchId());
                  data.setCurrNid(eventData.getCurrNid());
                  data.setStartTime(eventData.getStartTime());
                  data.setEndTime(eventData.getEndTime());
                  data.setFirstTime(eventData.getFirstTime());
                  data.setNumber(eventData.getNumber());
                  data.setSize(eventData.getSize());
                  data.setExts(eventData.getExts());
              }
              offerTermin(data);
              progress.remove(processId);
          }
  
          // 重新初始化一下select调度
          initSelect();
      }