content: package com.xxl.job.core.thread;
  
  import com.xxl.job.core.biz.model.HandleCallbackParam;
  import com.xxl.job.core.biz.model.ReturnT;
  import com.xxl.job.core.biz.model.TriggerParam;
  import com.xxl.job.core.executor.XxlJobExecutor;
  import com.xxl.job.core.handler.IJobHandler;
  import com.xxl.job.core.log.XxlJobFileAppender;
  import com.xxl.job.core.log.XxlJobLogger;
  import com.xxl.job.core.util.ShardingUtil;
  import org.eclipse.jetty.util.ConcurrentHashSet;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.io.PrintWriter;
  import java.io.StringWriter;
  import java.util.Arrays;
  import java.util.Date;
  import java.util.concurrent.LinkedBlockingQueue;
  import java.util.concurrent.TimeUnit;
  
  /**
   * handler thread
   * @author xuxueli 2016-1-16 19:52:47
   */
  public class JobThread extends Thread{
  	private static Logger logger = LoggerFactory.getLogger(JobThread.class);
  
  	private int jobId;
  	private IJobHandler handler;
  	private LinkedBlockingQueue<TriggerParam> triggerQueue;
  	private ConcurrentHashSet<Integer> triggerLogIdSet;		// avoid repeat trigger for the same TRIGGER_LOG_ID
  
  	private 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<TriggerParam>();
  		this.triggerLogIdSet = new ConcurrentHashSet<Integer>();
  	}
  	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<String>(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 triggerParam = null;
              ReturnT<String> executeResult = null;
              try {
  				// to check toStop signal, we need cycle, so wo cannot use queue.take(), instand of poll(timeout)
  				triggerParam = triggerQueue.poll(3L, TimeUnit.SECONDS);
  				if (triggerParam!=null) {
  					running = true;
  					idleTimes = 0;
  					triggerLogIdSet.remove(triggerParam.getLogId());
  
  					// log filename, like "logPath/yyyy-MM-dd/9999.log"
  					String logFileName = XxlJobFileAppender.makeLogFileName(new Date(triggerParam.getLogDateTim()), triggerParam.getLogId());
  					XxlJobFileAppender.contextHolder.set(logFileName);
  					ShardingUtil.setShardingVo(new ShardingUtil.ShardingVO(triggerParam.getBroadcastIndex(), triggerParam.getBroadcastTotal()));
  
  					// execute
  					XxlJobLogger.log("<br>----------- xxl-job job execute start -----------<br>----------- Param:" + triggerParam.getExecutorParams());
  					executeResult = handler.execute(triggerParam.getExecutorParams());
  					if (executeResult == null) {
  						executeResult = IJobHandler.FAIL;
  					}
  					XxlJobLogger.log("<br>----------- xxl-job job execute end(finish) -----------<br>----------- ReturnT:" + executeResult);
  
  				} else {
  					if (idleTimes > 30) {
  						XxlJobExecutor.removeJobThread(jobId, "excutor idel times over limit.");
  					}
  				}
  			} catch (Throwable e) {
  				if (toStop) {
  					XxlJobLogger.log("<br>----------- JobThread toStop, stopReason:" + stopReason);
  				}
  
  				StringWriter stringWriter = new StringWriter();
  				e.printStackTrace(new PrintWriter(stringWriter));
  				String errorMsg = stringWriter.toString();
  				executeResult = new ReturnT<String>(ReturnT.FAIL_CODE, errorMsg);
  
  				XxlJobLogger.log("<br>----------- JobThread Exception:" + errorMsg + "<br>----------- xxl-job job execute end(error) -----------");
  			} finally {
                  if(triggerParam != null) {
                      // callback handler info
                      if (!toStop) {
                          // commonm
                          TriggerCallbackThread.pushCallBack(new HandleCallbackParam(triggerParam.getLogId(), executeResult));
                      } else {
                          // is killed
                          ReturnT<String> stopResult = new ReturnT<String>(ReturnT.FAIL_CODE, stopReason + " [业务运行中，被强制终止]");
                          TriggerCallbackThread.pushCallBack(new HandleCallbackParam(triggerParam.getLogId(), 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 + " [任务尚未执行，在调度队列中被终止]");
  				TriggerCallbackThread.pushCallBack(new HandleCallbackParam(triggerParam.getLogId(), stopResult));
  			}
  		}
  
  		// destroy
  		try {
  			handler.destroy();
  		} catch (Throwable e) {
  			logger.error(e.getMessage(), e);
  		}
  
  		logger.info(">>>>>>>>>>> xxl-job JobThread stoped, hashCode:{}", Thread.currentThread());
  	}
  }
  