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);
  	}
  
  }
  