content: package com.xxl.job.admin.core.thread;
  
  import com.xxl.job.admin.core.conf.XxlJobAdminConfig;
  import com.xxl.job.admin.core.trigger.TriggerTypeEnum;
  import com.xxl.job.admin.core.trigger.XxlJobTrigger;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.util.concurrent.*;
  import java.util.concurrent.atomic.AtomicInteger;
  
  /**
   * job trigger thread pool helper
   *
   * @author xuxueli 2018-07-03 21:08:07
   */
  public class JobTriggerPoolHelper {
      private static Logger logger = LoggerFactory.getLogger(JobTriggerPoolHelper.class);
  
  
      // ---------------------- trigger pool ----------------------
  
      // fast/slow thread pool
      private ThreadPoolExecutor fastTriggerPool = null;
      private ThreadPoolExecutor slowTriggerPool = null;
  
      public void start(){
          fastTriggerPool = new ThreadPoolExecutor(
                  10,
                  XxlJobAdminConfig.getAdminConfig().getTriggerPoolFastMax(),
                  60L,
                  TimeUnit.SECONDS,
                  new LinkedBlockingQueue<Runnable>(1000),
                  new ThreadFactory() {
                      @Override
                      public Thread newThread(Runnable r) {
                          return new Thread(r, "xxl-job, admin JobTriggerPoolHelper-fastTriggerPool-" + r.hashCode());
                      }
                  });
  
          slowTriggerPool = new ThreadPoolExecutor(
                  10,
                  XxlJobAdminConfig.getAdminConfig().getTriggerPoolSlowMax(),
                  60L,
                  TimeUnit.SECONDS,
                  new LinkedBlockingQueue<Runnable>(2000),
                  new ThreadFactory() {
                      @Override
                      public Thread newThread(Runnable r) {
                          return new Thread(r, "xxl-job, admin JobTriggerPoolHelper-slowTriggerPool-" + r.hashCode());
                      }
                  });
      }
  
  
      public void stop() {
          //triggerPool.shutdown();
          fastTriggerPool.shutdownNow();
          slowTriggerPool.shutdownNow();
          logger.info(">>>>>>>>> xxl-job trigger thread pool shutdown success.");
      }
  
  
      // job timeout count
      private volatile long minTim = System.currentTimeMillis()/60000;     // ms > min
      private volatile ConcurrentMap<Integer, AtomicInteger> jobTimeoutCountMap = new ConcurrentHashMap<>();
  
  
      /**
       * add trigger
       */
      public void addTrigger(final int jobId,
                             final TriggerTypeEnum triggerType,
                             final int failRetryCount,
                             final String executorShardingParam,
                             final String executorParam,
                             final String addressList) {
  
          // choose thread pool
          ThreadPoolExecutor triggerPool_ = fastTriggerPool;
          AtomicInteger jobTimeoutCount = jobTimeoutCountMap.get(jobId);
          if (jobTimeoutCount!=null && jobTimeoutCount.get() > 10) {      // job-timeout 10 times in 1 min
              triggerPool_ = slowTriggerPool;
          }
  
          // trigger
          triggerPool_.execute(new Runnable() {
              @Override
              public void run() {
  
                  long start = System.currentTimeMillis();
  
                  try {
                      // do trigger
                      XxlJobTrigger.trigger(jobId, triggerType, failRetryCount, executorShardingParam, executorParam, addressList);
                  } catch (Exception e) {
                      logger.error(e.getMessage(), e);
                  } finally {
  
                      // check timeout-count-map
                      long minTim_now = System.currentTimeMillis()/60000;
                      if (minTim != minTim_now) {
                          minTim = minTim_now;
                          jobTimeoutCountMap.clear();
                      }
  
                      // incr timeout-count-map
                      long cost = System.currentTimeMillis()-start;
                      if (cost > 500) {       // ob-timeout threshold 500ms
                          AtomicInteger timeoutCount = jobTimeoutCountMap.putIfAbsent(jobId, new AtomicInteger(1));
                          if (timeoutCount != null) {
                              timeoutCount.incrementAndGet();
                          }
                      }
  
                  }
  
              }
          });
      }
  
  
  
      // ---------------------- helper ----------------------
  
      private static JobTriggerPoolHelper helper = new JobTriggerPoolHelper();
  
      public static void toStart() {
          helper.start();
      }
      public static void toStop() {
          helper.stop();
      }
  
      /**
       * @param jobId
       * @param triggerType
       * @param failRetryCount
       * 			>=0: use this param
       * 			<0: use param from job info config
       * @param executorShardingParam
       * @param executorParam
       *          null: use job param
       *          not null: cover job param
       */
      public static void trigger(int jobId, TriggerTypeEnum triggerType, int failRetryCount, String executorShardingParam, String executorParam, String addressList) {
          helper.addTrigger(jobId, triggerType, failRetryCount, executorShardingParam, executorParam, addressList);
      }
  
  }
  