content: package com.xxl.job.admin.core.thread;
  
  import com.xxl.job.admin.core.complete.XxlJobCompleter;
  import com.xxl.job.admin.core.conf.XxlJobAdminConfig;
  import com.xxl.job.admin.core.model.XxlJobLog;
  import com.xxl.job.admin.core.util.I18nUtil;
  import com.xxl.job.core.biz.model.HandleCallbackParam;
  import com.xxl.job.core.biz.model.ReturnT;
  import com.xxl.job.core.util.DateUtil;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.util.Date;
  import java.util.List;
  import java.util.concurrent.*;
  
  /**
   * job lose-monitor instance
   *
   * @author xuxueli 2015-9-1 18:05:56
   */
  public class JobCompleteHelper {
  
  	private static Logger logger = LoggerFactory.getLogger(JobCompleteHelper.class);
  
  	private static JobCompleteHelper instance = new JobCompleteHelper();
  
  	public static JobCompleteHelper getInstance() {
  		return instance;
  	}
  
  	// ---------------------- monitor ----------------------
  
  	private ThreadPoolExecutor callbackThreadPool = null;
  
  	private Thread monitorThread;
  
  	private volatile boolean toStop = false;
  
  	public void start() {
  
  		// for callback
  		callbackThreadPool = new ThreadPoolExecutor(2, 20, 30L, TimeUnit.SECONDS,
  				new LinkedBlockingQueue<Runnable>(3000), new ThreadFactory() {
  					@Override
  					public Thread newThread(Runnable r) {
  						return new Thread(r, "xxl-job, admin JobLosedMonitorHelper-callbackThreadPool-" + r.hashCode());
  					}
  				}, new RejectedExecutionHandler() {
  					@Override
  					public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
  						r.run();
  						logger.warn(
  								">>>>>>>>>>> xxl-job, callback too fast, match threadpool rejected handler(run now).");
  					}
  				});
  
  		// for monitor
  		monitorThread = new Thread(new Runnable() {
  
  			@Override
  			public void run() {
  
  				// wait for JobTriggerPoolHelper-init
  				try {
  					TimeUnit.MILLISECONDS.sleep(50);
  				}
  				catch (InterruptedException e) {
  					if (!toStop) {
  						logger.error(e.getMessage(), e);
  					}
  				}
  
  				// monitor
  				while (!toStop) {
  					try {
  						// 任务结果丢失处理：调度记录停留在 "运行中" 状态超过10min，且对应执行器心跳注册失败不在线，则将本地调度主动标记失败；
  						Date losedTime = DateUtil.addMinutes(new Date(), -10);
  						List<Long> losedJobIds = XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao()
  								.findLostJobIds(losedTime);
  
  						if (losedJobIds != null && losedJobIds.size() > 0) {
  							for (Long logId : losedJobIds) {
  
  								XxlJobLog jobLog = new XxlJobLog();
  								jobLog.setId(logId);
  
  								jobLog.setHandleTime(new Date());
  								jobLog.setHandleCode(ReturnT.FAIL_CODE);
  								jobLog.setHandleMsg(I18nUtil.getString("joblog_lost_fail"));
  
  								XxlJobCompleter.updateHandleInfoAndFinish(jobLog);
  							}
  
  						}
  					}
  					catch (Exception e) {
  						if (!toStop) {
  							logger.error(">>>>>>>>>>> xxl-job, job fail monitor thread error:{}", e);
  						}
  					}
  
  					try {
  						TimeUnit.SECONDS.sleep(60);
  					}
  					catch (Exception e) {
  						if (!toStop) {
  							logger.error(e.getMessage(), e);
  						}
  					}
  
  				}
  
  				logger.info(">>>>>>>>>>> xxl-job, JobLosedMonitorHelper stop");
  
  			}
  		});
  		monitorThread.setDaemon(true);
  		monitorThread.setName("xxl-job, admin JobLosedMonitorHelper");
  		monitorThread.start();
  	}
  
  	public void toStop() {
  		toStop = true;
  
  		// stop registryOrRemoveThreadPool
  		callbackThreadPool.shutdownNow();
  
  		// stop monitorThread (interrupt and wait)
  		monitorThread.interrupt();
  		try {
  			monitorThread.join();
  		}
  		catch (InterruptedException e) {
  			logger.error(e.getMessage(), e);
  		}
  	}
  
  	// ---------------------- helper ----------------------
  
  	public ReturnT<String> callback(List<HandleCallbackParam> callbackParamList) {
  
  		callbackThreadPool.execute(new Runnable() {
  			@Override
  			public void run() {
  				for (HandleCallbackParam handleCallbackParam : callbackParamList) {
  					ReturnT<String> callbackResult = callback(handleCallbackParam);
  					logger.debug(">>>>>>>>> JobApiController.callback {}, handleCallbackParam={}, callbackResult={}",
  							(callbackResult.getCode() == ReturnT.SUCCESS_CODE ? "success" : "fail"),
  							handleCallbackParam, callbackResult);
  				}
  			}
  		});
  
  		return ReturnT.SUCCESS;
  	}
  
  	private ReturnT<String> callback(HandleCallbackParam handleCallbackParam) {
  		// valid log item
  		XxlJobLog log = XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().load(handleCallbackParam.getLogId());
  		if (log == null) {
  			return new ReturnT<String>(ReturnT.FAIL_CODE, "log item not found.");
  		}
  		if (log.getHandleCode() > 0) {
  			return new ReturnT<String>(ReturnT.FAIL_CODE, "log repeate callback."); // avoid
  																					// repeat
  																					// callback,
  																					// trigger
  																					// child
  																					// job
  																					// etc
  		}
  
  		// handle msg
  		StringBuffer handleMsg = new StringBuffer();
  		if (log.getHandleMsg() != null) {
  			handleMsg.append(log.getHandleMsg()).append("<br>");
  		}
  		if (handleCallbackParam.getHandleMsg() != null) {
  			handleMsg.append(handleCallbackParam.getHandleMsg());
  		}
  
  		// success, save log
  		log.setHandleTime(new Date());
  		log.setHandleCode(handleCallbackParam.getHandleCode());
  		log.setHandleMsg(handleMsg.toString());
  		XxlJobCompleter.updateHandleInfoAndFinish(log);
  
  		return ReturnT.SUCCESS;
  	}
  
  }
  