• xxl-job源码—调度器/执行器工作原理

    目录

    一、架构图

    1.1 功能架构图

    2.2 任务调度工作原理

    二、ER图

    三、调度器

    3.1 启动过程时序图

    3.2 启动过程核心代码解析

    3.2.1 启动初始化

    3.2.2 执行器健康检查

    3.2.3 任务执行失败告警

    3.2.4 调度线程池

    3.2.5 调度中心

    3.3 任务执行时序图

    3.4 任务执行核心代码解析

    3.4.1 根据路由策略定位到执行器

    3.4.2 定位到任务执行Handler&线程

    3.4.3 任务执行线程

    四、执行器

    4.1 启动过程时序图

    4.2 启动过程核心代码解析

    4.2.1 启动初始化

    4.2.2 执行器初始化

    4.2.3 任务执行回调

    4.3 执行器注册到调度中心时序图

    4.4 执行器注册到调度中心核心代码解析

    4.4.1 执行注册（30秒一次）

    一、架构图

    1.1 功能架构图

    2.2 任务调度工作原理

    二、ER图

    三、调度器

    3.1 启动过程时序图

    3.2 启动过程核心代码解析

    入口类：XxlJobAdminConfig（PS：启动初始化XxlJobScheduler）

    3.2.1 启动初始化

    调度器启动初始化：XxlJobScheduler

    1. package com.xxl.job.admin.core.scheduler;
    2.  
    3. import com.xxl.job.admin.core.conf.XxlJobAdminConfig;
    4. import com.xxl.job.admin.core.thread.*;
    5. import com.xxl.job.admin.core.util.I18nUtil;
    6. import com.xxl.job.core.biz.ExecutorBiz;
    7. import com.xxl.job.core.enums.ExecutorBlockStrategyEnum;
    8. import com.xxl.rpc.remoting.invoker.call.CallType;
    9. import com.xxl.rpc.remoting.invoker.reference.XxlRpcReferenceBean;
    10. import com.xxl.rpc.remoting.invoker.route.LoadBalance;
    11. import com.xxl.rpc.remoting.net.impl.netty_http.client.NettyHttpClient;
    12. import com.xxl.rpc.serialize.impl.HessianSerializer;
    13. import org.slf4j.Logger;
    14. import org.slf4j.LoggerFactory;
    15.  
    16. import java.util.concurrent.ConcurrentHashMap;
    17. import java.util.concurrent.ConcurrentMap;
    18.  
    19. /**
    20. * @author xuxueli 2018-10-28 00:18:17
    21. * Description：调度程序
    22. */
    23.  
    24. public class XxlJobScheduler  {
    25.     private static final Logger logger = LoggerFactory.getLogger(XxlJobScheduler.class);
    26.     
    27.     
    28.     public void init() throws Exception {
    29.         // init i18n
    30.         /**
    31.         * 主要作用：调度中心语言选择
    32.         */
    33.         initI18n();
    34.         
    35.         // admin registry monitor run
    36.         /**
    37.         * 主要作用是开通了一个守护线程，每隔30s扫描一次执行器的注册信息表
    38.         * 1、剔除90s内没有进行健康检查的执行器信息
    39.         * 2、将自动注册类型的执行器注册信息（XxlJobRegistry）经过处理更新执行器信息(XxlJobGroup)
    40.         */
    41.         JobRegistryMonitorHelper.getInstance().start();
    42.         
    43.         // admin monitor run
    44.         /**
    45.         * 主要作用：开通了一个守10s扫描一次失败日志
    46.         * 1、如果任务失败可重试次数>0，那么重新触发任务
    47.         * 2、如果任务执行失败，会进行告警，默认采用邮件形式进行告警
    48.         */
    49.         JobFailMonitorHelper.getInstance().start();
    50.         
    51.         // admin trigger pool start
    52.         /**
    53.         * 主要作用：初始化一个快速执行的线程池，一个稍后执行的线程池
    54.         *
    55.         */
    56.         JobTriggerPoolHelper.toStart();
    57.         
    58.         // admin log report start
    59.         /**
    60.         * 主要作用：开通一个守护线程，每隔1min扫描一次最近3天的调度日志
    61.         * 1、更新每天总任务数、正在执行数、执行成功数、执行失败数
    62.         */
    63.         JobLogReportHelper.getInstance().start();
    64.         
    65.         // start-schedule
    66.         /**
    67.         * 主要作用：
    68.         * 1、开通一个守护线程，每隔5s扫表一次执行器的任务表
    69.         *   1.1 执行已到执行时间的任务，并更新任务的下次执行时间
    70.         *
    71.         * 2、开通一个守护线程，每隔1s循环执行一次待执行的任务
    72.         *  2.1 避免任务执行遗漏
    73.         */
    74.         JobScheduleHelper.getInstance().start();
    75.         
    76.         logger.info(">>>>>>>>> init xxl-job admin success.");
    77.     }
    78.     
    79.     
    80.     public void destroy() throws Exception {
    81.         
    82.         // stop-schedule
    83.         JobScheduleHelper.getInstance().toStop();
    84.         
    85.         // admin log report stop
    86.         JobLogReportHelper.getInstance().toStop();
    87.         
    88.         // admin trigger pool stop
    89.         JobTriggerPoolHelper.toStop();
    90.         
    91.         // admin monitor stop
    92.         JobFailMonitorHelper.getInstance().toStop();
    93.         
    94.         // admin registry stop
    95.         JobRegistryMonitorHelper.getInstance().toStop();
    96.         
    97.     }
    98.     
    99.     // ---------------------- I18n ----------------------
    100.     
    101.     private void initI18n(){
    102.         for (ExecutorBlockStrategyEnum item:ExecutorBlockStrategyEnum.values()) {
    103.             // 根据语言选择，去对应的message_xxx.properties中取阻塞策略的title
    104.             item.setTitle(I18nUtil.getString("jobconf_block_".concat(item.name())));
    105.         }
    106.     }
    107.     
    108.     // ---------------------- executor-client ----------------------
    109.     private static ConcurrentMap executorBizRepository = new ConcurrentHashMap();
    110.     public static ExecutorBiz getExecutorBiz(String address) throws Exception {
    111.         // valid
    112.         if (address==null || address.trim().length()==0) {
    113.             return null;
    114.         }
    115.         
    116.         // load-cache
    117.         address = address.trim();
    118.         ExecutorBiz executorBiz = executorBizRepository.get(address);
    119.         if (executorBiz != null) {
    120.             return executorBiz;
    121.         }
    122.         
    123.         // set-cache
    124.         // 创建ExecutorBiz的代理对象，重点在这个getObject方法
    125.         XxlRpcReferenceBean referenceBean = new XxlRpcReferenceBean();
    126.         referenceBean.setClient(NettyHttpClient.class);
    127.         referenceBean.setSerializer(HessianSerializer.class);
    128.         referenceBean.setCallType(CallType.SYNC);
    129.         referenceBean.setLoadBalance(LoadBalance.ROUND);
    130.         referenceBean.setIface(ExecutorBiz.class);
    131.         referenceBean.setVersion(null);
    132.         referenceBean.setTimeout(3000);
    133.         referenceBean.setAddress(address);
    134.         referenceBean.setAccessToken(XxlJobAdminConfig.getAdminConfig().getAccessToken());
    135.         referenceBean.setInvokeCallback(null);
    136.         referenceBean.setInvokerFactory(null);
    137.         
    138.         executorBiz = (ExecutorBiz) referenceBean.getObject();
    139.         
    140.         executorBizRepository.put(address, executorBiz);
    141.         return executorBiz;
    142.     }
    143.     
    144. }

    3.2.2 执行器健康检查

    执行器注册健康检查：JobRegistryMonitorHelper

    1. package com.xxl.job.admin.core.thread;
    2.  
    3. import com.xxl.job.admin.core.conf.XxlJobAdminConfig;
    4. import com.xxl.job.admin.core.model.XxlJobGroup;
    5. import com.xxl.job.admin.core.model.XxlJobRegistry;
    6. import com.xxl.job.core.enums.RegistryConfig;
    7. import org.slf4j.Logger;
    8. import org.slf4j.LoggerFactory;
    9.  
    10. import java.util.*;
    11. import java.util.concurrent.TimeUnit;
    12.  
    13. /**
    14.  * job registry instance
    15.  * @author xuxueli 2016-10-02 19:10:24
    16.  * @Description：任务注册监控助手
    17.  */
    18. public class JobRegistryMonitorHelper {
    19. 	private static Logger logger = LoggerFactory.getLogger(JobRegistryMonitorHelper.class);
    20.  
    21. 	private static JobRegistryMonitorHelper instance = new JobRegistryMonitorHelper();
    22. 	public static JobRegistryMonitorHelper getInstance(){
    23. 		return instance;
    24. 	}
    25.  
    26. 	private Thread registryThread;
    27. 	private volatile boolean toStop = false;
    28. 	public void start(){
    29. 		//创建一个线程
    30. 		registryThread = new Thread(new Runnable() {
    31. 			@Override
    32. 			public void run() {
    33. 				//当toStop为false时进入该循环（注意：toStop是用volatile修饰的）
    34. 				while (!toStop) {
    35. 					try {
    36. 						// auto registry group
    37. 						//获取类型为自动注册的执行器（XxlJobGroup）地址列表
    38. 						List groupList = XxlJobAdminConfig.getAdminConfig().getXxlJobGroupDao().findByAddressType(0);
    39. 						if (groupList!=null && !groupList.isEmpty()) {
    40.  
    41. 							// remove dead address (admin/executor)
    42. 							//删除90秒内没有更新的注册机器信息，90秒没有心跳信息返回代表机器已经出现问题，所以移除
    43. 							List ids = XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().findDead(RegistryConfig.DEAD_TIMEOUT, new Date());
    44. 							if (ids!=null && ids.size()>0) {
    45. 								XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().removeDead(ids);
    46. 							}
    47.  
    48. 							// fresh online address (admin/executor)
    49. 							HashMap> appAddressMap = new HashMap>();
    50. 							//查询90秒内有更新的注册机器信息列表
    51. 							List list = XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().findAll(RegistryConfig.DEAD_TIMEOUT, new Date());
    52. 							if (list != null) {
    53. 								//遍历注册信息列表，得到自动注册类型的执行器与其对应的地址信息关系Map
    54. 								for (XxlJobRegistry item: list) {
    55. 									if (RegistryConfig.RegistType.EXECUTOR.name().equals(item.getRegistryGroup())) {
    56. 										String appName = item.getRegistryKey();
    57. 										List registryList = appAddressMap.get(appName);
    58. 										if (registryList == null) {
    59. 											registryList = new ArrayList();
    60. 										}
    61.  
    62. 										if (!registryList.contains(item.getRegistryValue())) {
    63. 											registryList.add(item.getRegistryValue());
    64. 										}
    65. 										//收集执行器信息，根据执行器appName做区分：appAddressMap的key为appName，value为此执行器的注册地址列表（集群环境下会有多个注册地址）
    66. 										appAddressMap.put(appName, registryList);
    67. 									}
    68. 								}
    69. 							}
    70.  
    71. 							// fresh group address
    72. 							//遍历所有的自动注册的执行器
    73. 							for (XxlJobGroup group: groupList) {
    74. 								//通过执行器的appName从刚刚区分的Map中拿到该执行器下的集群机器注册地址
    75. 								List registryList = appAddressMap.get(group.getAppName());
    76. 								String addressListStr = null;
    77. 								if (registryList!=null && !registryList.isEmpty()) {
    78. 									Collections.sort(registryList);
    79. 									addressListStr = "";
    80. 									//集群的多个注册地址通过逗号拼接转为字符串
    81. 									for (String item:registryList) {
    82. 										addressListStr += item + ",";
    83. 									}
    84. 									addressListStr = addressListStr.substring(0, addressListStr.length()-1);
    85. 								}
    86. 								//集群的多个注册地址通过逗号拼接转成的字符串设置进执行器的addressList属性中，并更新执行器信息，保存到DB
    87. 								group.setAddressList(addressListStr);
    88. 								XxlJobAdminConfig.getAdminConfig().getXxlJobGroupDao().update(group);
    89. 							}
    90. 						}
    91. 					} catch (Exception e) {
    92. 						if (!toStop) {
    93. 							logger.error(">>>>>>>>>>> xxl-job, job registry monitor thread error:{}", e);
    94. 						}
    95. 					}
    96. 					try {
    97. 						//线程停顿30秒
    98. 						TimeUnit.SECONDS.sleep(RegistryConfig.BEAT_TIMEOUT);
    99. 					} catch (InterruptedException e) {
    100. 						if (!toStop) {
    101. 							logger.error(">>>>>>>>>>> xxl-job, job registry monitor thread error:{}", e);
    102. 						}
    103. 					}
    104. 				}
    105. 				logger.info(">>>>>>>>>>> xxl-job, job registry monitor thread stop");
    106. 			}
    107. 		});
    108. 		//将此线程设置成守护线程
    109. 		registryThread.setDaemon(true);
    110. 		registryThread.setName("xxl-job, admin JobRegistryMonitorHelper");
    111. 		//执行该线程
    112. 		registryThread.start();
    113. 	}
    114.  
    115. 	public void toStop(){
    116. 		toStop = true;
    117. 		// interrupt and wait
    118. 		registryThread.interrupt();
    119. 		try {
    120. 			registryThread.join();
    121. 		} catch (InterruptedException e) {
    122. 			logger.error(e.getMessage(), e);
    123. 		}
    124. 	}
    125.  
    126. }

    3.2.3 任务执行失败告警

    任务失败监控助手：JobFailMonitorHelper

    1. package com.xxl.job.admin.core.thread;
    2.  
    3. import com.xxl.job.admin.core.conf.XxlJobAdminConfig;
    4. import com.xxl.job.admin.core.model.XxlJobInfo;
    5. import com.xxl.job.admin.core.model.XxlJobLog;
    6. import com.xxl.job.admin.core.trigger.TriggerTypeEnum;
    7. import com.xxl.job.admin.core.util.I18nUtil;
    8. import org.slf4j.Logger;
    9. import org.slf4j.LoggerFactory;
    10.  
    11. import java.util.List;
    12. import java.util.concurrent.TimeUnit;
    13.  
    14. /**
    15.  * job monitor instance
    16.  *
    17.  * @author xuxueli 2015-9-1 18:05:56
    18.  * @Description：任务失败监控助手
    19.  */
    20. public class JobFailMonitorHelper {
    21. 	private static Logger logger = LoggerFactory.getLogger(JobFailMonitorHelper.class);
    22. 	
    23. 	private static JobFailMonitorHelper instance = new JobFailMonitorHelper();
    24. 	public static JobFailMonitorHelper getInstance(){
    25. 		return instance;
    26. 	}
    27.  
    28. 	// ---------------------- monitor ----------------------
    29.  
    30. 	private Thread monitorThread;
    31. 	private volatile boolean toStop = false;
    32. 	public void start(){
    33. 		//创建一个监控线程
    34. 		monitorThread = new Thread(new Runnable() {
    35.  
    36. 			@Override
    37. 			public void run() {
    38.  
    39. 				// monitor
    40. 				while (!toStop) {
    41. 					try {
    42. 						//从数据库中取出所有执行失败且告警状态为0（默认）的日志ID，虽然这里传了pageSize为1000，实际没有进行分页，取出来的是所有符合条件的失败日志
    43. 						List failLogIds = XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().findFailJobLogIds(1000);
    44. 						if (failLogIds!=null && !failLogIds.isEmpty()) {
    45. 							for (long failLogId: failLogIds) {
    46.  
    47. 								// 锁定日志
    48. 								int lockRet = XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().updateAlarmStatus(failLogId, 0, -1);
    49. 								if (lockRet < 1) {
    50. 									continue;
    51. 								}
    52. 								//取出失败日志完整信息
    53. 								XxlJobLog log = XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().load(failLogId);
    54. 								//获取失败日志对应的任务信息
    55. 								XxlJobInfo info = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().loadById(log.getJobId());
    56.  
    57. 								// 如果剩余失败可重试次数>0（注意：日志里存的失败重试次数实为剩余可重复次数）
    58. 								if (log.getExecutorFailRetryCount() > 0) {
    59. 									//触发任务执行
    60. 									JobTriggerPoolHelper.trigger(log.getJobId(), TriggerTypeEnum.RETRY, (log.getExecutorFailRetryCount()-1), log.getExecutorShardingParam(), log.getExecutorParam());
    61. 									String retryMsg = "

       >>>>>>>>>>>"+ I18nUtil.getString("jobconf_trigger_type_retry") +"<<<<<<<<<<< 
      "      ;
    62. 									log.setTriggerMsg(log.getTriggerMsg() + retryMsg);
    63. 									XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().updateTriggerInfo(log);
    64. 								}
    65.  
    66. 								// 2、fail alarm monitor
    67. 								//失败告警
    68. 								int newAlarmStatus = 0;		// 告警状态：0-默认、-1=锁定状态、1-无需告警、2-告警成功、3-告警失败
    69. 								if (info!=null && info.getAlarmEmail()!=null && info.getAlarmEmail().trim().length()>0) {
    70. 									//发送失败告警：默认是邮件通知
    71. 									boolean alarmResult = XxlJobAdminConfig.getAdminConfig().getJobAlarmer().alarm(info, log);
    72. 									newAlarmStatus = alarmResult?2:3;
    73. 								} else {
    74. 									newAlarmStatus = 1;
    75. 								}
    76. 								//更新告警状态
    77. 								XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().updateAlarmStatus(failLogId, -1, newAlarmStatus);
    78. 							}
    79. 						}
    80.  
    81. 					} catch (Exception e) {
    82. 						if (!toStop) {
    83. 							logger.error(">>>>>>>>>>> xxl-job, job fail monitor thread error:{}", e);
    84. 						}
    85. 					}
    86.  
    87.                     try {
    88.                         TimeUnit.SECONDS.sleep(10);
    89.                     } catch (Exception e) {
    90.                         if (!toStop) {
    91.                             logger.error(e.getMessage(), e);
    92.                         }
    93.                     }
    94.  
    95.                 }
    96.  
    97. 				logger.info(">>>>>>>>>>> xxl-job, job fail monitor thread stop");
    98.  
    99. 			}
    100. 		});
    101. 		monitorThread.setDaemon(true);
    102. 		monitorThread.setName("xxl-job, admin JobFailMonitorHelper");
    103. 		monitorThread.start();
    104. 	}
    105.  
    106. 	public void toStop(){
    107. 		toStop = true;
    108. 		// interrupt and wait
    109. 		monitorThread.interrupt();
    110. 		try {
    111. 			monitorThread.join();
    112. 		} catch (InterruptedException e) {
    113. 			logger.error(e.getMessage(), e);
    114. 		}
    115. 	}
    116.  
    117. }

    3.2.4 调度线程池

    分配任务执行线程：JobTriggerPoolHelper

    1. package com.xxl.job.admin.core.thread;
    2.  
    3. import com.xxl.job.admin.core.conf.XxlJobAdminConfig;
    4. import com.xxl.job.admin.core.trigger.TriggerTypeEnum;
    5. import com.xxl.job.admin.core.trigger.XxlJobTrigger;
    6. import org.slf4j.Logger;
    7. import org.slf4j.LoggerFactory;
    8.  
    9. import java.util.concurrent.*;
    10. import java.util.concurrent.atomic.AtomicInteger;
    11.  
    12. /**
    13.  * job trigger thread pool helper
    14.  *
    15.  * @author xuxueli 2018-07-03 21:08:07
    16.  */
    17. public class JobTriggerPoolHelper {
    18.     private static Logger logger = LoggerFactory.getLogger(JobTriggerPoolHelper.class);
    19.  
    20.  
    21.     // ---------------------- trigger pool ----------------------
    22.  
    23.     // fast/slow thread pool
    24.     private ThreadPoolExecutor fastTriggerPool = null;
    25.     private ThreadPoolExecutor slowTriggerPool = null;
    26.  
    27.     public void start(){
    28.         fastTriggerPool = new ThreadPoolExecutor(
    29.                 10,
    30.                 XxlJobAdminConfig.getAdminConfig().getTriggerPoolFastMax(),
    31.                 60L,
    32.                 TimeUnit.SECONDS,
    33.                 new LinkedBlockingQueue(1000),
    34.                 new ThreadFactory() {
    35.                     @Override
    36.                     public Thread newThread(Runnable r) {
    37.                         return new Thread(r, "xxl-job, admin JobTriggerPoolHelper-fastTriggerPool-" + r.hashCode());
    38.                     }
    39.                 });
    40.  
    41.         slowTriggerPool = new ThreadPoolExecutor(
    42.                 10,
    43.                 XxlJobAdminConfig.getAdminConfig().getTriggerPoolSlowMax(),
    44.                 60L,
    45.                 TimeUnit.SECONDS,
    46.                 new LinkedBlockingQueue(2000),
    47.                 new ThreadFactory() {
    48.                     @Override
    49.                     public Thread newThread(Runnable r) {
    50.                         return new Thread(r, "xxl-job, admin JobTriggerPoolHelper-slowTriggerPool-" + r.hashCode());
    51.                     }
    52.                 });
    53.     }
    54.  
    55.  
    56.     public void stop() {
    57.         //triggerPool.shutdown();
    58.         fastTriggerPool.shutdownNow();
    59.         slowTriggerPool.shutdownNow();
    60.         logger.info(">>>>>>>>> xxl-job trigger thread pool shutdown success.");
    61.     }
    62.  
    63.  
    64.     // job timeout count
    65.     private volatile long minTim = System.currentTimeMillis()/60000;     // ms > min
    66.     private volatile ConcurrentMap jobTimeoutCountMap = new ConcurrentHashMap<>();
    67.  
    68.  
    69.     /**
    70.      * add trigger
    71.      */
    72.     public void addTrigger(final int jobId, final TriggerTypeEnum triggerType, final int failRetryCount, final String executorShardingParam, final String executorParam) {
    73.  
    74.         // choose thread pool
    75.         //选择线程池类型
    76.         ThreadPoolExecutor triggerPool_ = fastTriggerPool;
    77.         AtomicInteger jobTimeoutCount = jobTimeoutCountMap.get(jobId);
    78.         // 1分钟窗口期内任务耗时达500ms超过10次则判定为慢任务，慢任务自动降级进入"Slow"线程池，避免耗尽调度线程，提高系统稳定性；
    79.         if (jobTimeoutCount!=null && jobTimeoutCount.get() > 10) {      // job-timeout 10 times in 1 min
    80.             triggerPool_ = slowTriggerPool;
    81.         }
    82.  
    83.         // trigger
    84.         triggerPool_.execute(new Runnable() {
    85.             @Override
    86.             public void run() {
    87.  
    88.                 long start = System.currentTimeMillis();
    89.  
    90.                 try {
    91.                     // do trigger
    92.                     //这里是重点，触发任务的执行
    93.                     XxlJobTrigger.trigger(jobId, triggerType, failRetryCount, executorShardingParam, executorParam);
    94.                 } catch (Exception e) {
    95.                     logger.error(e.getMessage(), e);
    96.                 } finally {
    97.                     //在finally代码块中通过执行耗时时间是否>500ms将jobId与超时次数记录到到超时记录容器
    98.                     // check timeout-count-map
    99.                     //检查超时记录容器
    100.                     long minTim_now = System.currentTimeMillis()/60000;
    101.                     if (minTim != minTim_now) {
    102.                         //说明当前分钟数和上面的默认分钟数不同，即超过了1分钟
    103.                         //这里的逻辑主要是为了使得超时记录容器以1分钟为时间间隔，也就是说超时容器1分钟记录1次超时的任务，到了下一分钟清空容器，重新记录
    104.                         minTim = minTim_now;
    105.                         jobTimeoutCountMap.clear();
    106.                     }
    107.  
    108.                     // incr timeout-count-map
    109.                     //如果线程执行时间>500毫秒：将此任务ID与次数记录进超时记录容器jobTimeoutCountMap
    110.                     long cost = System.currentTimeMillis()-start;
    111.                     if (cost > 500) {       // ob-timeout threshold 500ms
    112.                         //这里利用AtomicInteger的原子性，保证多线程并发结果的准确性
    113.                         AtomicInteger timeoutCount = jobTimeoutCountMap.putIfAbsent(jobId, new AtomicInteger(1));
    114.                         if (timeoutCount != null) {
    115.                             timeoutCount.incrementAndGet();
    116.                         }
    117.                     }
    118.  
    119.                 }
    120.  
    121.             }
    122.         });
    123.     }
    124.  
    125.  
    126.  
    127.     // ---------------------- helper ----------------------
    128.  
    129.     private static JobTriggerPoolHelper helper = new JobTriggerPoolHelper();
    130.  
    131.     public static void toStart() {
    132.         helper.start();
    133.     }
    134.     public static void toStop() {
    135.         helper.stop();
    136.     }
    137.  
    138.     /**
    139.      * @param jobId
    140.      * @param triggerType
    141.      * @param failRetryCount
    142.      * 			>=0: use this param
    143.      * 			<0: use param from job info config
    144.      * @param executorShardingParam
    145.      * @param executorParam
    146.      *          null: use job param
    147.      *          not null: cover job param
    148.      */
    149.     public static void trigger(int jobId, TriggerTypeEnum triggerType, int failRetryCount, String executorShardingParam, String executorParam) {
    150.         helper.addTrigger(jobId, triggerType, failRetryCount, executorShardingParam, executorParam);
    151.     }
    152.  
    153. }

    3.2.5 调度中心

    核心调度程序：JobScheduleHelper

    1. package com.xxl.job.admin.core.thread;
    2.  
    3. import com.xxl.job.admin.core.conf.XxlJobAdminConfig;
    4. import com.xxl.job.admin.core.cron.CronExpression;
    5. import com.xxl.job.admin.core.model.XxlJobInfo;
    6. import com.xxl.job.admin.core.trigger.TriggerTypeEnum;
    7. import org.slf4j.Logger;
    8. import org.slf4j.LoggerFactory;
    9.  
    10. import java.sql.Connection;
    11. import java.sql.PreparedStatement;
    12. import java.sql.SQLException;
    13. import java.text.ParseException;
    14. import java.util.*;
    15. import java.util.concurrent.ConcurrentHashMap;
    16. import java.util.concurrent.TimeUnit;
    17.  
    18. /**
    19.  * @author xuxueli 2019-05-21
    20.  * @Description：调度中心
    21.  */
    22. public class JobScheduleHelper {
    23.     private static Logger logger = LoggerFactory.getLogger(JobScheduleHelper.class);
    24.  
    25.     private static JobScheduleHelper instance = new JobScheduleHelper();
    26.     public static JobScheduleHelper getInstance(){
    27.         return instance;
    28.     }
    29.  
    30.     public static final long PRE_READ_MS = 5000;    // pre read
    31.  
    32.     private Thread scheduleThread;
    33.     private Thread ringThread;
    34.     private volatile boolean scheduleThreadToStop = false;
    35.     private volatile boolean ringThreadToStop = false;
    36.     private volatile static Map> ringData = new ConcurrentHashMap<>();
    37.  
    38.     public void start(){
    39.  
    40.         // schedule thread
    41.         scheduleThread = new Thread(new Runnable() {
    42.             @Override
    43.             public void run() {
    44.  
    45.                 try {
    46.                     TimeUnit.MILLISECONDS.sleep(5000 - System.currentTimeMillis()%1000 );
    47.                 } catch (InterruptedException e) {
    48.                     if (!scheduleThreadToStop) {
    49.                         logger.error(e.getMessage(), e);
    50.                     }
    51.                 }
    52.                 logger.info(">>>>>>>>> init xxl-job admin scheduler success.");
    53.  
    54.                 // pre-read count: treadpool-size * trigger-qps (each trigger cost 50ms, qps = 1000/50 = 20)
    55.                 // 预读触发数 = 线程数 * 触发执行qps（限制qps：20/s）
    56.                 int preReadCount = (XxlJobAdminConfig.getAdminConfig().getTriggerPoolFastMax() + XxlJobAdminConfig.getAdminConfig().getTriggerPoolSlowMax()) * 20;
    57.  
    58.                 while (!scheduleThreadToStop) {
    59.  
    60.                     // Scan Job
    61.                     long start = System.currentTimeMillis();
    62.  
    63.                     Connection conn = null;
    64.                     Boolean connAutoCommit = null;
    65.                     PreparedStatement preparedStatement = null;
    66.  
    67.                     boolean preReadSuc = true;
    68.                     try {
    69.  
    70.                         conn = XxlJobAdminConfig.getAdminConfig().getDataSource().getConnection();
    71.                         connAutoCommit = conn.getAutoCommit();
    72.                         conn.setAutoCommit(false);
    73.  
    74.                         // 调度中心支持多节点部署，基于数据库行锁保证同时只有一个调度中心节点触发任务调度
    75.                         preparedStatement = conn.prepareStatement(  "select * from xxl_job_lock where lock_name = 'schedule_lock' for update" );
    76.                         preparedStatement.execute();
    77.  
    78.                         // tx start
    79.  
    80.                         // 1、pre read
    81.                         long nowTime = System.currentTimeMillis();
    82.                         // 查询最近5s等待执行的任务（限制数量为preReadCount）
    83.                         List scheduleList = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().scheduleJobQuery(nowTime + PRE_READ_MS, preReadCount);
    84.                         if (scheduleList!=null && scheduleList.size()>0) {
    85.                             // 2、push time-ring
    86.                             for (XxlJobInfo jobInfo: scheduleList) {
    87.  
    88.                                 // time-ring jump
    89.                                 if (nowTime > jobInfo.getTriggerNextTime() + PRE_READ_MS) {
    90.                                     // 2.1、trigger-expire > 5s：pass && make next-trigger-time
    91.                                     // 任务触发时间已过，则跳过执行
    92.                                     logger.warn(">>>>>>>>>>> xxl-job, schedule misfire, jobId = " + jobInfo.getId());
    93.  
    94.                                     // fresh next
    95.                                     // 更新该任务下次执行时间
    96.                                     refreshNextValidTime(jobInfo, new Date());
    97.  
    98.                                 } else if (nowTime > jobInfo.getTriggerNextTime()) {
    99.                                     // 2.2、trigger-expire < 5s：direct-trigger && make next-trigger-time
    100.                                     // 任务触发时间已到，则立即执行
    101.                                     // 1、trigger
    102.                                     JobTriggerPoolHelper.trigger(jobInfo.getId(), TriggerTypeEnum.CRON, -1, null, null);
    103.                                     logger.debug(">>>>>>>>>>> xxl-job, schedule push trigger : jobId = " + jobInfo.getId() );
    104.  
    105.                                     // 2、fresh next
    106.                                     // 更新该任务下次执行时间
    107.                                     refreshNextValidTime(jobInfo, new Date());
    108.  
    109.                                     // next-trigger-time in 5s, pre-read again
    110.                                     // 任务下次触发时间在5s内，则需再执行一次
    111.                                     if (jobInfo.getTriggerStatus()==1 && nowTime + PRE_READ_MS > jobInfo.getTriggerNextTime()) {
    112.                                         // 针对待下次执行，任务会已过执行时间，因此必须当前立即执行，避免任务执行遗漏
    113.  
    114.                                         // 1、make ring second
    115.                                         // 计算不足1s的剩余毫秒
    116.                                         int ringSecond = (int)((jobInfo.getTriggerNextTime()/1000)%60);
    117.  
    118.                                         // 2、push time ring
    119.                                         // 存放待执行的循环执行容器
    120.                                         pushTimeRing(ringSecond, jobInfo.getId());
    121.  
    122.                                         // 3、fresh next
    123.                                         // 更新该任务下次执行时间
    124.                                         refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime()));
    125.  
    126.                                     }
    127.  
    128.                                 } else {
    129.                                     // 2.3、trigger-pre-read：time-ring trigger && make next-trigger-time
    130.  
    131.                                     // 1、make ring second
    132.                                     int ringSecond = (int)((jobInfo.getTriggerNextTime()/1000)%60);
    133.  
    134.                                     // 2、push time ring
    135.                                     pushTimeRing(ringSecond, jobInfo.getId());
    136.  
    137.                                     // 3、fresh next
    138.                                     refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime()));
    139.  
    140.                                 }
    141.  
    142.                             }
    143.  
    144.                             // 3、update trigger info
    145.                             for (XxlJobInfo jobInfo: scheduleList) {
    146.                                 XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().scheduleUpdate(jobInfo);
    147.                             }
    148.  
    149.                         } else {
    150.                             preReadSuc = false;
    151.                         }
    152.  
    153.                         // tx stop
    154.  
    155.  
    156.                     } catch (Exception e) {
    157.                         if (!scheduleThreadToStop) {
    158.                             logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#scheduleThread error:{}", e);
    159.                         }
    160.                     } finally {
    161.  
    162.                         // commit
    163.                         if (conn != null) {
    164.                             try {
    165.                                 conn.commit();
    166.                             } catch (SQLException e) {
    167.                                 if (!scheduleThreadToStop) {
    168.                                     logger.error(e.getMessage(), e);
    169.                                 }
    170.                             }
    171.                             try {
    172.                                 conn.setAutoCommit(connAutoCommit);
    173.                             } catch (SQLException e) {
    174.                                 if (!scheduleThreadToStop) {
    175.                                     logger.error(e.getMessage(), e);
    176.                                 }
    177.                             }
    178.                             try {
    179.                                 conn.close();
    180.                             } catch (SQLException e) {
    181.                                 if (!scheduleThreadToStop) {
    182.                                     logger.error(e.getMessage(), e);
    183.                                 }
    184.                             }
    185.                         }
    186.  
    187.                         // close PreparedStatement
    188.                         if (null != preparedStatement) {
    189.                             try {
    190.                                 preparedStatement.close();
    191.                             } catch (SQLException e) {
    192.                                 if (!scheduleThreadToStop) {
    193.                                     logger.error(e.getMessage(), e);
    194.                                 }
    195.                             }
    196.                         }
    197.                     }
    198.                     long cost = System.currentTimeMillis()-start;
    199.  
    200.  
    201.                     // Wait seconds, align second
    202.                     if (cost < 1000) {  // scan-overtime, not wait
    203.                         try {
    204.                             // pre-read period: success > scan each second; fail > skip this period;
    205.                             TimeUnit.MILLISECONDS.sleep((preReadSuc?1000:PRE_READ_MS) - System.currentTimeMillis()%1000);
    206.                         } catch (InterruptedException e) {
    207.                             if (!scheduleThreadToStop) {
    208.                                 logger.error(e.getMessage(), e);
    209.                             }
    210.                         }
    211.                     }
    212.  
    213.                 }
    214.  
    215.                 logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper#scheduleThread stop");
    216.             }
    217.         });
    218.         scheduleThread.setDaemon(true);
    219.         scheduleThread.setName("xxl-job, admin JobScheduleHelper#scheduleThread");
    220.         scheduleThread.start();
    221.  
    222.  
    223.         // ring thread
    224.         ringThread = new Thread(new Runnable() {
    225.             @Override
    226.             public void run() {
    227.  
    228.                 // align second
    229.                 try {
    230.                     TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis()%1000 );
    231.                 } catch (InterruptedException e) {
    232.                     if (!ringThreadToStop) {
    233.                         logger.error(e.getMessage(), e);
    234.                     }
    235.                 }
    236.  
    237.                 while (!ringThreadToStop) {
    238.  
    239.                     try {
    240.                         // second data
    241.                         List ringItemData = new ArrayList<>();
    242.                         int nowSecond = Calendar.getInstance().get(Calendar.SECOND);   // 避免处理耗时太长，跨过刻度，向前校验一个刻度；
    243.                         for (int i = 0; i < 2; i++) {
    244.                             List tmpData = ringData.remove( (nowSecond+60-i)%60 );
    245.                             if (tmpData != null) {
    246.                                 ringItemData.addAll(tmpData);
    247.                             }
    248.                         }
    249.  
    250.                         // ring trigger
    251.                         logger.debug(">>>>>>>>>>> xxl-job, time-ring beat : " + nowSecond + " = " + Arrays.asList(ringItemData) );
    252.                         if (ringItemData.size() > 0) {
    253.                             // do trigger
    254.                             for (int jobId: ringItemData) {
    255.                                 // do trigger
    256.                                 JobTriggerPoolHelper.trigger(jobId, TriggerTypeEnum.CRON, -1, null, null);
    257.                             }
    258.                             // clear
    259.                             ringItemData.clear();
    260.                         }
    261.                     } catch (Exception e) {
    262.                         if (!ringThreadToStop) {
    263.                             logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#ringThread error:{}", e);
    264.                         }
    265.                     }
    266.  
    267.                     // next second, align second
    268.                     try {
    269.                         TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis()%1000);
    270.                     } catch (InterruptedException e) {
    271.                         if (!ringThreadToStop) {
    272.                             logger.error(e.getMessage(), e);
    273.                         }
    274.                     }
    275.                 }
    276.                 logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper#ringThread stop");
    277.             }
    278.         });
    279.         ringThread.setDaemon(true);
    280.         ringThread.setName("xxl-job, admin JobScheduleHelper#ringThread");
    281.         ringThread.start();
    282.     }
    283.  
    284.     private void refreshNextValidTime(XxlJobInfo jobInfo, Date fromTime) throws ParseException {
    285.         Date nextValidTime = new CronExpression(jobInfo.getJobCron()).getNextValidTimeAfter(fromTime);
    286.         if (nextValidTime != null) {
    287.             jobInfo.setTriggerLastTime(jobInfo.getTriggerNextTime());
    288.             jobInfo.setTriggerNextTime(nextValidTime.getTime());
    289.         } else {
    290.             jobInfo.setTriggerStatus(0);
    291.             jobInfo.setTriggerLastTime(0);
    292.             jobInfo.setTriggerNextTime(0);
    293.         }
    294.     }
    295.  
    296.     private void pushTimeRing(int ringSecond, int jobId){
    297.         // push async ring
    298.         List ringItemData = ringData.get(ringSecond);
    299.         if (ringItemData == null) {
    300.             ringItemData = new ArrayList();
    301.             ringData.put(ringSecond, ringItemData);
    302.         }
    303.         ringItemData.add(jobId);
    304.  
    305.         logger.debug(">>>>>>>>>>> xxl-job, schedule push time-ring : " + ringSecond + " = " + Arrays.asList(ringItemData) );
    306.     }
    307.  
    308.     public void toStop(){
    309.  
    310.         // 1、stop schedule
    311.         scheduleThreadToStop = true;
    312.         try {
    313.             TimeUnit.SECONDS.sleep(1);  // wait
    314.         } catch (InterruptedException e) {
    315.             logger.error(e.getMessage(), e);
    316.         }
    317.         if (scheduleThread.getState() != Thread.State.TERMINATED){
    318.             // interrupt and wait
    319.             scheduleThread.interrupt();
    320.             try {
    321.                 scheduleThread.join();
    322.             } catch (InterruptedException e) {
    323.                 logger.error(e.getMessage(), e);
    324.             }
    325.         }
    326.  
    327.         // if has ring data
    328.         boolean hasRingData = false;
    329.         if (!ringData.isEmpty()) {
    330.             for (int second : ringData.keySet()) {
    331.                 List tmpData = ringData.get(second);
    332.                 if (tmpData!=null && tmpData.size()>0) {
    333.                     hasRingData = true;
    334.                     break;
    335.                 }
    336.             }
    337.         }
    338.         if (hasRingData) {
    339.             try {
    340.                 TimeUnit.SECONDS.sleep(8);
    341.             } catch (InterruptedException e) {
    342.                 logger.error(e.getMessage(), e);
    343.             }
    344.         }
    345.  
    346.         // stop ring (wait job-in-memory stop)
    347.         ringThreadToStop = true;
    348.         try {
    349.             TimeUnit.SECONDS.sleep(1);
    350.         } catch (InterruptedException e) {
    351.             logger.error(e.getMessage(), e);
    352.         }
    353.         if (ringThread.getState() != Thread.State.TERMINATED){
    354.             // interrupt and wait
    355.             ringThread.interrupt();
    356.             try {
    357.                 ringThread.join();
    358.             } catch (InterruptedException e) {
    359.                 logger.error(e.getMessage(), e);
    360.             }
    361.         }
    362.  
    363.         logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper stop");
    364.     }
    365. }

    3.3 任务执行时序图

    3.4 任务执行核心代码解析

    3.4.1 根据路由策略定位到执行器

    任务执行：XxlJobTrigger

    1. package com.xxl.job.admin.core.trigger;
    2.  
    3. import com.xxl.job.admin.core.conf.XxlJobAdminConfig;
    4. import com.xxl.job.admin.core.scheduler.XxlJobScheduler;
    5. import com.xxl.job.admin.core.model.XxlJobGroup;
    6. import com.xxl.job.admin.core.model.XxlJobInfo;
    7. import com.xxl.job.admin.core.model.XxlJobLog;
    8. import com.xxl.job.admin.core.route.ExecutorRouteStrategyEnum;
    9. import com.xxl.job.admin.core.util.I18nUtil;
    10. import com.xxl.job.core.biz.ExecutorBiz;
    11. import com.xxl.job.core.biz.model.ReturnT;
    12. import com.xxl.job.core.biz.model.TriggerParam;
    13. import com.xxl.job.core.enums.ExecutorBlockStrategyEnum;
    14. import com.xxl.rpc.util.IpUtil;
    15. import com.xxl.rpc.util.ThrowableUtil;
    16. import org.slf4j.Logger;
    17. import org.slf4j.LoggerFactory;
    18.  
    19. import java.util.Date;
    20.  
    21. /**
    22.  * xxl-job trigger
    23.  * Created by xuxueli on 17/7/13.
    24.  */
    25. public class XxlJobTrigger {
    26.     private static Logger logger = LoggerFactory.getLogger(XxlJobTrigger.class);
    27.  
    28.     /**
    29.      * trigger job
    30.      *
    31.      * @param jobId
    32.      * @param triggerType
    33.      * @param failRetryCount
    34.      * 			>=0: use this param
    35.      * 			<0: use param from job info config
    36.      * @param executorShardingParam
    37.      * @param executorParam
    38.      *          null: use job param
    39.      *          not null: cover job param
    40.      */
    41.     public static void trigger(int jobId, TriggerTypeEnum triggerType, int failRetryCount, String executorShardingParam, String executorParam) {
    42.         // load data
    43.         //通过jobId查询出任务信息
    44.         XxlJobInfo jobInfo = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().loadById(jobId);
    45.         if (jobInfo == null) {
    46.             logger.warn(">>>>>>>>>>>> trigger fail, jobId invalid，jobId={}", jobId);
    47.             return;
    48.         }
    49.         if (executorParam != null) {
    50.             jobInfo.setExecutorParam(executorParam);
    51.         }
    52.         //计算失败重试次数
    53.         int finalFailRetryCount = failRetryCount>=0?failRetryCount:jobInfo.getExecutorFailRetryCount();
    54.         XxlJobGroup group = XxlJobAdminConfig.getAdminConfig().getXxlJobGroupDao().load(jobInfo.getJobGroup());
    55.  
    56.         // sharding param
    57.         //处理分片参数
    58.         int[] shardingParam = null;
    59.         if (executorShardingParam!=null){
    60.             String[] shardingArr = executorShardingParam.split("/");
    61.             if (shardingArr.length==2 && isNumeric(shardingArr[0]) && isNumeric(shardingArr[1])) {
    62.                 shardingParam = new int[2];
    63.                 shardingParam[0] = Integer.valueOf(shardingArr[0]);
    64.                 shardingParam[1] = Integer.valueOf(shardingArr[1]);
    65.             }
    66.         }
    67.         //处理路由策略
    68.         if (ExecutorRouteStrategyEnum.SHARDING_BROADCAST==ExecutorRouteStrategyEnum.match(jobInfo.getExecutorRouteStrategy(), null)
    69.                 && group.getRegistryList()!=null && !group.getRegistryList().isEmpty()
    70.                 && shardingParam==null) {
    71.             //如果路由策略为分片广播，且执行器地址不为空，则遍历执行器地址进行广播触发任务调度
    72.             for (int i = 0; i < group.getRegistryList().size(); i++) {
    73.                 processTrigger(group, jobInfo, finalFailRetryCount, triggerType, i, group.getRegistryList().size());
    74.             }
    75.         } else {
    76.             //否则初始化分片参数
    77.             if (shardingParam == null) {
    78.                 shardingParam = new int[]{0, 1};
    79.             }
    80.             processTrigger(group, jobInfo, finalFailRetryCount, triggerType, shardingParam[0], shardingParam[1]);
    81.         }
    82.  
    83.     }
    84.  
    85.     private static boolean isNumeric(String str){
    86.         try {
    87.             int result = Integer.valueOf(str);
    88.             return true;
    89.         } catch (NumberFormatException e) {
    90.             return false;
    91.         }
    92.     }
    93.  
    94.     /**
    95.      * @param group                     job group, registry list may be empty
    96.      * @param jobInfo
    97.      * @param finalFailRetryCount
    98.      * @param triggerType
    99.      * @param index                     sharding index
    100.      * @param total                     sharding index
    101.      */
    102.     private static void processTrigger(XxlJobGroup group, XxlJobInfo jobInfo, int finalFailRetryCount, TriggerTypeEnum triggerType, int index, int total){
    103.  
    104.         // param
    105.         //执行器阻塞策略：调度过于密集执行器来不及处理时的处理策略，策略包括：单机串行（默认）、丢弃后续调度、覆盖之前调度；
    106.         ExecutorBlockStrategyEnum blockStrategy = ExecutorBlockStrategyEnum.match(jobInfo.getExecutorBlockStrategy(), ExecutorBlockStrategyEnum.SERIAL_EXECUTION);  // block strategy
    107.         //执行器路由策略：执行器集群部署时提供丰富的路由策略，包括：第一个、最后一个、轮询、随机、一致性HASH、最不经常使用、最近最久未使用、故障转移、忙碌转移等
    108.         ExecutorRouteStrategyEnum executorRouteStrategyEnum = ExecutorRouteStrategyEnum.match(jobInfo.getExecutorRouteStrategy(), null);    // route strategy
    109.         //处理分片参数：如果为分片广播则将index和total使用/进行拼接
    110.         String shardingParam = (ExecutorRouteStrategyEnum.SHARDING_BROADCAST==executorRouteStrategyEnum)?String.valueOf(index).concat("/").concat(String.valueOf(total)):null;
    111.  
    112.         // 1、save log-id
    113.         //保存任务执行日志
    114.         XxlJobLog jobLog = new XxlJobLog();
    115.         jobLog.setJobGroup(jobInfo.getJobGroup());
    116.         jobLog.setJobId(jobInfo.getId());
    117.         jobLog.setTriggerTime(new Date());
    118.         XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().save(jobLog);
    119.         logger.debug(">>>>>>>>>>> xxl-job trigger start, jobId:{}", jobLog.getId());
    120.  
    121.         // 2、init trigger-param
    122.         //初始化触发器参数
    123.         TriggerParam triggerParam = new TriggerParam();
    124.         triggerParam.setJobId(jobInfo.getId());
    125.         //取出执行器任务handler名称
    126.         triggerParam.setExecutorHandler(jobInfo.getExecutorHandler());
    127.         triggerParam.setExecutorParams(jobInfo.getExecutorParam());
    128.         triggerParam.setExecutorBlockStrategy(jobInfo.getExecutorBlockStrategy());
    129.         triggerParam.setExecutorTimeout(jobInfo.getExecutorTimeout());
    130.         triggerParam.setLogId(jobLog.getId());
    131.         triggerParam.setLogDateTime(jobLog.getTriggerTime().getTime());
    132.         triggerParam.setGlueType(jobInfo.getGlueType());
    133.         triggerParam.setGlueSource(jobInfo.getGlueSource());
    134.         triggerParam.setGlueUpdatetime(jobInfo.getGlueUpdatetime().getTime());
    135.         triggerParam.setBroadcastIndex(index);
    136.         triggerParam.setBroadcastTotal(total);
    137.  
    138.         // 3、init address
    139.         //初始化执行器地址信息
    140.         String address = null;
    141.         ReturnT routeAddressResult = null;
    142.         if (group.getRegistryList()!=null && !group.getRegistryList().isEmpty()) {
    143.             //如果执行器的注册地址不为空
    144.             if (ExecutorRouteStrategyEnum.SHARDING_BROADCAST == executorRouteStrategyEnum) {
    145.                 //如果路由策略是分片广播，取出index对应的执行器地址信息
    146.                 if (index < group.getRegistryList().size()) {
    147.                     address = group.getRegistryList().get(index);
    148.                 } else {
    149.                     address = group.getRegistryList().get(0);
    150.                 }
    151.             } else {
    152.                 //获取执行器路由地址信息
    153.                 //⚠️⚠️️⚠️非常重要：此处使用了策略模式， 根据不同的策略 使用不同的实现类，从而选举出本次使用执行器集群地址列表中不同的地址信息，这是xxl-job实现任务路由的地方
    154.                 routeAddressResult = executorRouteStrategyEnum.getRouter().route(triggerParam, group.getRegistryList());
    155.                 if (routeAddressResult.getCode() == ReturnT.SUCCESS_CODE) {
    156.                     address = routeAddressResult.getContent();
    157.                 }
    158.             }
    159.         } else {
    160.             routeAddressResult = new ReturnT(ReturnT.FAIL_CODE, I18nUtil.getString("jobconf_trigger_address_empty"));
    161.         }
    162.  
    163.         // 4、trigger remote executor
    164.         //触发远程执行器执行任务（执行器中的handler）
    165.         ReturnT triggerResult = null;
    166.         if (address != null) {
    167.             //这里是重点⚠️⚠️⚠️：启动执行器， 向执行器发送指令都是从这个方法中执行的
    168.             triggerResult = runExecutor(triggerParam, address);
    169.         } else {
    170.             triggerResult = new ReturnT(ReturnT.FAIL_CODE, null);
    171.         }
    172.  
    173.         // 5、collection trigger info
    174.         StringBuffer triggerMsgSb = new StringBuffer();
    175.         triggerMsgSb.append(I18nUtil.getString("jobconf_trigger_type")).append("：").append(triggerType.getTitle());
    176.         triggerMsgSb.append("
      "      ).append(I18nUtil.getString("jobconf_trigger_admin_adress")).append("：").append(IpUtil.getIp());
    177.         triggerMsgSb.append("
      "      ).append(I18nUtil.getString("jobconf_trigger_exe_regtype")).append("：")
    178.                 .append( (group.getAddressType() == 0)?I18nUtil.getString("jobgroup_field_addressType_0"):I18nUtil.getString("jobgroup_field_addressType_1") );
    179.         triggerMsgSb.append("
      "      ).append(I18nUtil.getString("jobconf_trigger_exe_regaddress")).append("：").append(group.getRegistryList());
    180.         triggerMsgSb.append("
      "      ).append(I18nUtil.getString("jobinfo_field_executorRouteStrategy")).append("：").append(executorRouteStrategyEnum.getTitle());
    181.         if (shardingParam != null) {
    182.             triggerMsgSb.append("("+shardingParam+")");
    183.         }
    184.         triggerMsgSb.append("
      "      ).append(I18nUtil.getString("jobinfo_field_executorBlockStrategy")).append("：").append(blockStrategy.getTitle());
    185.         triggerMsgSb.append("
      "      ).append(I18nUtil.getString("jobinfo_field_timeout")).append("：").append(jobInfo.getExecutorTimeout());
    186.         triggerMsgSb.append("
      "      ).append(I18nUtil.getString("jobinfo_field_executorFailRetryCount")).append("：").append(finalFailRetryCount);
    187.  
    188.         triggerMsgSb.append("

       >>>>>>>>>>>"+ I18nUtil.getString("jobconf_trigger_run") +"<<<<<<<<<<< 
      "      )
    189.                 .append((routeAddressResult!=null&&routeAddressResult.getMsg()!=null)?routeAddressResult.getMsg()+"

      "      :"").append(triggerResult.getMsg()!=null?triggerResult.getMsg():"");
    190.  
    191.         // 6、save log trigger-info
    192.         //更新执行日志
    193.         jobLog.setExecutorAddress(address);
    194.         jobLog.setExecutorHandler(jobInfo.getExecutorHandler());
    195.         jobLog.setExecutorParam(jobInfo.getExecutorParam());
    196.         jobLog.setExecutorShardingParam(shardingParam);
    197.         jobLog.setExecutorFailRetryCount(finalFailRetryCount);
    198.         //jobLog.setTriggerTime();
    199.         jobLog.setTriggerCode(triggerResult.getCode());
    200.         jobLog.setTriggerMsg(triggerMsgSb.toString());
    201.         XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().updateTriggerInfo(jobLog);
    202.  
    203.         logger.debug(">>>>>>>>>>> xxl-job trigger end, jobId:{}", jobLog.getId());
    204.     }
    205.  
    206.     /**
    207.      * run executor
    208.      * @param triggerParam
    209.      * @param address
    210.      * @return
    211.      */
    212.     public static ReturnT runExecutor(TriggerParam triggerParam, String address){
    213.         ReturnT runResult = null;
    214.         try {
    215.             //这是重点：获取ExecutorBiz代理对象，先从缓存（内存map）中取，取不到new一个XxlRpcReferenceBean放进缓存map
    216.             ExecutorBiz executorBiz = XxlJobScheduler.getExecutorBiz(address);
    217.             // 这个run方法不会最终执行，仅仅只是为了触发代理对象的invoke方法，同时将目标的类型传送给服务端，因为在代理对象的invoke的方法里面没有执行目标对象的方法
    218.             runResult = executorBiz.run(triggerParam);
    219.         } catch (Exception e) {
    220.             logger.error(">>>>>>>>>>> xxl-job trigger error, please check if the executor[{}] is running.", address, e);
    221.             runResult = new ReturnT(ReturnT.FAIL_CODE, ThrowableUtil.toString(e));
    222.         }
    223.  
    224.         StringBuffer runResultSB = new StringBuffer(I18nUtil.getString("jobconf_trigger_run") + "：");
    225.         runResultSB.append("
      address："      ).append(address);
    226.         runResultSB.append("
      code："      ).append(runResult.getCode());
    227.         runResultSB.append("
      msg："      ).append(runResult.getMsg());
    228.  
    229.         runResult.setMsg(runResultSB.toString());
    230.         return runResult;
    231.     }
    232.  
    233. }

    3.4.2 定位到任务执行Handler&线程

    获得执行Handler&线程：ExecutorBizImpl

    1. package com.xxl.job.core.biz.impl;
    2.  
    3. import com.xxl.job.core.biz.ExecutorBiz;
    4. import com.xxl.job.core.biz.model.LogResult;
    5. import com.xxl.job.core.biz.model.ReturnT;
    6. import com.xxl.job.core.biz.model.TriggerParam;
    7. import com.xxl.job.core.enums.ExecutorBlockStrategyEnum;
    8. import com.xxl.job.core.executor.XxlJobExecutor;
    9. import com.xxl.job.core.glue.GlueFactory;
    10. import com.xxl.job.core.glue.GlueTypeEnum;
    11. import com.xxl.job.core.handler.IJobHandler;
    12. import com.xxl.job.core.handler.impl.GlueJobHandler;
    13. import com.xxl.job.core.handler.impl.ScriptJobHandler;
    14. import com.xxl.job.core.log.XxlJobFileAppender;
    15. import com.xxl.job.core.thread.JobThread;
    16. import org.slf4j.Logger;
    17. import org.slf4j.LoggerFactory;
    18.  
    19. import java.util.Date;
    20.  
    21. /**
    22.  * Created by xuxueli on 17/3/1.
    23.  * @Description：阻塞处理策略，当执行器节点存在多个相同任务id的任务未执行完成，则需要基于阻塞策略对任务进行取舍： 串行策略：默认策略，任务进行排队、丢弃旧任务策略、丢弃新任务策略
    24.  */
    25. public class ExecutorBizImpl implements ExecutorBiz {
    26.     private static Logger logger = LoggerFactory.getLogger(ExecutorBizImpl.class);
    27.  
    28.     @Override
    29.     public ReturnT beat() {
    30.         return ReturnT.SUCCESS;
    31.     }
    32.  
    33.     @Override
    34.     public ReturnT idleBeat(int jobId) {
    35.  
    36.         // isRunningOrHasQueue
    37.         boolean isRunningOrHasQueue = false;
    38.         JobThread jobThread = XxlJobExecutor.loadJobThread(jobId);
    39.         if (jobThread != null && jobThread.isRunningOrHasQueue()) {
    40.             isRunningOrHasQueue = true;
    41.         }
    42.  
    43.         if (isRunningOrHasQueue) {
    44.             return new ReturnT(ReturnT.FAIL_CODE, "job thread is running or has trigger queue.");
    45.         }
    46.         return ReturnT.SUCCESS;
    47.     }
    48.  
    49.     @Override
    50.     public ReturnT kill(int jobId) {
    51.         // kill handlerThread, and create new one
    52.         JobThread jobThread = XxlJobExecutor.loadJobThread(jobId);
    53.         if (jobThread != null) {
    54.             XxlJobExecutor.removeJobThread(jobId, "scheduling center kill job.");
    55.             return ReturnT.SUCCESS;
    56.         }
    57.  
    58.         return new ReturnT(ReturnT.SUCCESS_CODE, "job thread already killed.");
    59.     }
    60.  
    61.     @Override
    62.     public ReturnT log(long logDateTim, long logId, int fromLineNum) {
    63.         // log filename: logPath/yyyy-MM-dd/9999.log
    64.         String logFileName = XxlJobFileAppender.makeLogFileName(new Date(logDateTim), logId);
    65.  
    66.         LogResult logResult = XxlJobFileAppender.readLog(logFileName, fromLineNum);
    67.         return new ReturnT(logResult);
    68.     }
    69.  
    70.     @Override
    71.     public ReturnT run(TriggerParam triggerParam) {
    72.         // load old：jobHandler + jobThread
    73.         JobThread jobThread = XxlJobExecutor.loadJobThread(triggerParam.getJobId());
    74.         IJobHandler jobHandler = jobThread!=null?jobThread.getHandler():null;
    75.         String removeOldReason = null;
    76.  
    77.         // valid：jobHandler + jobThread
    78.         GlueTypeEnum glueTypeEnum = GlueTypeEnum.match(triggerParam.getGlueType());
    79.         if (GlueTypeEnum.BEAN == glueTypeEnum) {
    80.  
    81.             // new jobhandler
    82.             IJobHandler newJobHandler = XxlJobExecutor.loadJobHandler(triggerParam.getExecutorHandler());
    83.  
    84.             // valid old jobThread
    85.             if (jobThread!=null && jobHandler != newJobHandler) {
    86.                 // change handler, need kill old thread
    87.                 removeOldReason = "change jobhandler or glue type, and terminate the old job thread.";
    88.  
    89.                 jobThread = null;
    90.                 jobHandler = null;
    91.             }
    92.  
    93.             // valid handler
    94.             if (jobHandler == null) {
    95.                 jobHandler = newJobHandler;
    96.                 if (jobHandler == null) {
    97.                     return new ReturnT(ReturnT.FAIL_CODE, "job handler [" + triggerParam.getExecutorHandler() + "] not found.");
    98.                 }
    99.             }
    100.  
    101.         } else if (GlueTypeEnum.GLUE_GROOVY == glueTypeEnum) {
    102.  
    103.             // valid old jobThread
    104.             if (jobThread != null &&
    105.                     !(jobThread.getHandler() instanceof GlueJobHandler
    106.                         && ((GlueJobHandler) jobThread.getHandler()).getGlueUpdatetime()==triggerParam.getGlueUpdatetime() )) {
    107.                 // change handler or gluesource updated, need kill old thread
    108.                 removeOldReason = "change job source or glue type, and terminate the old job thread.";
    109.  
    110.                 jobThread = null;
    111.                 jobHandler = null;
    112.             }
    113.  
    114.             // valid handler
    115.             if (jobHandler == null) {
    116.                 try {
    117.                     IJobHandler originJobHandler = GlueFactory.getInstance().loadNewInstance(triggerParam.getGlueSource());
    118.                     jobHandler = new GlueJobHandler(originJobHandler, triggerParam.getGlueUpdatetime());
    119.                 } catch (Exception e) {
    120.                     logger.error(e.getMessage(), e);
    121.                     return new ReturnT(ReturnT.FAIL_CODE, e.getMessage());
    122.                 }
    123.             }
    124.         } else if (glueTypeEnum!=null && glueTypeEnum.isScript()) {
    125.  
    126.             // valid old jobThread
    127.             if (jobThread != null &&
    128.                     !(jobThread.getHandler() instanceof ScriptJobHandler
    129.                             && ((ScriptJobHandler) jobThread.getHandler()).getGlueUpdatetime()==triggerParam.getGlueUpdatetime() )) {
    130.                 // change script or gluesource updated, need kill old thread
    131.                 removeOldReason = "change job source or glue type, and terminate the old job thread.";
    132.  
    133.                 jobThread = null;
    134.                 jobHandler = null;
    135.             }
    136.  
    137.             // valid handler
    138.             if (jobHandler == null) {
    139.                 jobHandler = new ScriptJobHandler(triggerParam.getJobId(), triggerParam.getGlueUpdatetime(), triggerParam.getGlueSource(), GlueTypeEnum.match(triggerParam.getGlueType()));
    140.             }
    141.         } else {
    142.             return new ReturnT(ReturnT.FAIL_CODE, "glueType[" + triggerParam.getGlueType() + "] is not valid.");
    143.         }
    144.  
    145.         // executor block strategy
    146.         if (jobThread != null) {
    147.             ExecutorBlockStrategyEnum blockStrategy = ExecutorBlockStrategyEnum.match(triggerParam.getExecutorBlockStrategy(), null);
    148.             if (ExecutorBlockStrategyEnum.DISCARD_LATER == blockStrategy) {
    149.                 // discard when running
    150.                 if (jobThread.isRunningOrHasQueue()) {
    151.                     return new ReturnT(ReturnT.FAIL_CODE, "block strategy effect："+ExecutorBlockStrategyEnum.DISCARD_LATER.getTitle());
    152.                 }
    153.             } else if (ExecutorBlockStrategyEnum.COVER_EARLY == blockStrategy) {
    154.                 // kill running jobThread
    155.                 if (jobThread.isRunningOrHasQueue()) {
    156.                     removeOldReason = "block strategy effect：" + ExecutorBlockStrategyEnum.COVER_EARLY.getTitle();
    157.  
    158.                     jobThread = null;
    159.                 }
    160.             } else {
    161.                 // just queue trigger
    162.             }
    163.         }
    164.  
    165.         // replace thread (new or exists invalid)
    166.         if (jobThread == null) {
    167.             // 任务执行
    168.             jobThread = XxlJobExecutor.registJobThread(triggerParam.getJobId(), jobHandler, removeOldReason);
    169.         }
    170.  
    171.         // push data to queue
    172.         ReturnT pushResult = jobThread.pushTriggerQueue(triggerParam);
    173.         return pushResult;
    174.     }
    175.  
    176. }

    3.4.3 任务执行线程

    任务线程执行&放入回调队列：JobThread

    1. package com.xxl.job.core.thread;
    2.  
    3. import com.xxl.job.core.biz.model.HandleCallbackParam;
    4. import com.xxl.job.core.biz.model.ReturnT;
    5. import com.xxl.job.core.biz.model.TriggerParam;
    6. import com.xxl.job.core.executor.XxlJobExecutor;
    7. import com.xxl.job.core.handler.IJobHandler;
    8. import com.xxl.job.core.log.XxlJobFileAppender;
    9. import com.xxl.job.core.log.XxlJobLogger;
    10. import com.xxl.job.core.util.ShardingUtil;
    11. import org.slf4j.Logger;
    12. import org.slf4j.LoggerFactory;
    13.  
    14. import java.io.PrintWriter;
    15. import java.io.StringWriter;
    16. import java.util.Collections;
    17. import java.util.Date;
    18. import java.util.HashSet;
    19. import java.util.Set;
    20. import java.util.concurrent.*;
    21.  
    22.  
    23. /**
    24.  * handler thread
    25.  * @author xuxueli 2016-1-16 19:52:47
    26.  */
    27. public class JobThread extends Thread{
    28. 	private static Logger logger = LoggerFactory.getLogger(JobThread.class);
    29.  
    30. 	private int jobId;
    31. 	private IJobHandler handler;
    32. 	private LinkedBlockingQueue triggerQueue;
    33. 	private Set triggerLogIdSet;		// avoid repeat trigger for the same TRIGGER_LOG_ID
    34.  
    35. 	private volatile boolean toStop = false;
    36. 	private String stopReason;
    37.  
    38.     private boolean running = false;    // if running job
    39. 	private int idleTimes = 0;			// idel times
    40.  
    41.  
    42. 	public JobThread(int jobId, IJobHandler handler) {
    43. 		this.jobId = jobId;
    44. 		this.handler = handler;
    45. 		this.triggerQueue = new LinkedBlockingQueue();
    46. 		this.triggerLogIdSet = Collections.synchronizedSet(new HashSet());
    47. 	}
    48. 	public IJobHandler getHandler() {
    49. 		return handler;
    50. 	}
    51.  
    52.     /**
    53.      * new trigger to queue
    54.      *
    55.      * @param triggerParam
    56.      * @return
    57.      */
    58. 	public ReturnT pushTriggerQueue(TriggerParam triggerParam) {
    59. 		// avoid repeat
    60. 		if (triggerLogIdSet.contains(triggerParam.getLogId())) {
    61. 			logger.info(">>>>>>>>>>> repeate trigger job, logId:{}", triggerParam.getLogId());
    62. 			return new ReturnT(ReturnT.FAIL_CODE, "repeate trigger job, logId:" + triggerParam.getLogId());
    63. 		}
    64.  
    65. 		triggerLogIdSet.add(triggerParam.getLogId());
    66. 		triggerQueue.add(triggerParam);
    67.         return ReturnT.SUCCESS;
    68. 	}
    69.  
    70.     /**
    71.      * kill job thread
    72.      *
    73.      * @param stopReason
    74.      */
    75. 	public void toStop(String stopReason) {
    76. 		/**
    77. 		 * Thread.interrupt只支持终止线程的阻塞状态(wait、join、sleep)，
    78. 		 * 在阻塞出抛出InterruptedException异常,但是并不会终止运行的线程本身；
    79. 		 * 所以需要注意，此处彻底销毁本线程，需要通过共享变量方式；
    80. 		 */
    81. 		this.toStop = true;
    82. 		this.stopReason = stopReason;
    83. 	}
    84.  
    85.     /**
    86.      * is running job
    87.      * @return
    88.      */
    89.     public boolean isRunningOrHasQueue() {
    90.         return running || triggerQueue.size()>0;
    91.     }
    92.  
    93.     @Override
    94. 	public void run() {
    95.  
    96.     	// init
    97.     	try {
    98. 			handler.init();
    99. 		} catch (Throwable e) {
    100.     		logger.error(e.getMessage(), e);
    101. 		}
    102.  
    103. 		// execute
    104. 		while(!toStop){
    105. 			running = false;
    106. 			idleTimes++;
    107.  
    108.             TriggerParam triggerParam = null;
    109.             ReturnT executeResult = null;
    110.             try {
    111. 				// to check toStop signal, we need cycle, so wo cannot use queue.take(), instand of poll(timeout)
    112. 				triggerParam = triggerQueue.poll(3L, TimeUnit.SECONDS);
    113. 				if (triggerParam!=null) {
    114. 					running = true;
    115. 					idleTimes = 0;
    116. 					triggerLogIdSet.remove(triggerParam.getLogId());
    117.  
    118. 					// log filename, like "logPath/yyyy-MM-dd/9999.log"
    119. 					String logFileName = XxlJobFileAppender.makeLogFileName(new Date(triggerParam.getLogDateTime()), triggerParam.getLogId());
    120. 					XxlJobFileAppender.contextHolder.set(logFileName);
    121. 					ShardingUtil.setShardingVo(new ShardingUtil.ShardingVO(triggerParam.getBroadcastIndex(), triggerParam.getBroadcastTotal()));
    122.  
    123. 					// execute
    124. 					XxlJobLogger.log("
      ----------- xxl-job job execute start -----------
      ----------- Param:"       + triggerParam.getExecutorParams());
    125.  
    126. 					if (triggerParam.getExecutorTimeout() > 0) {
    127. 						// limit timeout
    128. 						Thread futureThread = null;
    129. 						try {
    130. 							final TriggerParam triggerParamTmp = triggerParam;
    131. 							FutureTask> futureTask = new FutureTask>(new Callable>() {
    132. 								@Override
    133. 								public ReturnT call() throws Exception {
    134. 									return handler.execute(triggerParamTmp.getExecutorParams());
    135. 								}
    136. 							});
    137. 							futureThread = new Thread(futureTask);
    138. 							futureThread.start();
    139.  
    140. 							executeResult = futureTask.get(triggerParam.getExecutorTimeout(), TimeUnit.SECONDS);
    141. 						} catch (TimeoutException e) {
    142.  
    143. 							XxlJobLogger.log("
      ----------- xxl-job job execute timeout"      );
    144. 							XxlJobLogger.log(e);
    145.  
    146. 							executeResult = new ReturnT(IJobHandler.FAIL_TIMEOUT.getCode(), "job execute timeout ");
    147. 						} finally {
    148. 							futureThread.interrupt();
    149. 						}
    150. 					} else {
    151. 						// just execute
    152. 						executeResult = handler.execute(triggerParam.getExecutorParams());
    153. 					}
    154.  
    155. 					if (executeResult == null) {
    156. 						executeResult = IJobHandler.FAIL;
    157. 					} else {
    158. 						executeResult.setMsg(
    159. 								(executeResult!=null&&executeResult.getMsg()!=null&&executeResult.getMsg().length()>50000)
    160. 										?executeResult.getMsg().substring(0, 50000).concat("...")
    161. 										:executeResult.getMsg());
    162. 						executeResult.setContent(null);	// limit obj size
    163. 					}
    164. 					XxlJobLogger.log("
      ----------- xxl-job job execute end(finish) -----------
      ----------- ReturnT:"       + executeResult);
    165.  
    166. 				} else {
    167. 					if (idleTimes > 30) {
    168. 						if(triggerQueue.size() == 0) {	// avoid concurrent trigger causes jobId-lost
    169. 							XxlJobExecutor.removeJobThread(jobId, "excutor idel times over limit.");
    170. 						}
    171. 					}
    172. 				}
    173. 			} catch (Throwable e) {
    174. 				if (toStop) {
    175. 					XxlJobLogger.log("
      ----------- JobThread toStop, stopReason:"       + stopReason);
    176. 				}
    177.  
    178. 				StringWriter stringWriter = new StringWriter();
    179. 				e.printStackTrace(new PrintWriter(stringWriter));
    180. 				String errorMsg = stringWriter.toString();
    181. 				executeResult = new ReturnT(ReturnT.FAIL_CODE, errorMsg);
    182.  
    183. 				XxlJobLogger.log("
      ----------- JobThread Exception:"       + errorMsg + "
      ----------- xxl-job job execute end(error) -----------"      );
    184. 			} finally {
    185.                 if(triggerParam != null) {
    186.                     // callback handler info
    187.                     if (!toStop) {
    188.                         // commonm
    189.                         TriggerCallbackThread.pushCallBack(new HandleCallbackParam(triggerParam.getLogId(), triggerParam.getLogDateTime(), executeResult));
    190.                     } else {
    191.                         // is killed
    192.                         ReturnT stopResult = new ReturnT(ReturnT.FAIL_CODE, stopReason + " [job running, killed]");
    193.                         TriggerCallbackThread.pushCallBack(new HandleCallbackParam(triggerParam.getLogId(), triggerParam.getLogDateTime(), stopResult));
    194.                     }
    195.                 }
    196.             }
    197.         }
    198.  
    199. 		// callback trigger request in queue
    200. 		while(triggerQueue !=null && triggerQueue.size()>0){
    201. 			TriggerParam triggerParam = triggerQueue.poll();
    202. 			if (triggerParam!=null) {
    203. 				// is killed
    204. 				ReturnT stopResult = new ReturnT(ReturnT.FAIL_CODE, stopReason + " [job not executed, in the job queue, killed.]");
    205. 				TriggerCallbackThread.pushCallBack(new HandleCallbackParam(triggerParam.getLogId(), triggerParam.getLogDateTime(), stopResult));
    206. 			}
    207. 		}
    208.  
    209. 		// destroy
    210. 		try {
    211. 			handler.destroy();
    212. 		} catch (Throwable e) {
    213. 			logger.error(e.getMessage(), e);
    214. 		}
    215.  
    216. 		logger.info(">>>>>>>>>>> xxl-job JobThread stoped, hashCode:{}", Thread.currentThread());
    217. 	}
    218. }

    四、执行器

    4.1 启动过程时序图

    4.2 启动过程核心代码解析

    入口类：XxlJobSpringExecutor（PS：继承XxlJobExecutor，所有的任务触发最终都是通过这个类）

    4.2.1 启动初始化

    执行器启动初始化：XxlJobSpringExecutor

    1. package com.xxl.job.core.executor.impl;
    2.  
    3. import com.xxl.job.core.biz.model.ReturnT;
    4. import com.xxl.job.core.executor.XxlJobExecutor;
    5. import com.xxl.job.core.glue.GlueFactory;
    6. import com.xxl.job.core.handler.IJobHandler;
    7. import com.xxl.job.core.handler.annotation.JobHandler;
    8. import com.xxl.job.core.handler.annotation.XxlJob;
    9. import com.xxl.job.core.handler.impl.MethodJobHandler;
    10. import org.springframework.beans.BeansException;
    11. import org.springframework.beans.factory.DisposableBean;
    12. import org.springframework.beans.factory.InitializingBean;
    13. import org.springframework.context.ApplicationContext;
    14. import org.springframework.context.ApplicationContextAware;
    15. import org.springframework.core.annotation.AnnotationUtils;
    16.  
    17. import java.lang.reflect.Method;
    18. import java.util.Map;
    19.  
    20. /**
    21.  * xxl-job executor (for spring)
    22.  *
    23.  * @author xuxueli 2018-11-01 09:24:52
    24.  * @Description：执行器启动过程
    25.  */
    26. public class XxlJobSpringExecutor extends XxlJobExecutor implements ApplicationContextAware, InitializingBean, DisposableBean {
    27.  
    28.  
    29.     // start
    30.     @Override
    31.     public void afterPropertiesSet() throws Exception {
    32.  
    33.         // init JobHandler Repository
    34.         /**
    35.          *  老版本：获取使用了JobHandler注解的bean，即任务处理器，并注册到jobHandlerRepository（ConcurrentMap）缓存中
    36.          */
    37.         initJobHandlerRepository(applicationContext);
    38.  
    39.         // init JobHandler Repository (for method)
    40.         /**
    41.          *  新版本：获取使用了XxlJob注解的method，即任务处理器，并注册到jobHandlerRepository（ConcurrentMap）缓存中
    42.          */
    43.         initJobHandlerMethodRepository(applicationContext);
    44.  
    45.         // refresh GlueFactory
    46.         GlueFactory.refreshInstance(1);
    47.  
    48.         // super start
    49.         /**
    50.          * 核心启动
    51.          */
    52.         super.start();
    53.     }
    54.  
    55.     // destroy
    56.     @Override
    57.     public void destroy() {
    58.         super.destroy();
    59.     }
    60.  
    61.  
    62.     private void initJobHandlerRepository(ApplicationContext applicationContext) {
    63.         if (applicationContext == null) {
    64.             return;
    65.         }
    66.  
    67.         // init job handler action
    68.         Map serviceBeanMap = applicationContext.getBeansWithAnnotation(JobHandler.class);
    69.  
    70.         if (serviceBeanMap != null && serviceBeanMap.size() > 0) {
    71.             for (Object serviceBean : serviceBeanMap.values()) {
    72.                 if (serviceBean instanceof IJobHandler) {
    73.                     String name = serviceBean.getClass().getAnnotation(JobHandler.class).value();
    74.                     IJobHandler handler = (IJobHandler) serviceBean;
    75.                     if (loadJobHandler(name) != null) {
    76.                         throw new RuntimeException("xxl-job jobhandler[" + name + "] naming conflicts.");
    77.                     }
    78.                     registJobHandler(name, handler);
    79.                 }
    80.             }
    81.         }
    82.     }
    83.  
    84.     private void initJobHandlerMethodRepository(ApplicationContext applicationContext) {
    85.         if (applicationContext == null) {
    86.             return;
    87.         }
    88.  
    89.         // init job handler from method
    90.         String[] beanDefinitionNames = applicationContext.getBeanDefinitionNames();
    91.         if (beanDefinitionNames!=null && beanDefinitionNames.length>0) {
    92.             for (String beanDefinitionName : beanDefinitionNames) {
    93.                 Object bean = applicationContext.getBean(beanDefinitionName);
    94.                 Method[] methods = bean.getClass().getDeclaredMethods();
    95.                 for (Method method: methods) {
    96.                     XxlJob xxlJob = AnnotationUtils.findAnnotation(method, XxlJob.class);
    97.                     if (xxlJob != null) {
    98.  
    99.                         // name
    100.                         String name = xxlJob.value();
    101.                         if (name.trim().length() == 0) {
    102.                             throw new RuntimeException("xxl-job method-jobhandler name invalid, for[" + bean.getClass() + "#"+ method.getName() +"] .");
    103.                         }
    104.                         if (loadJobHandler(name) != null) {
    105.                             throw new RuntimeException("xxl-job jobhandler[" + name + "] naming conflicts.");
    106.                         }
    107.  
    108.                         // execute method
    109.                         if (!(method.getParameterTypes()!=null && method.getParameterTypes().length==1 && method.getParameterTypes()[0].isAssignableFrom(String.class))) {
    110.                             throw new RuntimeException("xxl-job method-jobhandler param-classtype invalid, for[" + bean.getClass() + "#"+ method.getName() +"] , " +
    111.                                     "The correct method format like \" public ReturnT execute(String param) \" .");
    112.                         }
    113.                         if (!method.getReturnType().isAssignableFrom(ReturnT.class)) {
    114.                             throw new RuntimeException("xxl-job method-jobhandler return-classtype invalid, for[" + bean.getClass() + "#"+ method.getName() +"] , " +
    115.                                     "The correct method format like \" public ReturnT execute(String param) \" .");
    116.                         }
    117.                         method.setAccessible(true);
    118.  
    119.                         // init and destory
    120.                         Method initMethod = null;
    121.                         Method destroyMethod = null;
    122.  
    123.                         if(xxlJob.init().trim().length() > 0) {
    124.                             try {
    125.                                 initMethod = bean.getClass().getDeclaredMethod(xxlJob.init());
    126.                                 initMethod.setAccessible(true);
    127.                             } catch (NoSuchMethodException e) {
    128.                                 throw new RuntimeException("xxl-job method-jobhandler initMethod invalid, for[" + bean.getClass() + "#"+ method.getName() +"] .");
    129.                             }
    130.                         }
    131.                         if(xxlJob.destroy().trim().length() > 0) {
    132.                             try {
    133.                                 destroyMethod = bean.getClass().getDeclaredMethod(xxlJob.destroy());
    134.                                 destroyMethod.setAccessible(true);
    135.                             } catch (NoSuchMethodException e) {
    136.                                 throw new RuntimeException("xxl-job method-jobhandler destroyMethod invalid, for[" + bean.getClass() + "#"+ method.getName() +"] .");
    137.                             }
    138.                         }
    139.  
    140.                         // registry jobhandler
    141.                         registJobHandler(name, new MethodJobHandler(bean, method, initMethod, destroyMethod));
    142.                     }
    143.                 }
    144.             }
    145.         }
    146.  
    147.     }
    148.  
    149.     // ---------------------- applicationContext ----------------------
    150.     private static ApplicationContext applicationContext;
    151.  
    152.     @Override
    153.     public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
    154.         this.applicationContext = applicationContext;
    155.     }
    156.  
    157.     public static ApplicationContext getApplicationContext() {
    158.         return applicationContext;
    159.     }
    160.  
    161. }

    4.2.2 执行器初始化

    执行器相关信息准备：XxlJobExecutor

    1. package com.xxl.job.core.executor;
    2.  
    3. import com.xxl.job.core.biz.AdminBiz;
    4. import com.xxl.job.core.biz.ExecutorBiz;
    5. import com.xxl.job.core.biz.client.AdminBizClient;
    6. import com.xxl.job.core.biz.impl.ExecutorBizImpl;
    7. import com.xxl.job.core.handler.IJobHandler;
    8. import com.xxl.job.core.log.XxlJobFileAppender;
    9. import com.xxl.job.core.thread.ExecutorRegistryThread;
    10. import com.xxl.job.core.thread.JobLogFileCleanThread;
    11. import com.xxl.job.core.thread.JobThread;
    12. import com.xxl.job.core.thread.TriggerCallbackThread;
    13. import com.xxl.rpc.registry.ServiceRegistry;
    14. import com.xxl.rpc.remoting.net.impl.netty_http.server.NettyHttpServer;
    15. import com.xxl.rpc.remoting.provider.XxlRpcProviderFactory;
    16. import com.xxl.rpc.serialize.Serializer;
    17. import com.xxl.rpc.serialize.impl.HessianSerializer;
    18. import com.xxl.rpc.util.IpUtil;
    19. import com.xxl.rpc.util.NetUtil;
    20. import org.slf4j.Logger;
    21. import org.slf4j.LoggerFactory;
    22.  
    23. import java.util.*;
    24. import java.util.concurrent.ConcurrentHashMap;
    25. import java.util.concurrent.ConcurrentMap;
    26.  
    27. /**
    28.  * Created by xuxueli on 2016/3/2 21:14.
    29.  * @Description：任务执行器
    30.  */
    31. public class XxlJobExecutor  {
    32.     private static final Logger logger = LoggerFactory.getLogger(XxlJobExecutor.class);
    33.  
    34.     // ---------------------- param ----------------------
    35.     private String adminAddresses;
    36.     private String appName;
    37.     private String ip;
    38.     private int port;
    39.     private String accessToken;
    40.     private String logPath;
    41.     private int logRetentionDays;
    42.  
    43.     public void setAdminAddresses(String adminAddresses) {
    44.         this.adminAddresses = adminAddresses;
    45.     }
    46.     public void setAppName(String appName) {
    47.         this.appName = appName;
    48.     }
    49.     public void setIp(String ip) {
    50.         this.ip = ip;
    51.     }
    52.     public void setPort(int port) {
    53.         this.port = port;
    54.     }
    55.     public void setAccessToken(String accessToken) {
    56.         this.accessToken = accessToken;
    57.     }
    58.     public void setLogPath(String logPath) {
    59.         this.logPath = logPath;
    60.     }
    61.     public void setLogRetentionDays(int logRetentionDays) {
    62.         this.logRetentionDays = logRetentionDays;
    63.     }
    64.  
    65.  
    66.     // ---------------------- start + stop ----------------------
    67.     public void start() throws Exception {
    68.  
    69.         // init logpath
    70.         /**
    71.          * 主要作用：初始化日志路径
    72.          */
    73.         XxlJobFileAppender.initLogPath(logPath);
    74.  
    75.         // init invoker, admin-client
    76.         /**
    77.          * 主要作用：初始化注册中心列表 （把注册地址放到 List）
    78.          * 1、初始化执行器与调度中心建立通信的地址
    79.          */
    80.         initAdminBizList(adminAddresses, accessToken);
    81.  
    82.  
    83.         // init JobLogFileCleanThread
    84.         /**
    85.          * 主要作用：
    86.          * 1、启动日志文件清理线程 （一天清理一次）
    87.          * 2、每天清理一次过期日志，配置参数必须大于3才有效
    88.          */
    89.         JobLogFileCleanThread.getInstance().start(logRetentionDays);
    90.  
    91.         // init TriggerCallbackThread
    92.         /**
    93.          * 主要作用：开启触发器回调线程
    94.          */
    95.         TriggerCallbackThread.getInstance().start();
    96.  
    97.         // init executor-server
    98.         /**
    99.          * 主要端口：指定端口
    100.          */
    101.         port = port>0?port: NetUtil.findAvailablePort(9999);
    102.         /**
    103.          * 主要端口：指定IP
    104.          */
    105.         ip = (ip!=null&&ip.trim().length()>0)?ip: IpUtil.getIp();
    106.         /**
    107.          * 主要作用：启动注册中心的RPC服务，使得执行器项目可以通过RPC进行注册和心跳检测
    108.          * 1、将执行器注册到调度中心 30秒一次
    109.          */
    110.         initRpcProvider(ip, port, appName, accessToken);
    111.     }
    112.     public void destroy(){
    113.         // destory executor-server
    114.         stopRpcProvider();
    115.  
    116.         // destory jobThreadRepository
    117.         if (jobThreadRepository.size() > 0) {
    118.             for (Map.Entry item: jobThreadRepository.entrySet()) {
    119.                 removeJobThread(item.getKey(), "web container destroy and kill the job.");
    120.             }
    121.             jobThreadRepository.clear();
    122.         }
    123.         jobHandlerRepository.clear();
    124.  
    125.  
    126.         // destory JobLogFileCleanThread
    127.         JobLogFileCleanThread.getInstance().toStop();
    128.  
    129.         // destory TriggerCallbackThread
    130.         TriggerCallbackThread.getInstance().toStop();
    131.  
    132.     }
    133.  
    134.  
    135.     // ---------------------- admin-client (rpc invoker) ----------------------
    136.     private static List adminBizList;
    137.     private static Serializer serializer = new HessianSerializer();
    138.     private void initAdminBizList(String adminAddresses, String accessToken) throws Exception {
    139.         if (adminAddresses!=null && adminAddresses.trim().length()>0) {
    140.             for (String address: adminAddresses.trim().split(",")) {
    141.                 if (address!=null && address.trim().length()>0) {
    142.  
    143.                     AdminBiz adminBiz = new AdminBizClient(address.trim(), accessToken);
    144.  
    145.                     if (adminBizList == null) {
    146.                         adminBizList = new ArrayList();
    147.                     }
    148.                     adminBizList.add(adminBiz);
    149.                 }
    150.             }
    151.         }
    152.     }
    153.     public static List getAdminBizList(){
    154.         return adminBizList;
    155.     }
    156.     public static Serializer getSerializer() {
    157.         return serializer;
    158.     }
    159.  
    160.  
    161.     // ---------------------- executor-server (rpc provider) ----------------------
    162.     private XxlRpcProviderFactory xxlRpcProviderFactory = null;
    163.  
    164.     private void initRpcProvider(String ip, int port, String appName, String accessToken) throws Exception {
    165.  
    166.         // init, provider factory
    167.         String address = IpUtil.getIpPort(ip, port);
    168.         Map serviceRegistryParam = new HashMap();
    169.         serviceRegistryParam.put("appName", appName);
    170.         serviceRegistryParam.put("address", address);
    171.  
    172.         //初始化RPC配置
    173.         xxlRpcProviderFactory = new XxlRpcProviderFactory();
    174.  
    175.         xxlRpcProviderFactory.setServer(NettyHttpServer.class);
    176.         xxlRpcProviderFactory.setSerializer(HessianSerializer.class);
    177.         xxlRpcProviderFactory.setCorePoolSize(20);
    178.         xxlRpcProviderFactory.setMaxPoolSize(200);
    179.         xxlRpcProviderFactory.setIp(ip);
    180.         xxlRpcProviderFactory.setPort(port);
    181.         xxlRpcProviderFactory.setAccessToken(accessToken);
    182.         xxlRpcProviderFactory.setServiceRegistry(ExecutorServiceRegistry.class);
    183.         xxlRpcProviderFactory.setServiceRegistryParam(serviceRegistryParam);
    184.  
    185.         // add services
    186.         //给xxlRpcProviderFactory加入服务
    187.         xxlRpcProviderFactory.addService(ExecutorBiz.class.getName(), null, new ExecutorBizImpl());
    188.  
    189.         // start
    190.         /**
    191.          * 开启注册
    192.          */
    193.         xxlRpcProviderFactory.start();
    194.  
    195.     }
    196.  
    197.     public static class ExecutorServiceRegistry extends ServiceRegistry {
    198.  
    199.         @Override
    200.         public void start(Map param) {
    201.             // start registry
    202.             ExecutorRegistryThread.getInstance().start(param.get("appName"), param.get("address"));
    203.         }
    204.         @Override
    205.         public void stop() {
    206.             // stop registry
    207.             ExecutorRegistryThread.getInstance().toStop();
    208.         }
    209.  
    210.         @Override
    211.         public boolean registry(Set keys, String value) {
    212.             return false;
    213.         }
    214.         @Override
    215.         public boolean remove(Set keys, String value) {
    216.             return false;
    217.         }
    218.         @Override
    219.         public Map> discovery(Set keys) {
    220.             return null;
    221.         }
    222.         @Override
    223.         public TreeSet discovery(String key) {
    224.             return null;
    225.         }
    226.  
    227.     }
    228.  
    229.     private void stopRpcProvider() {
    230.         // stop provider factory
    231.         try {
    232.             xxlRpcProviderFactory.stop();
    233.         } catch (Exception e) {
    234.             logger.error(e.getMessage(), e);
    235.         }
    236.     }
    237.  
    238.     /**
    239.      * 一个JobThread对应一个JobHandler
    240.      */
    241.  
    242.     // ---------------------- job handler repository（key为XxlJob注解中的bean名称，value为JobHandler） ----------------------
    243.     private static ConcurrentMap jobHandlerRepository = new ConcurrentHashMap();
    244.     public static IJobHandler registJobHandler(String name, IJobHandler jobHandler){
    245.         logger.info(">>>>>>>>>>> xxl-job register jobhandler success, name:{}, jobHandler:{}", name, jobHandler);
    246.         return jobHandlerRepository.put(name, jobHandler);
    247.     }
    248.     public static IJobHandler loadJobHandler(String name){
    249.         return jobHandlerRepository.get(name);
    250.     }
    251.  
    252.  
    253.     // ---------------------- job thread repository（key为任务id，即jobId，value为JobThread对象）----------------------
    254.     private static ConcurrentMap jobThreadRepository = new ConcurrentHashMap();
    255.     public static JobThread registJobThread(int jobId, IJobHandler handler, String removeOldReason){
    256.         JobThread newJobThread = new JobThread(jobId, handler);
    257.         newJobThread.start();
    258.         logger.info(">>>>>>>>>>> xxl-job regist JobThread success, jobId:{}, handler:{}", new Object[]{jobId, handler});
    259.  
    260.         JobThread oldJobThread = jobThreadRepository.put(jobId, newJobThread);	// putIfAbsent | oh my god, map's put method return the old value!!!
    261.         if (oldJobThread != null) {
    262.             oldJobThread.toStop(removeOldReason);
    263.             oldJobThread.interrupt();
    264.         }
    265.  
    266.         return newJobThread;
    267.     }
    268.     public static void removeJobThread(int jobId, String removeOldReason){
    269.         JobThread oldJobThread = jobThreadRepository.remove(jobId);
    270.         if (oldJobThread != null) {
    271.             oldJobThread.toStop(removeOldReason);
    272.             oldJobThread.interrupt();
    273.         }
    274.     }
    275.     public static JobThread loadJobThread(int jobId){
    276.         JobThread jobThread = jobThreadRepository.get(jobId);
    277.         return jobThread;
    278.     }
    279.  
    280. }

    4.2.3 任务执行回调

    任务执行后结果回调：TriggerCallbackThread

    1. package com.xxl.job.core.thread;
    2.  
    3. import com.xxl.job.core.biz.AdminBiz;
    4. import com.xxl.job.core.biz.model.HandleCallbackParam;
    5. import com.xxl.job.core.biz.model.ReturnT;
    6. import com.xxl.job.core.enums.RegistryConfig;
    7. import com.xxl.job.core.executor.XxlJobExecutor;
    8. import com.xxl.job.core.log.XxlJobFileAppender;
    9. import com.xxl.job.core.log.XxlJobLogger;
    10. import com.xxl.job.core.util.FileUtil;
    11. import org.slf4j.Logger;
    12. import org.slf4j.LoggerFactory;
    13.  
    14. import java.io.File;
    15. import java.util.ArrayList;
    16. import java.util.Date;
    17. import java.util.List;
    18. import java.util.concurrent.LinkedBlockingQueue;
    19. import java.util.concurrent.TimeUnit;
    20.  
    21. /**
    22.  * Created by xuxueli on 16/7/22.
    23.  */
    24. public class TriggerCallbackThread {
    25.     private static Logger logger = LoggerFactory.getLogger(TriggerCallbackThread.class);
    26.  
    27.     private static TriggerCallbackThread instance = new TriggerCallbackThread();
    28.     public static TriggerCallbackThread getInstance(){
    29.         return instance;
    30.     }
    31.  
    32.     /**
    33.      * job results callback queue
    34.      */
    35.     private LinkedBlockingQueue callBackQueue = new LinkedBlockingQueue();
    36.     public static void pushCallBack(HandleCallbackParam callback){
    37.         getInstance().callBackQueue.add(callback);
    38.         logger.debug(">>>>>>>>>>> xxl-job, push callback request, logId:{}", callback.getLogId());
    39.     }
    40.  
    41.     /**
    42.      * callback thread
    43.      */
    44.     private Thread triggerCallbackThread;
    45.     private Thread triggerRetryCallbackThread;
    46.     private volatile boolean toStop = false;
    47.     public void start() {
    48.  
    49.         // valid
    50.         if (XxlJobExecutor.getAdminBizList() == null) {
    51.             logger.warn(">>>>>>>>>>> xxl-job, executor callback config fail, adminAddresses is null.");
    52.             return;
    53.         }
    54.  
    55.         // callback
    56.         triggerCallbackThread = new Thread(new Runnable() {
    57.  
    58.             @Override
    59.             public void run() {
    60.  
    61.                 // normal callback
    62.                 while(!toStop){
    63.                     try {
    64.                         HandleCallbackParam callback = getInstance().callBackQueue.take();
    65.                         if (callback != null) {
    66.  
    67.                             // callback list param
    68.                             List callbackParamList = new ArrayList();
    69.                             int drainToNum = getInstance().callBackQueue.drainTo(callbackParamList);
    70.                             callbackParamList.add(callback);
    71.  
    72.                             // callback, will retry if error
    73.                             if (callbackParamList!=null && callbackParamList.size()>0) {
    74.                                 doCallback(callbackParamList);
    75.                             }
    76.                         }
    77.                     } catch (Exception e) {
    78.                         if (!toStop) {
    79.                             logger.error(e.getMessage(), e);
    80.                         }
    81.                     }
    82.                 }
    83.  
    84.                 // last callback
    85.                 try {
    86.                     List callbackParamList = new ArrayList();
    87.                     int drainToNum = getInstance().callBackQueue.drainTo(callbackParamList);
    88.                     if (callbackParamList!=null && callbackParamList.size()>0) {
    89.                         doCallback(callbackParamList);
    90.                     }
    91.                 } catch (Exception e) {
    92.                     if (!toStop) {
    93.                         logger.error(e.getMessage(), e);
    94.                     }
    95.                 }
    96.                 logger.info(">>>>>>>>>>> xxl-job, executor callback thread destory.");
    97.  
    98.             }
    99.         });
    100.         triggerCallbackThread.setDaemon(true);
    101.         triggerCallbackThread.setName("xxl-job, executor TriggerCallbackThread");
    102.         triggerCallbackThread.start();
    103.  
    104.  
    105.         // retry
    106.         triggerRetryCallbackThread = new Thread(new Runnable() {
    107.             @Override
    108.             public void run() {
    109.                 while(!toStop){
    110.                     try {
    111.                         retryFailCallbackFile();
    112.                     } catch (Exception e) {
    113.                         if (!toStop) {
    114.                             logger.error(e.getMessage(), e);
    115.                         }
    116.  
    117.                     }
    118.                     try {
    119.                         TimeUnit.SECONDS.sleep(RegistryConfig.BEAT_TIMEOUT);
    120.                     } catch (InterruptedException e) {
    121.                         if (!toStop) {
    122.                             logger.error(e.getMessage(), e);
    123.                         }
    124.                     }
    125.                 }
    126.                 logger.info(">>>>>>>>>>> xxl-job, executor retry callback thread destory.");
    127.             }
    128.         });
    129.         triggerRetryCallbackThread.setDaemon(true);
    130.         triggerRetryCallbackThread.start();
    131.  
    132.     }
    133.     public void toStop(){
    134.         toStop = true;
    135.         // stop callback, interrupt and wait
    136.         if (triggerCallbackThread != null) {    // support empty admin address
    137.             triggerCallbackThread.interrupt();
    138.             try {
    139.                 triggerCallbackThread.join();
    140.             } catch (InterruptedException e) {
    141.                 logger.error(e.getMessage(), e);
    142.             }
    143.         }
    144.  
    145.         // stop retry, interrupt and wait
    146.         if (triggerRetryCallbackThread != null) {
    147.             triggerRetryCallbackThread.interrupt();
    148.             try {
    149.                 triggerRetryCallbackThread.join();
    150.             } catch (InterruptedException e) {
    151.                 logger.error(e.getMessage(), e);
    152.             }
    153.         }
    154.  
    155.     }
    156.  
    157.     /**
    158.      * do callback, will retry if error
    159.      * @param callbackParamList
    160.      */
    161.     private void doCallback(List callbackParamList){
    162.         boolean callbackRet = false;
    163.         // callback, will retry if error
    164.         for (AdminBiz adminBiz: XxlJobExecutor.getAdminBizList()) {
    165.             try {
    166.                 ReturnT callbackResult = adminBiz.callback(callbackParamList);
    167.                 if (callbackResult!=null && ReturnT.SUCCESS_CODE == callbackResult.getCode()) {
    168.                     callbackLog(callbackParamList, "
      ----------- xxl-job job callback finish."      );
    169.                     callbackRet = true;
    170.                     break;
    171.                 } else {
    172.                     callbackLog(callbackParamList, "
      ----------- xxl-job job callback fail, callbackResult:"       + callbackResult);
    173.                 }
    174.             } catch (Exception e) {
    175.                 callbackLog(callbackParamList, "
      ----------- xxl-job job callback error, errorMsg:"       + e.getMessage());
    176.             }
    177.         }
    178.         if (!callbackRet) {
    179.             appendFailCallbackFile(callbackParamList);
    180.         }
    181.     }
    182.  
    183.     /**
    184.      * callback log
    185.      */
    186.     private void callbackLog(List callbackParamList, String logContent){
    187.         for (HandleCallbackParam callbackParam: callbackParamList) {
    188.             String logFileName = XxlJobFileAppender.makeLogFileName(new Date(callbackParam.getLogDateTim()), callbackParam.getLogId());
    189.             XxlJobFileAppender.contextHolder.set(logFileName);
    190.             XxlJobLogger.log(logContent);
    191.         }
    192.     }
    193.  
    194.  
    195.     // ---------------------- fail-callback file ----------------------
    196.  
    197.     private static String failCallbackFilePath = XxlJobFileAppender.getLogPath().concat(File.separator).concat("callbacklog").concat(File.separator);
    198.     private static String failCallbackFileName = failCallbackFilePath.concat("xxl-job-callback-{x}").concat(".log");
    199.  
    200.     private void appendFailCallbackFile(List callbackParamList){
    201.         // valid
    202.         if (callbackParamList==null || callbackParamList.size()==0) {
    203.             return;
    204.         }
    205.  
    206.         // append file
    207.         byte[] callbackParamList_bytes = XxlJobExecutor.getSerializer().serialize(callbackParamList);
    208.  
    209.         File callbackLogFile = new File(failCallbackFileName.replace("{x}", String.valueOf(System.currentTimeMillis())));
    210.         if (callbackLogFile.exists()) {
    211.             for (int i = 0; i < 100; i++) {
    212.                 callbackLogFile = new File(failCallbackFileName.replace("{x}", String.valueOf(System.currentTimeMillis()).concat("-").concat(String.valueOf(i)) ));
    213.                 if (!callbackLogFile.exists()) {
    214.                     break;
    215.                 }
    216.             }
    217.         }
    218.         FileUtil.writeFileContent(callbackLogFile, callbackParamList_bytes);
    219.     }
    220.  
    221.     private void retryFailCallbackFile(){
    222.  
    223.         // valid
    224.         File callbackLogPath = new File(failCallbackFilePath);
    225.         if (!callbackLogPath.exists()) {
    226.             return;
    227.         }
    228.         if (callbackLogPath.isFile()) {
    229.             callbackLogPath.delete();
    230.         }
    231.         if (!(callbackLogPath.isDirectory() && callbackLogPath.list()!=null && callbackLogPath.list().length>0)) {
    232.             return;
    233.         }
    234.  
    235.         // load and clear file, retry
    236.         for (File callbaclLogFile: callbackLogPath.listFiles()) {
    237.             byte[] callbackParamList_bytes = FileUtil.readFileContent(callbaclLogFile);
    238.             List callbackParamList = (List) XxlJobExecutor.getSerializer().deserialize(callbackParamList_bytes, HandleCallbackParam.class);
    239.  
    240.             callbaclLogFile.delete();
    241.             doCallback(callbackParamList);
    242.         }
    243.  
    244.     }
    245.  
    246. }

    4.3 执行器注册到调度中心时序图

    4.4 执行器注册到调度中心核心代码解析

    4.4.1 执行注册（30秒一次）

    将执行器注册到调度中心：ExecutorRegistryThread

    1. package com.xxl.job.core.thread;
    2.  
    3. import com.xxl.job.core.biz.AdminBiz;
    4. import com.xxl.job.core.biz.model.RegistryParam;
    5. import com.xxl.job.core.biz.model.ReturnT;
    6. import com.xxl.job.core.enums.RegistryConfig;
    7. import com.xxl.job.core.executor.XxlJobExecutor;
    8. import org.slf4j.Logger;
    9. import org.slf4j.LoggerFactory;
    10.  
    11. import java.util.concurrent.TimeUnit;
    12.  
    13. /**
    14.  * Created by xuxueli on 17/3/2.
    15.  * @Description：将执行器注册到调度中心
    16.  */
    17. public class ExecutorRegistryThread {
    18.     private static Logger logger = LoggerFactory.getLogger(ExecutorRegistryThread.class);
    19.  
    20.     private static ExecutorRegistryThread instance = new ExecutorRegistryThread();
    21.     public static ExecutorRegistryThread getInstance(){
    22.         return instance;
    23.     }
    24.  
    25.     private Thread registryThread;
    26.     private volatile boolean toStop = false;
    27.     public void start(final String appName, final String address){
    28.  
    29.         // valid
    30.         if (appName==null || appName.trim().length()==0) {
    31.             logger.warn(">>>>>>>>>>> xxl-job, executor registry config fail, appName is null.");
    32.             return;
    33.         }
    34.         if (XxlJobExecutor.getAdminBizList() == null) {
    35.             logger.warn(">>>>>>>>>>> xxl-job, executor registry config fail, adminAddresses is null.");
    36.             return;
    37.         }
    38.  
    39.         registryThread = new Thread(new Runnable() {
    40.             @Override
    41.             public void run() {
    42.  
    43.                 // registry
    44.                 while (!toStop) {
    45.                     try {
    46.                         RegistryParam registryParam = new RegistryParam(RegistryConfig.RegistType.EXECUTOR.name(), appName, address);
    47.                         for (AdminBiz adminBiz: XxlJobExecutor.getAdminBizList()) {
    48.                             try {
    49.                                 ReturnT registryResult = adminBiz.registry(registryParam);
    50.                                 if (registryResult!=null && ReturnT.SUCCESS_CODE == registryResult.getCode()) {
    51.                                     registryResult = ReturnT.SUCCESS;
    52.                                     logger.debug(">>>>>>>>>>> xxl-job registry success, registryParam:{}, registryResult:{}", new Object[]{registryParam, registryResult});
    53.                                     break;
    54.                                 } else {
    55.                                     logger.info(">>>>>>>>>>> xxl-job registry fail, registryParam:{}, registryResult:{}", new Object[]{registryParam, registryResult});
    56.                                 }
    57.                             } catch (Exception e) {
    58.                                 logger.info(">>>>>>>>>>> xxl-job registry error, registryParam:{}", registryParam, e);
    59.                             }
    60.  
    61.                         }
    62.                     } catch (Exception e) {
    63.                         if (!toStop) {
    64.                             logger.error(e.getMessage(), e);
    65.                         }
    66.  
    67.                     }
    68.  
    69.                     try {
    70.                         if (!toStop) {
    71.                             TimeUnit.SECONDS.sleep(RegistryConfig.BEAT_TIMEOUT);
    72.                         }
    73.                     } catch (InterruptedException e) {
    74.                         if (!toStop) {
    75.                             logger.warn(">>>>>>>>>>> xxl-job, executor registry thread interrupted, error msg:{}", e.getMessage());
    76.                         }
    77.                     }
    78.                 }
    79.  
    80.                 // registry remove
    81.                 try {
    82.                     RegistryParam registryParam = new RegistryParam(RegistryConfig.RegistType.EXECUTOR.name(), appName, address);
    83.                     for (AdminBiz adminBiz: XxlJobExecutor.getAdminBizList()) {
    84.                         try {
    85.                             ReturnT registryResult = adminBiz.registryRemove(registryParam);
    86.                             if (registryResult!=null && ReturnT.SUCCESS_CODE == registryResult.getCode()) {
    87.                                 registryResult = ReturnT.SUCCESS;
    88.                                 logger.info(">>>>>>>>>>> xxl-job registry-remove success, registryParam:{}, registryResult:{}", new Object[]{registryParam, registryResult});
    89.                                 break;
    90.                             } else {
    91.                                 logger.info(">>>>>>>>>>> xxl-job registry-remove fail, registryParam:{}, registryResult:{}", new Object[]{registryParam, registryResult});
    92.                             }
    93.                         } catch (Exception e) {
    94.                             if (!toStop) {
    95.                                 logger.info(">>>>>>>>>>> xxl-job registry-remove error, registryParam:{}", registryParam, e);
    96.                             }
    97.  
    98.                         }
    99.  
    100.                     }
    101.                 } catch (Exception e) {
    102.                     if (!toStop) {
    103.                         logger.error(e.getMessage(), e);
    104.                     }
    105.                 }
    106.                 logger.info(">>>>>>>>>>> xxl-job, executor registry thread destory.");
    107.  
    108.             }
    109.         });
    110.         registryThread.setDaemon(true);
    111.         registryThread.setName("xxl-job, executor ExecutorRegistryThread");
    112.         registryThread.start();
    113.     }
    114.  
    115.     public void toStop() {
    116.         toStop = true;
    117.         // interrupt and wait
    118.         registryThread.interrupt();
    119.         try {
    120.             registryThread.join();
    121.         } catch (InterruptedException e) {
    122.             logger.error(e.getMessage(), e);
    123.         }
    124.     }
    125.  
    126. }

  • 相关阅读:
    安装插件时Vscode XHR Failed 报错ERR_CERT_AUTHORITY_INVALID
    结构型-代理模式
    Dubbo框架基本使用
    数据结构-二叉树的递归遍历
    Linux中Docker挂载mysql/mariadb等数据库,数据库问题汇总
    React核心原理与实际开发
    Complete Binary Tree
    [Linux 基础] 一篇带你了解linux权限问题
    js 对象循环遍历
    CV 面试指南—深度学习知识点总结（5）
  • 原文地址：https://blog.csdn.net/YYQ_QYY/article/details/126366987