@Slf4j
@RestController
@RequestMapping("/order")
public class OrderController {
  @PostMapping("/f1")
  public Object f1(@RequestBody Object obj) {
  log.info("OrderController->f1,接收參數(shù),obj = " + obj.toString());
  Map objMap = (Map) obj;
  OrderInfo orderInfo = new OrderInfo();
  orderInfo.setUserName((String) objMap.get("userName"));
  orderInfo.setTradeName((String) objMap.get("tradeName"));
  orderInfo.setOrderTime(System.currentTimeMillis());
  LinkedBlockingQueue<StepModel> queueA = FlowQueue.getBlockingQueue("QUEUE_A");
  QueueUtils.putStepPutInQueue(queueA,orderInfo);
  log.info("OrderController->f1,返回." );
  return ResultObj.builder().code("200").message("成功").build();
  }
}

@Slf4j
@Service
public class FlowStarter implements InitializingBean {
  @Override
  public void afterPropertiesSet() throws Exception {
    log.info("FlowWorker創(chuàng)建流程.");
    FlowManager flowManager = new FlowManager();
    flowManager.buildContainer(ConstantUtils.STEP_01,5,
      FlowQueue.getBlockingQueue("QUEUE_A"), Step01Handler.class
    );
    flowManager.buildContainer(ConstantUtils.STEP_02,5,
      FlowQueue.getBlockingQueue("QUEUE_B"), Step02Handler.class
    );
    flowManager.startContainers();
    log.info("FlowWorker啟動(dòng)流程完成.");
  }
}

public class FlowManager {
  // 管理器名稱
  private String name;
  // 管理線程池容器
  private Map<String, StepContainer> stepContainerMap = new LinkedHashMap<>();
  public FlowManager() {}
  // 創(chuàng)建線程池容器
  public void buildContainer(String name, int poolSize, BlockingQueue<StepModel> queue,
                  Class<? extends StepHandler> handlerClazz) {
    StepContainer stepWorker = new StepContainer();
    stepWorker.createThreadPool(poolSize, queue, handlerClazz);
    stepWorker.setName(name);
    stepWorker.setFlowManager(this);
    this.stepContainerMap.put(name, stepWorker);
  }
  // 啟動(dòng)線程池容器
  public void startContainers() {
    for (StepContainer stepContainer : this.stepContainerMap.values()) {
        stepContainer.startRunExecutor();
    }
  }
  // 關(guān)閉線程池容器
  public void stopContainers() {
    for (StepContainer stepContainer : this.stepContainerMap.values()) {
        stepContainer.stopRunExecutor();
    }
    this.stepContainerMap.clear();
  }
  // 任務(wù)放入下一個(gè)線程池
  public boolean sendToNextContainer(String nextStepName, Object obj) {
    if (nextStepName != null && !StringUtils.equals(nextStepName, "")) {
      if (this.stepContainerMap.containsKey(nextStepName)) {
        this.stepContainerMap.get(nextStepName).putStepInQueue(obj);
        return true;
      } else {
        return false;
      }
    } else {
        return false;
    }
  }
  public String getName() {
     return name;
  }
}

@Slf4j
public class StepContainer {
  // 線程池名稱
  private String name;
  // 線程池
  private ExecutorService threadPool;
  // 線程數(shù)目
  private int nThreads = 0;
  // 線程處理業(yè)務(wù)handler類
  private Class handlerClazz;
  // 線程處理業(yè)務(wù)隊(duì)列
  private BlockingQueue<StepModel> queue = null;
  // 線程池內(nèi)線程管理
  private Map<String, StepExecutor> stepExecutorMap = new LinkedHashMap<>();
  // 線程池運(yùn)行狀態(tài)
  private boolean isRun = false;
  // 線程池管理器
  private FlowManager flowManager = null;
  // 構(gòu)造函數(shù)
  public StepContainer() {}
  // 創(chuàng)建線程池
  public boolean createThreadPool(int nThreads, BlockingQueue<StepModel> queue,
          Class<? extends StepHandler> handlerClazz) {
   try {
       this.nThreads = nThreads;
       this.queue = queue;
       this.handlerClazz = handlerClazz;
       this.threadPool = Executors.newFixedThreadPool(this.nThreads, new ThreadFactory() {
           @Override
           public Thread newThread(Runnable runnable) {
               return new Thread(runnable);
           }
       });
   } catch (Exception e) {
       e.printStackTrace();
       return false;
   }
   return true;
  }
  // 啟動(dòng)線程
  public void startRunExecutor() {
    if (!this.isRun) {
      if (this.handlerClazz != null) {
        log.info("線程池: " + this.name + ",啟動(dòng),加載線程Executor.");
        StepExecutor stepExecutor;
        String executorName = "";
        for (int num = 0; num < this.nThreads; num++) {
          try {
            executorName = this.name + "_" + (num + 1);
            StepHandler stepHandler = (StepHandler) createStepHandler(this.handlerClazz);
            stepExecutor = new StepExecutor(executorName, this.queue, stepHandler, this);
            this.threadPool.execute(stepExecutor);
            this.stepExecutorMap.put(executorName, stepExecutor);
          } catch (Exception e) {
              e.printStackTrace();
          }
        }
        this.isRun = true;
      }
    }
  }
  // 關(guān)閉線程
  public void stopRunExecutor() {
    if (isRun) {
        Iterator iterator = this.stepExecutorMap.values().iterator();
        while (iterator.hasNext()) {
            StepExecutor stepExecutor = (StepExecutor) iterator.next();
            stepExecutor.stop();
        }
        this.stepExecutorMap.clear();
        this.isRun = false;
    }
  }
  // 從隊(duì)列獲取任務(wù)
  public StepModel getStepFromQueue() {
    StepModel stepModel = null;
    synchronized (this.queue) {
      try {
          if (this.queue.size() > 0) {
              stepModel = this.queue.take();
          }
      } catch (Exception e) {
          log.info("從隊(duì)列獲取任務(wù)異常.");
          e.printStackTrace();
      }
    }
    return stepModel;
  }
  // 任務(wù)放入隊(duì)列
  public void putStepInQueue(Object obj) {
    try {
      StepModel stepModel = new StepModel(obj);
      stepModel.setPutInQueueTime(System.currentTimeMillis());
      this.queue.put(stepModel);
    } catch (InterruptedException e) {
      log.info("任務(wù)放入隊(duì)列異常.");
      e.printStackTrace();
    }
  }
  // 重新放入
  public void putStepInQueueAgain(StepModel stepModel) {
    stepModel.setFinished(false);
    stepModel.setPutInQueueNext(false);
    stepModel.setPutInQueueAgain(false);
    try {
        this.queue.put(stepModel);
    } catch (InterruptedException e) {
        log.info("任務(wù)重新放入隊(duì)列異常.");
        e.printStackTrace();
    }
  }
  // 清空隊(duì)列
  public void clearQueue() {
    if (this.queue != null) {
        this.queue.clear();
    }
  }
  // 初始化實(shí)例對象
  public Object createStepHandler(Class clazz)
        throws InstantiationException, IllegalAccessException {
    Object object = clazz.newInstance();
    return object;
  }
  public String getName() {
    return name;
  }
  public void setName(String name) {
    this.name = name;
  }
  public FlowManager getFlowManager() {
    return flowManager;
  }
  public void setFlowManager(FlowManager flowManager) {
    this.flowManager = flowManager;
  }
}

@Slf4j
public class StepExecutor implements Runnable {
  // 執(zhí)行器名稱
  private String name;
  // 線程執(zhí)行的任務(wù)
  private StepModel stepModel;
  // 線程執(zhí)行的隊(duì)列
  private BlockingQueue<StepModel> queue;
  // 線程執(zhí)行的業(yè)務(wù)處理邏輯
  private Object stepHandler;
  // 線程運(yùn)行狀態(tài)
  private volatile boolean isRun = false;
  // 線程開啟(True)和關(guān)閉(False)
  private volatile boolean isClose = false;
  // 線程隸屬容器
  private StepContainer stepContainer;
  // 線程計(jì)數(shù)器(關(guān)閉線程使用)
  private CountDownLatch countDownLatch = null;
  public StepExecutor() {}
  public StepExecutor(String name, BlockingQueue<StepModel> queue,
                      StepHandler stepHandler, StepContainer stepContainer) {
      this.name = name;
      this.queue = queue;
      this.stepHandler = stepHandler;
      this.stepContainer = stepContainer;
  }
  @Override
  public void run() {
   this.isRun = true;
   this.countDownLatch = new CountDownLatch(1);
   // 沒收到關(guān)閉信號(hào),則循環(huán)運(yùn)行
   while (!this.isClose) {
    this.stepModel = null;
    String threadName = "【線程池:" + this.stepContainer.getName() 
     + ",線程:" + Thread.currentThread().getName() + "】";
    // 循環(huán)運(yùn)行,為防止中斷和卡主,需捕獲異常
    try {
      StepHandler stepHandler = (StepHandler) this.stepHandler;
      this.stepModel = this.stepContainer.getStepFromQueue();
      if (this.stepModel != null) {
      log.info(threadName + ",處理任務(wù).");
      this.stepModel.getStepResultList().clear();
      stepHandler.execute(this.stepModel);
      // 執(zhí)行完成后結(jié)果數(shù)據(jù)
      List<StepResult> stepResultList = this.stepModel.getStepResultList();
      boolean isFinished = this.stepModel.isFinished();
      boolean isPutInQueueAgain = this.stepModel.isPutInQueueAgain();
      boolean isPutInQueueNext = this.stepModel.isPutInQueueNext();
      if (isFinished && !isPutInQueueAgain && !isPutInQueueNext) {
        log.info(threadName + ",任務(wù)結(jié)束.");
      }
      if (!isFinished && isPutInQueueAgain && !isPutInQueueNext) {
        log.info(threadName + ",任務(wù)在本步驟未完成,重新放隊(duì)列.");
        this.stepContainer.putStepInQueueAgain(this.stepModel);
      }
      if (!isFinished && !isPutInQueueAgain && isPutInQueueNext) {
        int resultNum = stepResultList.size();
        if (resultNum > 0) {
            for (StepResult stepResult : stepResultList) {
              log.info(threadName + ",任務(wù)在本步驟已經(jīng)完成,發(fā)送給下一個(gè)線程池: " 
  	     + stepResult.getNextStepName() + ",執(zhí)行.");
                this.stepContainer.getFlowManager().sendToNextContainer(
                        stepResult.getNextStepName(),
                        stepResult.getResult());
            }
        }
      }
      } else {
          threadToSleep(1000 * 3L);
      }
    } catch (Exception e) {
        log.info("執(zhí)行器異常.");
        e.printStackTrace();
        this.stepContainer.putStepInQueueAgain(this.stepModel);
    }
   }
   // 跳出循環(huán)后,線程計(jì)數(shù)減1
   this.countDownLatch.countDown();
   this.isRun = false;
  }
  public void stop() {
   this.isClose = true;
   if (this.countDownLatch != null) {
    while (this.countDownLatch.getCount() > 0L) {
      try {
          this.countDownLatch.await();
      } catch (InterruptedException e) {
          log.info("線程關(guān)閉異常.");
          e.printStackTrace();
      }
    }
   }
   this.isClose = false;
  }
  public void threadToSleep(long time) {
    try {
      Thread.sleep(time);
    } catch (Exception e) {
      log.info("線程休眠異常.");
      e.printStackTrace();
    }
  }
}

public abstract class StepHandler {
  public StepHandler() {}
  public abstract void execute(StepModel stepModel);
}

@Slf4j
public class Step01Handler extends StepHandler {
 @Override
 public void execute(StepModel stepModel) {
  log.info("Step01Handler執(zhí)行開始,stepModel: " + stepModel.toString());
  OrderInfo orderInfo = (OrderInfo) stepModel.getObj();
  List<StepResult> stepResultList = stepModel.getStepResultList();
  try {
      log.info("Step01Handler執(zhí)行,處理訂單.");
      String orderNo = UUID.randomUUID().toString()
              .replace("-", "").toUpperCase();
      orderInfo.setOrderNo(orderNo);
      orderInfo.setPlatformType("線上");
      orderInfo.setOrderSource("Web");
      stepModel.setFinished(false);
      stepModel.setPutInQueueNext(true);
      stepModel.setPutInQueueAgain(false);
      stepResultList.add(new StepResult(ConstantUtils.STEP_02, orderInfo));
  } catch (Exception e) {
      stepModel.setFinished(false);
      stepModel.setPutInQueueNext(false);
      stepModel.setPutInQueueAgain(true);
      stepResultList.add(new StepResult(ConstantUtils.STEP_01, orderInfo));
  }
  log.info("Step01Handler執(zhí)行完成,stepModel: " + stepModel.toString());
 }
}

@Slf4j
public class Step02Handler  extends StepHandler{
  @Override
  public void execute(StepModel stepModel) {
   log.info("Step02Handler執(zhí)行開始,stepModel: " + stepModel.toString());
   OrderInfo orderInfo = (OrderInfo) stepModel.getObj();
   List<StepResult> stepResultList = stepModel.getStepResultList();
   try {
    orderInfo.setEndTime(System.currentTimeMillis());
    stepModel.setFinished(true);
    stepModel.setPutInQueueNext(false);
    stepModel.setPutInQueueAgain(false);
    log.info("Step02Handler執(zhí)行,入庫.");
   } catch (Exception e) {
    stepModel.setFinished(true);
    stepModel.setPutInQueueNext(false);
    stepModel.setPutInQueueAgain(false);
   }
   log.info("Step02Handler執(zhí)行完成,stepModel: " + stepModel.toString());
  }
}

public class FlowQueue {
 private static final LinkedBlockingQueue<StepModel> queueA = new LinkedBlockingQueue<StepModel>();
 private static final LinkedBlockingQueue<StepModel> queueB = new LinkedBlockingQueue<StepModel>();
 public static LinkedBlockingQueue<StepModel> getBlockingQueue(String queueName) {
  LinkedBlockingQueue<StepModel> queue = null;
  switch (queueName) {
    case "QUEUE_A":
      queue = queueA;
      break;
    case "QUEUE_B":
      queue = queueB;
      break;
  }
  return queue;
 }
}

@Slf4j
public class QueueUtils {
 public static StepModel getStepFromQueue(
     LinkedBlockingQueue<StepModel> queue) {
   StepModel stepModel = null;
   try {
    if (queue.size() > 0) {
        stepModel = queue.take();
    }
   } catch (Exception e) {
     log.info("讀隊(duì)列異常.");
     e.printStackTrace();
   }
   return stepModel;
 }
 public static void putStepPutInQueue(
    LinkedBlockingQueue<StepModel> queue, Object obj) {
   try {
    StepModel stepModel = new StepModel(obj);
    stepModel.setPutInQueueTime(System.currentTimeMillis());
    queue.put(stepModel);
   } catch (Exception e) {
    log.info("寫隊(duì)列異常.");
    e.printStackTrace();
   }
 }
 public static int getQueueSize(
       LinkedBlockingQueue<StepModel> queue) {
   int size = 0;
   try {
       size = queue.size();
   } catch (Exception e) {
       log.info("獲取隊(duì)列Size異常.");
       e.printStackTrace();
   }
   return size;
 }
}

public class ConstantUtils {
 public static final String STEP_01 = "STEP_01_THREAD_POOL";
 public static final String STEP_02 = "STEP_02_THREAD_POOL";
}

@Data
public class StepModel {
  // 任務(wù)對象
  private Object obj;
  // 任務(wù)執(zhí)行結(jié)果
  private List<StepResult> stepResultList;
  // 任務(wù)接收時(shí)間
  private long putInQueueTime;
  // 任務(wù)完成標(biāo)識(shí)
  private boolean isFinished = false;
  // 任務(wù)重新放入隊(duì)列標(biāo)識(shí)
  private boolean isPutInQueueAgain = false;
  // 任務(wù)放入下一個(gè)隊(duì)列標(biāo)識(shí)
  private boolean isPutInQueueNext = false;
  public StepModel(Object object) {
      this.obj = object;
      this.stepResultList = new ArrayList<>();
  }
}

@Data
public class StepResult {
 // 目標(biāo)線程池名
 private String nextStepName;
 // 執(zhí)行結(jié)果
 private Object result;
 public StepResult(String nextStepName,Object result){
   this.nextStepName = nextStepName;
   this.result = result;
 }
}

@Data
@NoArgsConstructor
public class OrderInfo {
  private String userName;
  private String orderNo;
  private String tradeName;
  private String platformType;
  private String orderSource;
  private long orderTime;
  private long endTime;
}

@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class ResultObj {
  private String code;
  private String message;
}