content: package com.platform.core.thread;
  
  import com.platform.core.biz.model.HandleCallbackParam;
  import com.platform.core.biz.model.ReturnT;
  import com.platform.core.biz.model.TriggerParam;
  import com.platform.core.executor.JobExecutor;
  import com.platform.core.handler.IJobHandler;
  import com.platform.core.log.JobFileAppender;
  import com.platform.core.log.JobLogger;
  import com.platform.core.util.ShardingUtil;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.io.PrintWriter;
  import java.io.StringWriter;
  import java.util.Collections;
  import java.util.Date;
  import java.util.HashSet;
  import java.util.Set;
  import java.util.concurrent.FutureTask;
  import java.util.concurrent.LinkedBlockingQueue;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.TimeoutException;
  
  public class JobThread extends Thread {
      private static Logger logger = LoggerFactory.getLogger(JobThread.class);
  
      private int jobId;
      private IJobHandler handler;
      private LinkedBlockingQueue<TriggerParam> triggerQueue;
      private Set<Long> triggerLogIdSet;        // avoid repeat trigger for the same TRIGGER_LOG_ID
  
      private volatile boolean toStop = false;
      private String stopReason;
  
      private boolean running = false;    // if running job
      private int idleTimes = 0;            // idel times
  
  
      public JobThread(int jobId, IJobHandler handler) {
          this.jobId = jobId;
          this.handler = handler;
          this.triggerQueue = new LinkedBlockingQueue<>();
          this.triggerLogIdSet = Collections.synchronizedSet(new HashSet<>());
      }
  
      public IJobHandler getHandler() {
          return handler;
      }
  
      /**
       * new trigger to queue
       *
       * @param triggerParam
       * @return
       */
      public ReturnT<String> pushTriggerQueue(TriggerParam triggerParam) {
          // avoid repeat
          if (triggerLogIdSet.contains(triggerParam.getLogId())) {
              logger.info(">>>>>>>>>>> repeate trigger job, logId:{}", triggerParam.getLogId());
              return new ReturnT<>(ReturnT.FAIL_CODE, "repeate trigger job, logId:" + triggerParam.getLogId());
          }
  
          triggerLogIdSet.add(triggerParam.getLogId());
          triggerQueue.add(triggerParam);
          return ReturnT.SUCCESS;
      }
  
      /**
       * kill job thread
       *
       * @param stopReason
       */
      public void toStop(String stopReason) {
          /**
           * Thread.interrupt只支持终止线程的阻塞状态(wait、join、sleep)，
           * 在阻塞出抛出InterruptedException异常,但是并不会终止运行的线程本身；
           * 所以需要注意，此处彻底销毁本线程，需要通过共享变量方式；
           */
          this.toStop = true;
          this.stopReason = stopReason;
      }
  
      /**
       * is running job
       *
       * @return
       */
      public boolean isRunningOrHasQueue() {
          return running || triggerQueue.size() > 0;
      }
  
      @Override
      public void run() {
  
          // init
          try {
              handler.init();
          } catch (Throwable e) {
              logger.error(e.getMessage(), e);
          }
  
          // execute
          while (!toStop) {
              running = false;
              idleTimes++;
  
              TriggerParam tgParam = null;
              ReturnT<String> executeResult = null;
              try {
                  // to check toStop signal, we need cycle, so wo cannot use queue.take(), instand of poll(timeout)
                  tgParam = triggerQueue.poll(3L, TimeUnit.SECONDS);
                  if (tgParam != null) {
                      running = true;
                      idleTimes = 0;
                      triggerLogIdSet.remove(tgParam.getLogId());
  
                      // log filename, like "logPath/yyyy-MM-dd/9999.log"
                      String logFileName = JobFileAppender.makeLogFileName(new Date(tgParam.getLogDateTime()), tgParam.getLogId());
                      JobFileAppender.contextHolder.set(logFileName);
                      ShardingUtil.setShardingVo(new ShardingUtil.ShardingVO(tgParam.getBroadcastIndex(), tgParam.getBroadcastTotal()));
  
                      // execute
                      JobLogger.log("<br>----------- web job execute start -----------<br>----------- Param:" + tgParam.getExecutorParams());
  
                      if (tgParam.getExecutorTimeout() > 0) {
                          // limit timeout
                          Thread futureThread = null;
                          try {
                              final TriggerParam tgParamT = tgParam;
                              FutureTask<ReturnT<String>> futureTask = new FutureTask<>(() -> handler.execute(tgParamT));
                              futureThread = new Thread(futureTask);
                              futureThread.start();
  
                              executeResult = futureTask.get(tgParam.getExecutorTimeout(), TimeUnit.MINUTES);
                          } catch (TimeoutException e) {
  
                              JobLogger.log("<br>----------- web job execute timeout");
                              JobLogger.log(e);
  
                              executeResult = new ReturnT<>(IJobHandler.FAIL_TIMEOUT.getCode(), "job execute timeout ");
                          } finally {
                              futureThread.interrupt();
                          }
                      } else {
                          // just execute
                          executeResult = handler.execute(tgParam);
                      }
  
                      if (executeResult == null) {
                          executeResult = IJobHandler.FAIL;
                      } else {
                          executeResult.setMsg(
                                  (executeResult != null && executeResult.getMsg() != null && executeResult.getMsg().length() > 50000)
                                          ? executeResult.getMsg().substring(0, 50000).concat("...")
                                          : executeResult.getMsg());
                          executeResult.setContent(null);    // limit obj size
                      }
                      JobLogger.log("<br>----------- web job execute end(finish) -----------<br>----------- ReturnT:" + executeResult);
  
                  } else {
                      if (idleTimes > 30) {
                          if (triggerQueue.size() == 0) {    // avoid concurrent trigger causes jobId-lost
                              JobExecutor.removeJobThread(jobId, "executor idel times over limit.");
                          }
                      }
                  }
              } catch (Throwable e) {
                  if (toStop) {
                      JobLogger.log("<br>----------- JobThread toStop, stopReason:" + stopReason);
                  }
  
                  StringWriter stringWriter = new StringWriter();
                  e.printStackTrace(new PrintWriter(stringWriter));
                  String errorMsg = stringWriter.toString();
                  executeResult = new ReturnT<>(ReturnT.FAIL_CODE, errorMsg);
  
                  JobLogger.log("<br>----------- JobThread Exception:" + errorMsg + "<br>----------- web job execute end(error) -----------");
              } finally {
                  // 终止操作暂不监控状态
                  if (tgParam != null && tgParam.getJobId() != -1) {
                      // callback handler info
                      if (!toStop) {
                          // commonm
                          TriggerCallbackThread.pushCallBack(new HandleCallbackParam(tgParam.getLogId(), tgParam.getLogDateTime(), executeResult));
                      } else {
                          // is killed
                          ReturnT<String> stopResult = new ReturnT<String>(ReturnT.FAIL_CODE, stopReason + " [job running, killed]");
                          TriggerCallbackThread.pushCallBack(new HandleCallbackParam(tgParam.getLogId(), tgParam.getLogDateTime(), stopResult));
                      }
                  }
              }
          }
  
          // callback trigger request in queue
          while (triggerQueue != null && triggerQueue.size() > 0) {
              TriggerParam triggerParam = triggerQueue.poll();
              if (triggerParam != null) {
                  // is killed
                  ReturnT<String> stopResult = new ReturnT<String>(ReturnT.FAIL_CODE, stopReason + " [job not executed, in the job queue, killed.]");
                  TriggerCallbackThread.pushCallBack(new HandleCallbackParam(triggerParam.getLogId(), triggerParam.getLogDateTime(), stopResult));
              }
          }
  
          // destroy
          try {
              handler.destroy();
          } catch (Throwable e) {
              logger.error(e.getMessage(), e);
          }
  
          logger.info(">>>>>>>>>>> web JobThread stoped, hashCode:{}", Thread.currentThread());
      }
  }
  