xxl-job源码—调度器/执行器工作原理
·
目录
一、架构图
1.1 功能架构图
2.2 任务调度工作原理
二、ER图
三、调度器
3.1 启动过程时序图
3.2 启动过程核心代码解析
入口类：XxlJobAdminConfig（PS：启动初始化XxlJobScheduler）
3.2.1 启动初始化
调度器启动初始化：XxlJobScheduler
package com.xxl.job.admin.core.scheduler; import com.xxl.job.admin.core.conf.XxlJobAdminConfig; import com.xxl.job.admin.core.thread.*; import com.xxl.job.admin.core.util.I18nUtil; import com.xxl.job.core.biz.ExecutorBiz; import com.xxl.job.core.enums.ExecutorBlockStrategyEnum; import com.xxl.rpc.remoting.invoker.call.CallType; import com.xxl.rpc.remoting.invoker.reference.XxlRpcReferenceBean; import com.xxl.rpc.remoting.invoker.route.LoadBalance; import com.xxl.rpc.remoting.net.impl.netty_http.client.NettyHttpClient; import com.xxl.rpc.serialize.impl.HessianSerializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; /** * @author xuxueli 2018-10-28 00:18:17 * Description：调度程序 */ public class XxlJobScheduler { private static final Logger logger = LoggerFactory.getLogger(XxlJobScheduler.class); public void init() throws Exception { // init i18n /** * 主要作用：调度中心语言选择 */ initI18n(); // admin registry monitor run /** * 主要作用是开通了一个守护线程，每隔30s扫描一次执行器的注册信息表 * 1、剔除90s内没有进行健康检查的执行器信息 * 2、将自动注册类型的执行器注册信息（XxlJobRegistry）经过处理更新执行器信息(XxlJobGroup) */ JobRegistryMonitorHelper.getInstance().start(); // admin monitor run /** * 主要作用：开通了一个守10s扫描一次失败日志 * 1、如果任务失败可重试次数>0，那么重新触发任务 * 2、如果任务执行失败，会进行告警，默认采用邮件形式进行告警 */ JobFailMonitorHelper.getInstance().start(); // admin trigger pool start /** * 主要作用：初始化一个快速执行的线程池，一个稍后执行的线程池 * */ JobTriggerPoolHelper.toStart(); // admin log report start /** * 主要作用：开通一个守护线程，每隔1min扫描一次最近3天的调度日志 * 1、更新每天总任务数、正在执行数、执行成功数、执行失败数 */ JobLogReportHelper.getInstance().start(); // start-schedule /** * 主要作用： * 1、开通一个守护线程，每隔5s扫表一次执行器的任务表 * 1.1 执行已到执行时间的任务，并更新任务的下次执行时间 * * 2、开通一个守护线程，每隔1s循环执行一次待执行的任务 * 2.1 避免任务执行遗漏 */ JobScheduleHelper.getInstance().start(); logger.info(">>>>>>>>> init xxl-job admin success."); } public void destroy() throws Exception { // stop-schedule JobScheduleHelper.getInstance().toStop(); // admin log report stop JobLogReportHelper.getInstance().toStop(); // admin trigger pool stop JobTriggerPoolHelper.toStop(); // admin monitor stop JobFailMonitorHelper.getInstance().toStop(); // admin registry stop JobRegistryMonitorHelper.getInstance().toStop(); } // ---------------------- I18n ---------------------- private void initI18n(){ for (ExecutorBlockStrategyEnum item:ExecutorBlockStrategyEnum.values()) { // 根据语言选择，去对应的message_xxx.properties中取阻塞策略的title item.setTitle(I18nUtil.getString("jobconf_block_".concat(item.name()))); } } // ---------------------- executor-client ---------------------- private static ConcurrentMap<String, ExecutorBiz> executorBizRepository = new ConcurrentHashMap<String, ExecutorBiz>(); public static ExecutorBiz getExecutorBiz(String address) throws Exception { // valid if (address==null || address.trim().length()==0) { return null; } // load-cache address = address.trim(); ExecutorBiz executorBiz = executorBizRepository.get(address); if (executorBiz != null) { return executorBiz; } // set-cache // 创建ExecutorBiz的代理对象，重点在这个getObject方法 XxlRpcReferenceBean referenceBean = new XxlRpcReferenceBean(); referenceBean.setClient(NettyHttpClient.class); referenceBean.setSerializer(HessianSerializer.class); referenceBean.setCallType(CallType.SYNC); referenceBean.setLoadBalance(LoadBalance.ROUND); referenceBean.setIface(ExecutorBiz.class); referenceBean.setVersion(null); referenceBean.setTimeout(3000); referenceBean.setAddress(address); referenceBean.setAccessToken(XxlJobAdminConfig.getAdminConfig().getAccessToken()); referenceBean.setInvokeCallback(null); referenceBean.setInvokerFactory(null); executorBiz = (ExecutorBiz) referenceBean.getObject(); executorBizRepository.put(address, executorBiz); return executorBiz; } }
3.2.2 执行器健康检查
执行器注册健康检查：JobRegistryMonitorHelper
package com.xxl.job.admin.core.thread; import com.xxl.job.admin.core.conf.XxlJobAdminConfig; import com.xxl.job.admin.core.model.XxlJobGroup; import com.xxl.job.admin.core.model.XxlJobRegistry; import com.xxl.job.core.enums.RegistryConfig; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.*; import java.util.concurrent.TimeUnit; /** * job registry instance * @author xuxueli 2016-10-02 19:10:24 * @Description：任务注册监控助手 */ public class JobRegistryMonitorHelper { private static Logger logger = LoggerFactory.getLogger(JobRegistryMonitorHelper.class); private static JobRegistryMonitorHelper instance = new JobRegistryMonitorHelper(); public static JobRegistryMonitorHelper getInstance(){ return instance; } private Thread registryThread; private volatile boolean toStop = false; public void start(){ //创建一个线程 registryThread = new Thread(new Runnable() { @Override public void run() { //当toStop为false时进入该循环（注意：toStop是用volatile修饰的） while (!toStop) { try { // auto registry group //获取类型为自动注册的执行器（XxlJobGroup）地址列表 List<XxlJobGroup> groupList = XxlJobAdminConfig.getAdminConfig().getXxlJobGroupDao().findByAddressType(0); if (groupList!=null && !groupList.isEmpty()) { // remove dead address (admin/executor) //删除90秒内没有更新的注册机器信息，90秒没有心跳信息返回代表机器已经出现问题，所以移除 List<Integer> ids = XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().findDead(RegistryConfig.DEAD_TIMEOUT, new Date()); if (ids!=null && ids.size()>0) { XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().removeDead(ids); } // fresh online address (admin/executor) HashMap<String, List<String>> appAddressMap = new HashMap<String, List<String>>(); //查询90秒内有更新的注册机器信息列表 List<XxlJobRegistry> list = XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().findAll(RegistryConfig.DEAD_TIMEOUT, new Date()); if (list != null) { //遍历注册信息列表，得到自动注册类型的执行器与其对应的地址信息关系Map for (XxlJobRegistry item: list) { if (RegistryConfig.RegistType.EXECUTOR.name().equals(item.getRegistryGroup())) { String appName = item.getRegistryKey(); List<String> registryList = appAddressMap.get(appName); if (registryList == null) { registryList = new ArrayList<String>(); } if (!registryList.contains(item.getRegistryValue())) { registryList.add(item.getRegistryValue()); } //收集执行器信息，根据执行器appName做区分：appAddressMap的key为appName，value为此执行器的注册地址列表（集群环境下会有多个注册地址） appAddressMap.put(appName, registryList); } } } // fresh group address //遍历所有的自动注册的执行器 for (XxlJobGroup group: groupList) { //通过执行器的appName从刚刚区分的Map中拿到该执行器下的集群机器注册地址 List<String> registryList = appAddressMap.get(group.getAppName()); String addressListStr = null; if (registryList!=null && !registryList.isEmpty()) { Collections.sort(registryList); addressListStr = ""; //集群的多个注册地址通过逗号拼接转为字符串 for (String item:registryList) { addressListStr += item + ","; } addressListStr = addressListStr.substring(0, addressListStr.length()-1); } //集群的多个注册地址通过逗号拼接转成的字符串设置进执行器的addressList属性中，并更新执行器信息，保存到DB group.setAddressList(addressListStr); XxlJobAdminConfig.getAdminConfig().getXxlJobGroupDao().update(group); } } } catch (Exception e) { if (!toStop) { logger.error(">>>>>>>>>>> xxl-job, job registry monitor thread error:{}", e); } } try { //线程停顿30秒 TimeUnit.SECONDS.sleep(RegistryConfig.BEAT_TIMEOUT); } catch (InterruptedException e) { if (!toStop) { logger.error(">>>>>>>>>>> xxl-job, job registry monitor thread error:{}", e); } } } logger.info(">>>>>>>>>>> xxl-job, job registry monitor thread stop"); } }); //将此线程设置成守护线程 registryThread.setDaemon(true); registryThread.setName("xxl-job, admin JobRegistryMonitorHelper"); //执行该线程 registryThread.start(); } public void toStop(){ toStop = true; // interrupt and wait registryThread.interrupt(); try { registryThread.join(); } catch (InterruptedException e) { logger.error(e.getMessage(), e); } } }
3.2.3 任务执行失败告警
任务失败监控助手：JobFailMonitorHelper
package com.xxl.job.admin.core.thread; import com.xxl.job.admin.core.conf.XxlJobAdminConfig; import com.xxl.job.admin.core.model.XxlJobInfo; import com.xxl.job.admin.core.model.XxlJobLog; import com.xxl.job.admin.core.trigger.TriggerTypeEnum; import com.xxl.job.admin.core.util.I18nUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.List; import java.util.concurrent.TimeUnit; /** * job monitor instance * * @author xuxueli 2015-9-1 18:05:56 * @Description：任务失败监控助手 */ public class JobFailMonitorHelper { private static Logger logger = LoggerFactory.getLogger(JobFailMonitorHelper.class); private static JobFailMonitorHelper instance = new JobFailMonitorHelper(); public static JobFailMonitorHelper getInstance(){ return instance; } // ---------------------- monitor ---------------------- private Thread monitorThread; private volatile boolean toStop = false; public void start(){ //创建一个监控线程 monitorThread = new Thread(new Runnable() { @Override public void run() { // monitor while (!toStop) { try { //从数据库中取出所有执行失败且告警状态为0（默认）的日志ID，虽然这里传了pageSize为1000，实际没有进行分页，取出来的是所有符合条件的失败日志 List<Long> failLogIds = XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().findFailJobLogIds(1000); if (failLogIds!=null && !failLogIds.isEmpty()) { for (long failLogId: failLogIds) { // 锁定日志 int lockRet = XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().updateAlarmStatus(failLogId, 0, -1); if (lockRet < 1) { continue; } //取出失败日志完整信息 XxlJobLog log = XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().load(failLogId); //获取失败日志对应的任务信息 XxlJobInfo info = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().loadById(log.getJobId()); // 如果剩余失败可重试次数>0（注意：日志里存的失败重试次数实为剩余可重复次数） if (log.getExecutorFailRetryCount() > 0) { //触发任务执行 JobTriggerPoolHelper.trigger(log.getJobId(), TriggerTypeEnum.RETRY, (log.getExecutorFailRetryCount()-1), log.getExecutorShardingParam(), log.getExecutorParam()); String retryMsg = "<br><br><span style=\"color:#F39C12;\" > >>>>>>>>>>>"+ I18nUtil.getString("jobconf_trigger_type_retry") +"<<<<<<<<<<< </span><br>"; log.setTriggerMsg(log.getTriggerMsg() + retryMsg); XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().updateTriggerInfo(log); } // 2、fail alarm monitor //失败告警 int newAlarmStatus = 0; // 告警状态：0-默认、-1=锁定状态、1-无需告警、2-告警成功、3-告警失败 if (info!=null && info.getAlarmEmail()!=null && info.getAlarmEmail().trim().length()>0) { //发送失败告警：默认是邮件通知 boolean alarmResult = XxlJobAdminConfig.getAdminConfig().getJobAlarmer().alarm(info, log); newAlarmStatus = alarmResult?2:3; } else { newAlarmStatus = 1; } //更新告警状态 XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().updateAlarmStatus(failLogId, -1, newAlarmStatus); } } } catch (Exception e) { if (!toStop) { logger.error(">>>>>>>>>>> xxl-job, job fail monitor thread error:{}", e); } } try { TimeUnit.SECONDS.sleep(10); } catch (Exception e) { if (!toStop) { logger.error(e.getMessage(), e); } } } logger.info(">>>>>>>>>>> xxl-job, job fail monitor thread stop"); } }); monitorThread.setDaemon(true); monitorThread.setName("xxl-job, admin JobFailMonitorHelper"); monitorThread.start(); } public void toStop(){ toStop = true; // interrupt and wait monitorThread.interrupt(); try { monitorThread.join(); } catch (InterruptedException e) { logger.error(e.getMessage(), e); } } }
3.2.4 调度线程池
分配任务执行线程：JobTriggerPoolHelper
package com.xxl.job.admin.core.thread; import com.xxl.job.admin.core.conf.XxlJobAdminConfig; import com.xxl.job.admin.core.trigger.TriggerTypeEnum; import com.xxl.job.admin.core.trigger.XxlJobTrigger; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; /** * job trigger thread pool helper * * @author xuxueli 2018-07-03 21:08:07 */ public class JobTriggerPoolHelper { private static Logger logger = LoggerFactory.getLogger(JobTriggerPoolHelper.class); // ---------------------- trigger pool ---------------------- // fast/slow thread pool private ThreadPoolExecutor fastTriggerPool = null; private ThreadPoolExecutor slowTriggerPool = null; public void start(){ fastTriggerPool = new ThreadPoolExecutor( 10, XxlJobAdminConfig.getAdminConfig().getTriggerPoolFastMax(), 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(1000), new ThreadFactory() { @Override public Thread newThread(Runnable r) { return new Thread(r, "xxl-job, admin JobTriggerPoolHelper-fastTriggerPool-" + r.hashCode()); } }); slowTriggerPool = new ThreadPoolExecutor( 10, XxlJobAdminConfig.getAdminConfig().getTriggerPoolSlowMax(), 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(2000), new ThreadFactory() { @Override public Thread newThread(Runnable r) { return new Thread(r, "xxl-job, admin JobTriggerPoolHelper-slowTriggerPool-" + r.hashCode()); } }); } public void stop() { //triggerPool.shutdown(); fastTriggerPool.shutdownNow(); slowTriggerPool.shutdownNow(); logger.info(">>>>>>>>> xxl-job trigger thread pool shutdown success."); } // job timeout count private volatile long minTim = System.currentTimeMillis()/60000; // ms > min private volatile ConcurrentMap<Integer, AtomicInteger> jobTimeoutCountMap = new ConcurrentHashMap<>(); /** * add trigger */ public void addTrigger(final int jobId, final TriggerTypeEnum triggerType, final int failRetryCount, final String executorShardingParam, final String executorParam) { // choose thread pool //选择线程池类型 ThreadPoolExecutor triggerPool_ = fastTriggerPool; AtomicInteger jobTimeoutCount = jobTimeoutCountMap.get(jobId); // 1分钟窗口期内任务耗时达500ms超过10次则判定为慢任务，慢任务自动降级进入"Slow"线程池，避免耗尽调度线程，提高系统稳定性； if (jobTimeoutCount!=null && jobTimeoutCount.get() > 10) { // job-timeout 10 times in 1 min triggerPool_ = slowTriggerPool; } // trigger triggerPool_.execute(new Runnable() { @Override public void run() { long start = System.currentTimeMillis(); try { // do trigger //这里是重点，触发任务的执行 XxlJobTrigger.trigger(jobId, triggerType, failRetryCount, executorShardingParam, executorParam); } catch (Exception e) { logger.error(e.getMessage(), e); } finally { //在finally代码块中通过执行耗时时间是否>500ms将jobId与超时次数记录到到超时记录容器 // check timeout-count-map //检查超时记录容器 long minTim_now = System.currentTimeMillis()/60000; if (minTim != minTim_now) { //说明当前分钟数和上面的默认分钟数不同，即超过了1分钟 //这里的逻辑主要是为了使得超时记录容器以1分钟为时间间隔，也就是说超时容器1分钟记录1次超时的任务，到了下一分钟清空容器，重新记录 minTim = minTim_now; jobTimeoutCountMap.clear(); } // incr timeout-count-map //如果线程执行时间>500毫秒：将此任务ID与次数记录进超时记录容器jobTimeoutCountMap long cost = System.currentTimeMillis()-start; if (cost > 500) { // ob-timeout threshold 500ms //这里利用AtomicInteger的原子性，保证多线程并发结果的准确性 AtomicInteger timeoutCount = jobTimeoutCountMap.putIfAbsent(jobId, new AtomicInteger(1)); if (timeoutCount != null) { timeoutCount.incrementAndGet(); } } } } }); } // ---------------------- helper ---------------------- private static JobTriggerPoolHelper helper = new JobTriggerPoolHelper(); public static void toStart() { helper.start(); } public static void toStop() { helper.stop(); } /** * @param jobId * @param triggerType * @param failRetryCount * >=0: use this param * <0: use param from job info config * @param executorShardingParam * @param executorParam * null: use job param * not null: cover job param */ public static void trigger(int jobId, TriggerTypeEnum triggerType, int failRetryCount, String executorShardingParam, String executorParam) { helper.addTrigger(jobId, triggerType, failRetryCount, executorShardingParam, executorParam); } }
3.2.5 调度中心
核心调度程序：JobScheduleHelper
package com.xxl.job.admin.core.thread; import com.xxl.job.admin.core.conf.XxlJobAdminConfig; import com.xxl.job.admin.core.cron.CronExpression; import com.xxl.job.admin.core.model.XxlJobInfo; import com.xxl.job.admin.core.trigger.TriggerTypeEnum; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.sql.Connection; import java.sql.PreparedStatement; import java.sql.SQLException; import java.text.ParseException; import java.util.*; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; /** * @author xuxueli 2019-05-21 * @Description：调度中心 */ public class JobScheduleHelper { private static Logger logger = LoggerFactory.getLogger(JobScheduleHelper.class); private static JobScheduleHelper instance = new JobScheduleHelper(); public static JobScheduleHelper getInstance(){ return instance; } public static final long PRE_READ_MS = 5000; // pre read private Thread scheduleThread; private Thread ringThread; private volatile boolean scheduleThreadToStop = false; private volatile boolean ringThreadToStop = false; private volatile static Map<Integer, List<Integer>> ringData = new ConcurrentHashMap<>(); public void start(){ // schedule thread scheduleThread = new Thread(new Runnable() { @Override public void run() { try { TimeUnit.MILLISECONDS.sleep(5000 - System.currentTimeMillis()%1000 ); } catch (InterruptedException e) { if (!scheduleThreadToStop) { logger.error(e.getMessage(), e); } } logger.info(">>>>>>>>> init xxl-job admin scheduler success."); // pre-read count: treadpool-size * trigger-qps (each trigger cost 50ms, qps = 1000/50 = 20) // 预读触发数 = 线程数 * 触发执行qps（限制qps：20/s） int preReadCount = (XxlJobAdminConfig.getAdminConfig().getTriggerPoolFastMax() + XxlJobAdminConfig.getAdminConfig().getTriggerPoolSlowMax()) * 20; while (!scheduleThreadToStop) { // Scan Job long start = System.currentTimeMillis(); Connection conn = null; Boolean connAutoCommit = null; PreparedStatement preparedStatement = null; boolean preReadSuc = true; try { conn = XxlJobAdminConfig.getAdminConfig().getDataSource().getConnection(); connAutoCommit = conn.getAutoCommit(); conn.setAutoCommit(false); // 调度中心支持多节点部署，基于数据库行锁保证同时只有一个调度中心节点触发任务调度 preparedStatement = conn.prepareStatement( "select * from xxl_job_lock where lock_name = 'schedule_lock' for update" ); preparedStatement.execute(); // tx start // 1、pre read long nowTime = System.currentTimeMillis(); // 查询最近5s等待执行的任务（限制数量为preReadCount） List<XxlJobInfo> scheduleList = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().scheduleJobQuery(nowTime + PRE_READ_MS, preReadCount); if (scheduleList!=null && scheduleList.size()>0) { // 2、push time-ring for (XxlJobInfo jobInfo: scheduleList) { // time-ring jump if (nowTime > jobInfo.getTriggerNextTime() + PRE_READ_MS) { // 2.1、trigger-expire > 5s：pass && make next-trigger-time // 任务触发时间已过，则跳过执行 logger.warn(">>>>>>>>>>> xxl-job, schedule misfire, jobId = " + jobInfo.getId()); // fresh next // 更新该任务下次执行时间 refreshNextValidTime(jobInfo, new Date()); } else if (nowTime > jobInfo.getTriggerNextTime()) { // 2.2、trigger-expire < 5s：direct-trigger && make next-trigger-time // 任务触发时间已到，则立即执行 // 1、trigger JobTriggerPoolHelper.trigger(jobInfo.getId(), TriggerTypeEnum.CRON, -1, null, null); logger.debug(">>>>>>>>>>> xxl-job, schedule push trigger : jobId = " + jobInfo.getId() ); // 2、fresh next // 更新该任务下次执行时间 refreshNextValidTime(jobInfo, new Date()); // next-trigger-time in 5s, pre-read again // 任务下次触发时间在5s内，则需再执行一次 if (jobInfo.getTriggerStatus()==1 && nowTime + PRE_READ_MS > jobInfo.getTriggerNextTime()) { // 针对待下次执行，任务会已过执行时间，因此必须当前立即执行，避免任务执行遗漏 // 1、make ring second // 计算不足1s的剩余毫秒 int ringSecond = (int)((jobInfo.getTriggerNextTime()/1000)%60); // 2、push time ring // 存放待执行的循环执行容器 pushTimeRing(ringSecond, jobInfo.getId()); // 3、fresh next // 更新该任务下次执行时间 refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime())); } } else { // 2.3、trigger-pre-read：time-ring trigger && make next-trigger-time // 1、make ring second int ringSecond = (int)((jobInfo.getTriggerNextTime()/1000)%60); // 2、push time ring pushTimeRing(ringSecond, jobInfo.getId()); // 3、fresh next refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime())); } } // 3、update trigger info for (XxlJobInfo jobInfo: scheduleList) { XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().scheduleUpdate(jobInfo); } } else { preReadSuc = false; } // tx stop } catch (Exception e) { if (!scheduleThreadToStop) { logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#scheduleThread error:{}", e); } } finally { // commit if (conn != null) { try { conn.commit(); } catch (SQLException e) { if (!scheduleThreadToStop) { logger.error(e.getMessage(), e); } } try { conn.setAutoCommit(connAutoCommit); } catch (SQLException e) { if (!scheduleThreadToStop) { logger.error(e.getMessage(), e); } } try { conn.close(); } catch (SQLException e) { if (!scheduleThreadToStop) { logger.error(e.getMessage(), e); } } } // close PreparedStatement if (null != preparedStatement) { try { preparedStatement.close(); } catch (SQLException e) { if (!scheduleThreadToStop) { logger.error(e.getMessage(), e); } } } } long cost = System.currentTimeMillis()-start; // Wait seconds, align second if (cost < 1000) { // scan-overtime, not wait try { // pre-read period: success > scan each second; fail > skip this period; TimeUnit.MILLISECONDS.sleep((preReadSuc?1000:PRE_READ_MS) - System.currentTimeMillis()%1000); } catch (InterruptedException e) { if (!scheduleThreadToStop) { logger.error(e.getMessage(), e); } } } } logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper#scheduleThread stop"); } }); scheduleThread.setDaemon(true); scheduleThread.setName("xxl-job, admin JobScheduleHelper#scheduleThread"); scheduleThread.start(); // ring thread ringThread = new Thread(new Runnable() { @Override public void run() { // align second try { TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis()%1000 ); } catch (InterruptedException e) { if (!ringThreadToStop) { logger.error(e.getMessage(), e); } } while (!ringThreadToStop) { try { // second data List<Integer> ringItemData = new ArrayList<>(); int nowSecond = Calendar.getInstance().get(Calendar.SECOND); // 避免处理耗时太长，跨过刻度，向前校验一个刻度； for (int i = 0; i < 2; i++) { List<Integer> tmpData = ringData.remove( (nowSecond+60-i)%60 ); if (tmpData != null) { ringItemData.addAll(tmpData); } } // ring trigger logger.debug(">>>>>>>>>>> xxl-job, time-ring beat : " + nowSecond + " = " + Arrays.asList(ringItemData) ); if (ringItemData.size() > 0) { // do trigger for (int jobId: ringItemData) { // do trigger JobTriggerPoolHelper.trigger(jobId, TriggerTypeEnum.CRON, -1, null, null); } // clear ringItemData.clear(); } } catch (Exception e) { if (!ringThreadToStop) { logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#ringThread error:{}", e); } } // next second, align second try { TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis()%1000); } catch (InterruptedException e) { if (!ringThreadToStop) { logger.error(e.getMessage(), e); } } } logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper#ringThread stop"); } }); ringThread.setDaemon(true); ringThread.setName("xxl-job, admin JobScheduleHelper#ringThread"); ringThread.start(); } private void refreshNextValidTime(XxlJobInfo jobInfo, Date fromTime) throws ParseException { Date nextValidTime = new CronExpression(jobInfo.getJobCron()).getNextValidTimeAfter(fromTime); if (nextValidTime != null) { jobInfo.setTriggerLastTime(jobInfo.getTriggerNextTime()); jobInfo.setTriggerNextTime(nextValidTime.getTime()); } else { jobInfo.setTriggerStatus(0); jobInfo.setTriggerLastTime(0); jobInfo.setTriggerNextTime(0); } } private void pushTimeRing(int ringSecond, int jobId){ // push async ring List<Integer> ringItemData = ringData.get(ringSecond); if (ringItemData == null) { ringItemData = new ArrayList<Integer>(); ringData.put(ringSecond, ringItemData); } ringItemData.add(jobId); logger.debug(">>>>>>>>>>> xxl-job, schedule push time-ring : " + ringSecond + " = " + Arrays.asList(ringItemData) ); } public void toStop(){ // 1、stop schedule scheduleThreadToStop = true; try { TimeUnit.SECONDS.sleep(1); // wait } catch (InterruptedException e) { logger.error(e.getMessage(), e); } if (scheduleThread.getState() != Thread.State.TERMINATED){ // interrupt and wait scheduleThread.interrupt(); try { scheduleThread.join(); } catch (InterruptedException e) { logger.error(e.getMessage(), e); } } // if has ring data boolean hasRingData = false; if (!ringData.isEmpty()) { for (int second : ringData.keySet()) { List<Integer> tmpData = ringData.get(second); if (tmpData!=null && tmpData.size()>0) { hasRingData = true; break; } } } if (hasRingData) { try { TimeUnit.SECONDS.sleep(8); } catch (InterruptedException e) { logger.error(e.getMessage(), e); } } // stop ring (wait job-in-memory stop) ringThreadToStop = true; try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { logger.error(e.getMessage(), e); } if (ringThread.getState() != Thread.State.TERMINATED){ // interrupt and wait ringThread.interrupt(); try { ringThread.join(); } catch (InterruptedException e) { logger.error(e.getMessage(), e); } } logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper stop"); } }
3.3 任务执行时序图
3.4 任务执行核心代码解析
3.4.1 根据路由策略定位到执行器
任务执行：XxlJobTrigger
package com.xxl.job.admin.core.trigger; import com.xxl.job.admin.core.conf.XxlJobAdminConfig; import com.xxl.job.admin.core.scheduler.XxlJobScheduler; import com.xxl.job.admin.core.model.XxlJobGroup; import com.xxl.job.admin.core.model.XxlJobInfo; import com.xxl.job.admin.core.model.XxlJobLog; import com.xxl.job.admin.core.route.ExecutorRouteStrategyEnum; import com.xxl.job.admin.core.util.I18nUtil; import com.xxl.job.core.biz.ExecutorBiz; import com.xxl.job.core.biz.model.ReturnT; import com.xxl.job.core.biz.model.TriggerParam; import com.xxl.job.core.enums.ExecutorBlockStrategyEnum; import com.xxl.rpc.util.IpUtil; import com.xxl.rpc.util.ThrowableUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.Date; /** * xxl-job trigger * Created by xuxueli on 17/7/13. */ public class XxlJobTrigger { private static Logger logger = LoggerFactory.getLogger(XxlJobTrigger.class); /** * trigger job * * @param jobId * @param triggerType * @param failRetryCount * >=0: use this param * <0: use param from job info config * @param executorShardingParam * @param executorParam * null: use job param * not null: cover job param */ public static void trigger(int jobId, TriggerTypeEnum triggerType, int failRetryCount, String executorShardingParam, String executorParam) { // load data //通过jobId查询出任务信息 XxlJobInfo jobInfo = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().loadById(jobId); if (jobInfo == null) { logger.warn(">>>>>>>>>>>> trigger fail, jobId invalid，jobId={}", jobId); return; } if (executorParam != null) { jobInfo.setExecutorParam(executorParam); } //计算失败重试次数 int finalFailRetryCount = failRetryCount>=0?failRetryCount:jobInfo.getExecutorFailRetryCount(); XxlJobGroup group = XxlJobAdminConfig.getAdminConfig().getXxlJobGroupDao().load(jobInfo.getJobGroup()); // sharding param //处理分片参数 int[] shardingParam = null; if (executorShardingParam!=null){ String[] shardingArr = executorShardingParam.split("/"); if (shardingArr.length==2 && isNumeric(shardingArr[0]) && isNumeric(shardingArr[1])) { shardingParam = new int[2]; shardingParam[0] = Integer.valueOf(shardingArr[0]); shardingParam[1] = Integer.valueOf(shardingArr[1]); } } //处理路由策略 if (ExecutorRouteStrategyEnum.SHARDING_BROADCAST==ExecutorRouteStrategyEnum.match(jobInfo.getExecutorRouteStrategy(), null) && group.getRegistryList()!=null && !group.getRegistryList().isEmpty() && shardingParam==null) { //如果路由策略为分片广播，且执行器地址不为空，则遍历执行器地址进行广播触发任务调度 for (int i = 0; i < group.getRegistryList().size(); i++) { processTrigger(group, jobInfo, finalFailRetryCount, triggerType, i, group.getRegistryList().size()); } } else { //否则初始化分片参数 if (shardingParam == null) { shardingParam = new int[]{0, 1}; } processTrigger(group, jobInfo, finalFailRetryCount, triggerType, shardingParam[0], shardingParam[1]); } } private static boolean isNumeric(String str){ try { int result = Integer.valueOf(str); return true; } catch (NumberFormatException e) { return false; } } /** * @param group job group, registry list may be empty * @param jobInfo * @param finalFailRetryCount * @param triggerType * @param index sharding index * @param total sharding index */ private static void processTrigger(XxlJobGroup group, XxlJobInfo jobInfo, int finalFailRetryCount, TriggerTypeEnum triggerType, int index, int total){ // param //执行器阻塞策略：调度过于密集执行器来不及处理时的处理策略，策略包括：单机串行（默认）、丢弃后续调度、覆盖之前调度； ExecutorBlockStrategyEnum blockStrategy = ExecutorBlockStrategyEnum.match(jobInfo.getExecutorBlockStrategy(), ExecutorBlockStrategyEnum.SERIAL_EXECUTION); // block strategy //执行器路由策略：执行器集群部署时提供丰富的路由策略，包括：第一个、最后一个、轮询、随机、一致性HASH、最不经常使用、最近最久未使用、故障转移、忙碌转移等 ExecutorRouteStrategyEnum executorRouteStrategyEnum = ExecutorRouteStrategyEnum.match(jobInfo.getExecutorRouteStrategy(), null); // route strategy //处理分片参数：如果为分片广播则将index和total使用/进行拼接 String shardingParam = (ExecutorRouteStrategyEnum.SHARDING_BROADCAST==executorRouteStrategyEnum)?String.valueOf(index).concat("/").concat(String.valueOf(total)):null; // 1、save log-id //保存任务执行日志 XxlJobLog jobLog = new XxlJobLog(); jobLog.setJobGroup(jobInfo.getJobGroup()); jobLog.setJobId(jobInfo.getId()); jobLog.setTriggerTime(new Date()); XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().save(jobLog); logger.debug(">>>>>>>>>>> xxl-job trigger start, jobId:{}", jobLog.getId()); // 2、init trigger-param //初始化触发器参数 TriggerParam triggerParam = new TriggerParam(); triggerParam.setJobId(jobInfo.getId()); //取出执行器任务handler名称 triggerParam.setExecutorHandler(jobInfo.getExecutorHandler()); triggerParam.setExecutorParams(jobInfo.getExecutorParam()); triggerParam.setExecutorBlockStrategy(jobInfo.getExecutorBlockStrategy()); triggerParam.setExecutorTimeout(jobInfo.getExecutorTimeout()); triggerParam.setLogId(jobLog.getId()); triggerParam.setLogDateTime(jobLog.getTriggerTime().getTime()); triggerParam.setGlueType(jobInfo.getGlueType()); triggerParam.setGlueSource(jobInfo.getGlueSource()); triggerParam.setGlueUpdatetime(jobInfo.getGlueUpdatetime().getTime()); triggerParam.setBroadcastIndex(index); triggerParam.setBroadcastTotal(total); // 3、init address //初始化执行器地址信息 String address = null; ReturnT<String> routeAddressResult = null; if (group.getRegistryList()!=null && !group.getRegistryList().isEmpty()) { //如果执行器的注册地址不为空 if (ExecutorRouteStrategyEnum.SHARDING_BROADCAST == executorRouteStrategyEnum) { //如果路由策略是分片广播，取出index对应的执行器地址信息 if (index < group.getRegistryList().size()) { address = group.getRegistryList().get(index); } else { address = group.getRegistryList().get(0); } } else { //获取执行器路由地址信息 //⚠️⚠️️⚠️非常重要：此处使用了策略模式， 根据不同的策略 使用不同的实现类，从而选举出本次使用执行器集群地址列表中不同的地址信息，这是xxl-job实现任务路由的地方 routeAddressResult = executorRouteStrategyEnum.getRouter().route(triggerParam, group.getRegistryList()); if (routeAddressResult.getCode() == ReturnT.SUCCESS_CODE) { address = routeAddressResult.getContent(); } } } else { routeAddressResult = new ReturnT<String>(ReturnT.FAIL_CODE, I18nUtil.getString("jobconf_trigger_address_empty")); } // 4、trigger remote executor //触发远程执行器执行任务（执行器中的handler） ReturnT<String> triggerResult = null; if (address != null) { //这里是重点⚠️⚠️⚠️：启动执行器， 向执行器发送指令都是从这个方法中执行的 triggerResult = runExecutor(triggerParam, address); } else { triggerResult = new ReturnT<String>(ReturnT.FAIL_CODE, null); } // 5、collection trigger info StringBuffer triggerMsgSb = new StringBuffer(); triggerMsgSb.append(I18nUtil.getString("jobconf_trigger_type")).append("：").append(triggerType.getTitle()); triggerMsgSb.append("<br>").append(I18nUtil.getString("jobconf_trigger_admin_adress")).append("：").append(IpUtil.getIp()); triggerMsgSb.append("<br>").append(I18nUtil.getString("jobconf_trigger_exe_regtype")).append("：") .append( (group.getAddressType() == 0)?I18nUtil.getString("jobgroup_field_addressType_0"):I18nUtil.getString("jobgroup_field_addressType_1") ); triggerMsgSb.append("<br>").append(I18nUtil.getString("jobconf_trigger_exe_regaddress")).append("：").append(group.getRegistryList()); triggerMsgSb.append("<br>").append(I18nUtil.getString("jobinfo_field_executorRouteStrategy")).append("：").append(executorRouteStrategyEnum.getTitle()); if (shardingParam != null) { triggerMsgSb.append("("+shardingParam+")"); } triggerMsgSb.append("<br>").append(I18nUtil.getString("jobinfo_field_executorBlockStrategy")).append("：").append(blockStrategy.getTitle()); triggerMsgSb.append("<br>").append(I18nUtil.getString("jobinfo_field_timeout")).append("：").append(jobInfo.getExecutorTimeout()); triggerMsgSb.append("<br>").append(I18nUtil.getString("jobinfo_field_executorFailRetryCount")).append("：").append(finalFailRetryCount); triggerMsgSb.append("<br><br><span style=\"color:#00c0ef;\" > >>>>>>>>>>>"+ I18nUtil.getString("jobconf_trigger_run") +"<<<<<<<<<<< </span><br>") .append((routeAddressResult!=null&&routeAddressResult.getMsg()!=null)?routeAddressResult.getMsg()+"<br><br>":"").append(triggerResult.getMsg()!=null?triggerResult.getMsg():""); // 6、save log trigger-info //更新执行日志 jobLog.setExecutorAddress(address); jobLog.setExecutorHandler(jobInfo.getExecutorHandler()); jobLog.setExecutorParam(jobInfo.getExecutorParam()); jobLog.setExecutorShardingParam(shardingParam); jobLog.setExecutorFailRetryCount(finalFailRetryCount); //jobLog.setTriggerTime(); jobLog.setTriggerCode(triggerResult.getCode()); jobLog.setTriggerMsg(triggerMsgSb.toString()); XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().updateTriggerInfo(jobLog); logger.debug(">>>>>>>>>>> xxl-job trigger end, jobId:{}", jobLog.getId()); } /** * run executor * @param triggerParam * @param address * @return */ public static ReturnT<String> runExecutor(TriggerParam triggerParam, String address){ ReturnT<String> runResult = null; try { //这是重点：获取ExecutorBiz代理对象，先从缓存（内存map）中取，取不到new一个XxlRpcReferenceBean放进缓存map ExecutorBiz executorBiz = XxlJobScheduler.getExecutorBiz(address); // 这个run方法不会最终执行，仅仅只是为了触发代理对象的invoke方法，同时将目标的类型传送给服务端，因为在代理对象的invoke的方法里面没有执行目标对象的方法 runResult = executorBiz.run(triggerParam); } catch (Exception e) { logger.error(">>>>>>>>>>> xxl-job trigger error, please check if the executor[{}] is running.", address, e); runResult = new ReturnT<String>(ReturnT.FAIL_CODE, ThrowableUtil.toString(e)); } StringBuffer runResultSB = new StringBuffer(I18nUtil.getString("jobconf_trigger_run") + "："); runResultSB.append("<br>address：").append(address); runResultSB.append("<br>code：").append(runResult.getCode()); runResultSB.append("<br>msg：").append(runResult.getMsg()); runResult.setMsg(runResultSB.toString()); return runResult; } }
3.4.2 定位到任务执行Handler&线程
获得执行Handler&线程：ExecutorBizImpl
package com.xxl.job.core.biz.impl; import com.xxl.job.core.biz.ExecutorBiz; import com.xxl.job.core.biz.model.LogResult; import com.xxl.job.core.biz.model.ReturnT; import com.xxl.job.core.biz.model.TriggerParam; import com.xxl.job.core.enums.ExecutorBlockStrategyEnum; import com.xxl.job.core.executor.XxlJobExecutor; import com.xxl.job.core.glue.GlueFactory; import com.xxl.job.core.glue.GlueTypeEnum; import com.xxl.job.core.handler.IJobHandler; import com.xxl.job.core.handler.impl.GlueJobHandler; import com.xxl.job.core.handler.impl.ScriptJobHandler; import com.xxl.job.core.log.XxlJobFileAppender; import com.xxl.job.core.thread.JobThread; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.Date; /** * Created by xuxueli on 17/3/1. * @Description：阻塞处理策略，当执行器节点存在多个相同任务id的任务未执行完成，则需要基于阻塞策略对任务进行取舍： 串行策略：默认策略，任务进行排队、丢弃旧任务策略、丢弃新任务策略 */ public class ExecutorBizImpl implements ExecutorBiz { private static Logger logger = LoggerFactory.getLogger(ExecutorBizImpl.class); @Override public ReturnT<String> beat() { return ReturnT.SUCCESS; } @Override public ReturnT<String> idleBeat(int jobId) { // isRunningOrHasQueue boolean isRunningOrHasQueue = false; JobThread jobThread = XxlJobExecutor.loadJobThread(jobId); if (jobThread != null && jobThread.isRunningOrHasQueue()) { isRunningOrHasQueue = true; } if (isRunningOrHasQueue) { return new ReturnT<String>(ReturnT.FAIL_CODE, "job thread is running or has trigger queue."); } return ReturnT.SUCCESS; } @Override public ReturnT<String> kill(int jobId) { // kill handlerThread, and create new one JobThread jobThread = XxlJobExecutor.loadJobThread(jobId); if (jobThread != null) { XxlJobExecutor.removeJobThread(jobId, "scheduling center kill job."); return ReturnT.SUCCESS; } return new ReturnT<String>(ReturnT.SUCCESS_CODE, "job thread already killed."); } @Override public ReturnT<LogResult> log(long logDateTim, long logId, int fromLineNum) { // log filename: logPath/yyyy-MM-dd/9999.log String logFileName = XxlJobFileAppender.makeLogFileName(new Date(logDateTim), logId); LogResult logResult = XxlJobFileAppender.readLog(logFileName, fromLineNum); return new ReturnT<LogResult>(logResult); } @Override public ReturnT<String> run(TriggerParam triggerParam) { // load old：jobHandler + jobThread JobThread jobThread = XxlJobExecutor.loadJobThread(triggerParam.getJobId()); IJobHandler jobHandler = jobThread!=null?jobThread.getHandler():null; String removeOldReason = null; // valid：jobHandler + jobThread GlueTypeEnum glueTypeEnum = GlueTypeEnum.match(triggerParam.getGlueType()); if (GlueTypeEnum.BEAN == glueTypeEnum) { // new jobhandler IJobHandler newJobHandler = XxlJobExecutor.loadJobHandler(triggerParam.getExecutorHandler()); // valid old jobThread if (jobThread!=null && jobHandler != newJobHandler) { // change handler, need kill old thread removeOldReason = "change jobhandler or glue type, and terminate the old job thread."; jobThread = null; jobHandler = null; } // valid handler if (jobHandler == null) { jobHandler = newJobHandler; if (jobHandler == null) { return new ReturnT<String>(ReturnT.FAIL_CODE, "job handler [" + triggerParam.getExecutorHandler() + "] not found."); } } } else if (GlueTypeEnum.GLUE_GROOVY == glueTypeEnum) { // valid old jobThread if (jobThread != null && !(jobThread.getHandler() instanceof GlueJobHandler && ((GlueJobHandler) jobThread.getHandler()).getGlueUpdatetime()==triggerParam.getGlueUpdatetime() )) { // change handler or gluesource updated, need kill old thread removeOldReason = "change job source or glue type, and terminate the old job thread."; jobThread = null; jobHandler = null; } // valid handler if (jobHandler == null) { try { IJobHandler originJobHandler = GlueFactory.getInstance().loadNewInstance(triggerParam.getGlueSource()); jobHandler = new GlueJobHandler(originJobHandler, triggerParam.getGlueUpdatetime()); } catch (Exception e) { logger.error(e.getMessage(), e); return new ReturnT<String>(ReturnT.FAIL_CODE, e.getMessage()); } } } else if (glueTypeEnum!=null && glueTypeEnum.isScript()) { // valid old jobThread if (jobThread != null && !(jobThread.getHandler() instanceof ScriptJobHandler && ((ScriptJobHandler) jobThread.getHandler()).getGlueUpdatetime()==triggerParam.getGlueUpdatetime() )) { // change script or gluesource updated, need kill old thread removeOldReason = "change job source or glue type, and terminate the old job thread."; jobThread = null; jobHandler = null; } // valid handler if (jobHandler == null) { jobHandler = new ScriptJobHandler(triggerParam.getJobId(), triggerParam.getGlueUpdatetime(), triggerParam.getGlueSource(), GlueTypeEnum.match(triggerParam.getGlueType())); } } else { return new ReturnT<String>(ReturnT.FAIL_CODE, "glueType[" + triggerParam.getGlueType() + "] is not valid."); } // executor block strategy if (jobThread != null) { ExecutorBlockStrategyEnum blockStrategy = ExecutorBlockStrategyEnum.match(triggerParam.getExecutorBlockStrategy(), null); if (ExecutorBlockStrategyEnum.DISCARD_LATER == blockStrategy) { // discard when running if (jobThread.isRunningOrHasQueue()) { return new ReturnT<String>(ReturnT.FAIL_CODE, "block strategy effect："+ExecutorBlockStrategyEnum.DISCARD_LATER.getTitle()); } } else if (ExecutorBlockStrategyEnum.COVER_EARLY == blockStrategy) { // kill running jobThread if (jobThread.isRunningOrHasQueue()) { removeOldReason = "block strategy effect：" + ExecutorBlockStrategyEnum.COVER_EARLY.getTitle(); jobThread = null; } } else { // just queue trigger } } // replace thread (new or exists invalid) if (jobThread == null) { // 任务执行 jobThread = XxlJobExecutor.registJobThread(triggerParam.getJobId(), jobHandler, removeOldReason); } // push data to queue ReturnT<String> pushResult = jobThread.pushTriggerQueue(triggerParam); return pushResult; } }
3.4.3 任务执行线程
任务线程执行&放入回调队列：JobThread
package com.xxl.job.core.thread; import com.xxl.job.core.biz.model.HandleCallbackParam; import com.xxl.job.core.biz.model.ReturnT; import com.xxl.job.core.biz.model.TriggerParam; import com.xxl.job.core.executor.XxlJobExecutor; import com.xxl.job.core.handler.IJobHandler; import com.xxl.job.core.log.XxlJobFileAppender; import com.xxl.job.core.log.XxlJobLogger; import com.xxl.job.core.util.ShardingUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.PrintWriter; import java.io.StringWriter; import java.util.Collections; import java.util.Date; import java.util.HashSet; import java.util.Set; import java.util.concurrent.*; /** * handler thread * @author xuxueli 2016-1-16 19:52:47 */ public class JobThread extends Thread{ private static Logger logger = LoggerFactory.getLogger(JobThread.class); private int jobId; private IJobHandler handler; private LinkedBlockingQueue<TriggerParam> triggerQueue; private Set<Long> triggerLogIdSet; // avoid repeat trigger for the same TRIGGER_LOG_ID private volatile boolean toStop = false; private String stopReason; private boolean running = false; // if running job private int idleTimes = 0; // idel times public JobThread(int jobId, IJobHandler handler) { this.jobId = jobId; this.handler = handler; this.triggerQueue = new LinkedBlockingQueue<TriggerParam>(); this.triggerLogIdSet = Collections.synchronizedSet(new HashSet<Long>()); } public IJobHandler getHandler() { return handler; } /** * new trigger to queue * * @param triggerParam * @return */ public ReturnT<String> pushTriggerQueue(TriggerParam triggerParam) { // avoid repeat if (triggerLogIdSet.contains(triggerParam.getLogId())) { logger.info(">>>>>>>>>>> repeate trigger job, logId:{}", triggerParam.getLogId()); return new ReturnT<String>(ReturnT.FAIL_CODE, "repeate trigger job, logId:" + triggerParam.getLogId()); } triggerLogIdSet.add(triggerParam.getLogId()); triggerQueue.add(triggerParam); return ReturnT.SUCCESS; } /** * kill job thread * * @param stopReason */ public void toStop(String stopReason) { /** * Thread.interrupt只支持终止线程的阻塞状态(wait、join、sleep)， * 在阻塞出抛出InterruptedException异常,但是并不会终止运行的线程本身； * 所以需要注意，此处彻底销毁本线程，需要通过共享变量方式； */ this.toStop = true; this.stopReason = stopReason; } /** * is running job * @return */ public boolean isRunningOrHasQueue() { return running || triggerQueue.size()>0; } @Override public void run() { // init try { handler.init(); } catch (Throwable e) { logger.error(e.getMessage(), e); } // execute while(!toStop){ running = false; idleTimes++; TriggerParam triggerParam = null; ReturnT<String> executeResult = null; try { // to check toStop signal, we need cycle, so wo cannot use queue.take(), instand of poll(timeout) triggerParam = triggerQueue.poll(3L, TimeUnit.SECONDS); if (triggerParam!=null) { running = true; idleTimes = 0; triggerLogIdSet.remove(triggerParam.getLogId()); // log filename, like "logPath/yyyy-MM-dd/9999.log" String logFileName = XxlJobFileAppender.makeLogFileName(new Date(triggerParam.getLogDateTime()), triggerParam.getLogId()); XxlJobFileAppender.contextHolder.set(logFileName); ShardingUtil.setShardingVo(new ShardingUtil.ShardingVO(triggerParam.getBroadcastIndex(), triggerParam.getBroadcastTotal())); // execute XxlJobLogger.log("<br>----------- xxl-job job execute start -----------<br>----------- Param:" + triggerParam.getExecutorParams()); if (triggerParam.getExecutorTimeout() > 0) { // limit timeout Thread futureThread = null; try { final TriggerParam triggerParamTmp = triggerParam; FutureTask<ReturnT<String>> futureTask = new FutureTask<ReturnT<String>>(new Callable<ReturnT<String>>() { @Override public ReturnT<String> call() throws Exception { return handler.execute(triggerParamTmp.getExecutorParams()); } }); futureThread = new Thread(futureTask); futureThread.start(); executeResult = futureTask.get(triggerParam.getExecutorTimeout(), TimeUnit.SECONDS); } catch (TimeoutException e) { XxlJobLogger.log("<br>----------- xxl-job job execute timeout"); XxlJobLogger.log(e); executeResult = new ReturnT<String>(IJobHandler.FAIL_TIMEOUT.getCode(), "job execute timeout "); } finally { futureThread.interrupt(); } } else { // just execute executeResult = handler.execute(triggerParam.getExecutorParams()); } if (executeResult == null) { executeResult = IJobHandler.FAIL; } else { executeResult.setMsg( (executeResult!=null&&executeResult.getMsg()!=null&&executeResult.getMsg().length()>50000) ?executeResult.getMsg().substring(0, 50000).concat("...") :executeResult.getMsg()); executeResult.setContent(null); // limit obj size } XxlJobLogger.log("<br>----------- xxl-job job execute end(finish) -----------<br>----------- ReturnT:" + executeResult); } else { if (idleTimes > 30) { if(triggerQueue.size() == 0) { // avoid concurrent trigger causes jobId-lost XxlJobExecutor.removeJobThread(jobId, "excutor idel times over limit."); } } } } catch (Throwable e) { if (toStop) { XxlJobLogger.log("<br>----------- JobThread toStop, stopReason:" + stopReason); } StringWriter stringWriter = new StringWriter(); e.printStackTrace(new PrintWriter(stringWriter)); String errorMsg = stringWriter.toString(); executeResult = new ReturnT<String>(ReturnT.FAIL_CODE, errorMsg); XxlJobLogger.log("<br>----------- JobThread Exception:" + errorMsg + "<br>----------- xxl-job job execute end(error) -----------"); } finally { if(triggerParam != null) { // callback handler info if (!toStop) { // commonm TriggerCallbackThread.pushCallBack(new HandleCallbackParam(triggerParam.getLogId(), triggerParam.getLogDateTime(), executeResult)); } else { // is killed ReturnT<String> stopResult = new ReturnT<String>(ReturnT.FAIL_CODE, stopReason + " [job running, killed]"); TriggerCallbackThread.pushCallBack(new HandleCallbackParam(triggerParam.getLogId(), triggerParam.getLogDateTime(), stopResult)); } } } } // callback trigger request in queue while(triggerQueue !=null && triggerQueue.size()>0){ TriggerParam triggerParam = triggerQueue.poll(); if (triggerParam!=null) { // is killed ReturnT<String> stopResult = new ReturnT<String>(ReturnT.FAIL_CODE, stopReason + " [job not executed, in the job queue, killed.]"); TriggerCallbackThread.pushCallBack(new HandleCallbackParam(triggerParam.getLogId(), triggerParam.getLogDateTime(), stopResult)); } } // destroy try { handler.destroy(); } catch (Throwable e) { logger.error(e.getMessage(), e); } logger.info(">>>>>>>>>>> xxl-job JobThread stoped, hashCode:{}", Thread.currentThread()); } }
四、执行器
4.1 启动过程时序图
4.2 启动过程核心代码解析
入口类：XxlJobSpringExecutor（PS：继承XxlJobExecutor，所有的任务触发最终都是通过这个类）
4.2.1 启动初始化
执行器启动初始化：XxlJobSpringExecutor
package com.xxl.job.core.executor.impl; import com.xxl.job.core.biz.model.ReturnT; import com.xxl.job.core.executor.XxlJobExecutor; import com.xxl.job.core.glue.GlueFactory; import com.xxl.job.core.handler.IJobHandler; import com.xxl.job.core.handler.annotation.JobHandler; import com.xxl.job.core.handler.annotation.XxlJob; import com.xxl.job.core.handler.impl.MethodJobHandler; import org.springframework.beans.BeansException; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; import org.springframework.core.annotation.AnnotationUtils; import java.lang.reflect.Method; import java.util.Map; /** * xxl-job executor (for spring) * * @author xuxueli 2018-11-01 09:24:52 * @Description：执行器启动过程 */ public class XxlJobSpringExecutor extends XxlJobExecutor implements ApplicationContextAware, InitializingBean, DisposableBean { // start @Override public void afterPropertiesSet() throws Exception { // init JobHandler Repository /** * 老版本：获取使用了JobHandler注解的bean，即任务处理器，并注册到jobHandlerRepository（ConcurrentMap）缓存中 */ initJobHandlerRepository(applicationContext); // init JobHandler Repository (for method) /** * 新版本：获取使用了XxlJob注解的method，即任务处理器，并注册到jobHandlerRepository（ConcurrentMap）缓存中 */ initJobHandlerMethodRepository(applicationContext); // refresh GlueFactory GlueFactory.refreshInstance(1); // super start /** * 核心启动 */ super.start(); } // destroy @Override public void destroy() { super.destroy(); } private void initJobHandlerRepository(ApplicationContext applicationContext) { if (applicationContext == null) { return; } // init job handler action Map<String, Object> serviceBeanMap = applicationContext.getBeansWithAnnotation(JobHandler.class); if (serviceBeanMap != null && serviceBeanMap.size() > 0) { for (Object serviceBean : serviceBeanMap.values()) { if (serviceBean instanceof IJobHandler) { String name = serviceBean.getClass().getAnnotation(JobHandler.class).value(); IJobHandler handler = (IJobHandler) serviceBean; if (loadJobHandler(name) != null) { throw new RuntimeException("xxl-job jobhandler[" + name + "] naming conflicts."); } registJobHandler(name, handler); } } } } private void initJobHandlerMethodRepository(ApplicationContext applicationContext) { if (applicationContext == null) { return; } // init job handler from method String[] beanDefinitionNames = applicationContext.getBeanDefinitionNames(); if (beanDefinitionNames!=null && beanDefinitionNames.length>0) { for (String beanDefinitionName : beanDefinitionNames) { Object bean = applicationContext.getBean(beanDefinitionName); Method[] methods = bean.getClass().getDeclaredMethods(); for (Method method: methods) { XxlJob xxlJob = AnnotationUtils.findAnnotation(method, XxlJob.class); if (xxlJob != null) { // name String name = xxlJob.value(); if (name.trim().length() == 0) { throw new RuntimeException("xxl-job method-jobhandler name invalid, for[" + bean.getClass() + "#"+ method.getName() +"] ."); } if (loadJobHandler(name) != null) { throw new RuntimeException("xxl-job jobhandler[" + name + "] naming conflicts."); } // execute method if (!(method.getParameterTypes()!=null && method.getParameterTypes().length==1 && method.getParameterTypes()[0].isAssignableFrom(String.class))) { throw new RuntimeException("xxl-job method-jobhandler param-classtype invalid, for[" + bean.getClass() + "#"+ method.getName() +"] , " + "The correct method format like \" public ReturnT<String> execute(String param) \" ."); } if (!method.getReturnType().isAssignableFrom(ReturnT.class)) { throw new RuntimeException("xxl-job method-jobhandler return-classtype invalid, for[" + bean.getClass() + "#"+ method.getName() +"] , " + "The correct method format like \" public ReturnT<String> execute(String param) \" ."); } method.setAccessible(true); // init and destory Method initMethod = null; Method destroyMethod = null; if(xxlJob.init().trim().length() > 0) { try { initMethod = bean.getClass().getDeclaredMethod(xxlJob.init()); initMethod.setAccessible(true); } catch (NoSuchMethodException e) { throw new RuntimeException("xxl-job method-jobhandler initMethod invalid, for[" + bean.getClass() + "#"+ method.getName() +"] ."); } } if(xxlJob.destroy().trim().length() > 0) { try { destroyMethod = bean.getClass().getDeclaredMethod(xxlJob.destroy()); destroyMethod.setAccessible(true); } catch (NoSuchMethodException e) { throw new RuntimeException("xxl-job method-jobhandler destroyMethod invalid, for[" + bean.getClass() + "#"+ method.getName() +"] ."); } } // registry jobhandler registJobHandler(name, new MethodJobHandler(bean, method, initMethod, destroyMethod)); } } } } } // ---------------------- applicationContext ---------------------- private static ApplicationContext applicationContext; @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { this.applicationContext = applicationContext; } public static ApplicationContext getApplicationContext() { return applicationContext; } }
4.2.2 执行器初始化
执行器相关信息准备：XxlJobExecutor
package com.xxl.job.core.executor; import com.xxl.job.core.biz.AdminBiz; import com.xxl.job.core.biz.ExecutorBiz; import com.xxl.job.core.biz.client.AdminBizClient; import com.xxl.job.core.biz.impl.ExecutorBizImpl; import com.xxl.job.core.handler.IJobHandler; import com.xxl.job.core.log.XxlJobFileAppender; import com.xxl.job.core.thread.ExecutorRegistryThread; import com.xxl.job.core.thread.JobLogFileCleanThread; import com.xxl.job.core.thread.JobThread; import com.xxl.job.core.thread.TriggerCallbackThread; import com.xxl.rpc.registry.ServiceRegistry; import com.xxl.rpc.remoting.net.impl.netty_http.server.NettyHttpServer; import com.xxl.rpc.remoting.provider.XxlRpcProviderFactory; import com.xxl.rpc.serialize.Serializer; import com.xxl.rpc.serialize.impl.HessianSerializer; import com.xxl.rpc.util.IpUtil; import com.xxl.rpc.util.NetUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.*; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; /** * Created by xuxueli on 2016/3/2 21:14. * @Description：任务执行器 */ public class XxlJobExecutor { private static final Logger logger = LoggerFactory.getLogger(XxlJobExecutor.class); // ---------------------- param ---------------------- private String adminAddresses; private String appName; private String ip; private int port; private String accessToken; private String logPath; private int logRetentionDays; public void setAdminAddresses(String adminAddresses) { this.adminAddresses = adminAddresses; } public void setAppName(String appName) { this.appName = appName; } public void setIp(String ip) { this.ip = ip; } public void setPort(int port) { this.port = port; } public void setAccessToken(String accessToken) { this.accessToken = accessToken; } public void setLogPath(String logPath) { this.logPath = logPath; } public void setLogRetentionDays(int logRetentionDays) { this.logRetentionDays = logRetentionDays; } // ---------------------- start + stop ---------------------- public void start() throws Exception { // init logpath /** * 主要作用：初始化日志路径 */ XxlJobFileAppender.initLogPath(logPath); // init invoker, admin-client /** * 主要作用：初始化注册中心列表 （把注册地址放到 List） * 1、初始化执行器与调度中心建立通信的地址 */ initAdminBizList(adminAddresses, accessToken); // init JobLogFileCleanThread /** * 主要作用： * 1、启动日志文件清理线程 （一天清理一次） * 2、每天清理一次过期日志，配置参数必须大于3才有效 */ JobLogFileCleanThread.getInstance().start(logRetentionDays); // init TriggerCallbackThread /** * 主要作用：开启触发器回调线程 */ TriggerCallbackThread.getInstance().start(); // init executor-server /** * 主要端口：指定端口 */ port = port>0?port: NetUtil.findAvailablePort(9999); /** * 主要端口：指定IP */ ip = (ip!=null&&ip.trim().length()>0)?ip: IpUtil.getIp(); /** * 主要作用：启动注册中心的RPC服务，使得执行器项目可以通过RPC进行注册和心跳检测 * 1、将执行器注册到调度中心 30秒一次 */ initRpcProvider(ip, port, appName, accessToken); } public void destroy(){ // destory executor-server stopRpcProvider(); // destory jobThreadRepository if (jobThreadRepository.size() > 0) { for (Map.Entry<Integer, JobThread> item: jobThreadRepository.entrySet()) { removeJobThread(item.getKey(), "web container destroy and kill the job."); } jobThreadRepository.clear(); } jobHandlerRepository.clear(); // destory JobLogFileCleanThread JobLogFileCleanThread.getInstance().toStop(); // destory TriggerCallbackThread TriggerCallbackThread.getInstance().toStop(); } // ---------------------- admin-client (rpc invoker) ---------------------- private static List<AdminBiz> adminBizList; private static Serializer serializer = new HessianSerializer(); private void initAdminBizList(String adminAddresses, String accessToken) throws Exception { if (adminAddresses!=null && adminAddresses.trim().length()>0) { for (String address: adminAddresses.trim().split(",")) { if (address!=null && address.trim().length()>0) { AdminBiz adminBiz = new AdminBizClient(address.trim(), accessToken); if (adminBizList == null) { adminBizList = new ArrayList<AdminBiz>(); } adminBizList.add(adminBiz); } } } } public static List<AdminBiz> getAdminBizList(){ return adminBizList; } public static Serializer getSerializer() { return serializer; } // ---------------------- executor-server (rpc provider) ---------------------- private XxlRpcProviderFactory xxlRpcProviderFactory = null; private void initRpcProvider(String ip, int port, String appName, String accessToken) throws Exception { // init, provider factory String address = IpUtil.getIpPort(ip, port); Map<String, String> serviceRegistryParam = new HashMap<String, String>(); serviceRegistryParam.put("appName", appName); serviceRegistryParam.put("address", address); //初始化RPC配置 xxlRpcProviderFactory = new XxlRpcProviderFactory(); xxlRpcProviderFactory.setServer(NettyHttpServer.class); xxlRpcProviderFactory.setSerializer(HessianSerializer.class); xxlRpcProviderFactory.setCorePoolSize(20); xxlRpcProviderFactory.setMaxPoolSize(200); xxlRpcProviderFactory.setIp(ip); xxlRpcProviderFactory.setPort(port); xxlRpcProviderFactory.setAccessToken(accessToken); xxlRpcProviderFactory.setServiceRegistry(ExecutorServiceRegistry.class); xxlRpcProviderFactory.setServiceRegistryParam(serviceRegistryParam); // add services //给xxlRpcProviderFactory加入服务 xxlRpcProviderFactory.addService(ExecutorBiz.class.getName(), null, new ExecutorBizImpl()); // start /** * 开启注册 */ xxlRpcProviderFactory.start(); } public static class ExecutorServiceRegistry extends ServiceRegistry { @Override public void start(Map<String, String> param) { // start registry ExecutorRegistryThread.getInstance().start(param.get("appName"), param.get("address")); } @Override public void stop() { // stop registry ExecutorRegistryThread.getInstance().toStop(); } @Override public boolean registry(Set<String> keys, String value) { return false; } @Override public boolean remove(Set<String> keys, String value) { return false; } @Override public Map<String, TreeSet<String>> discovery(Set<String> keys) { return null; } @Override public TreeSet<String> discovery(String key) { return null; } } private void stopRpcProvider() { // stop provider factory try { xxlRpcProviderFactory.stop(); } catch (Exception e) { logger.error(e.getMessage(), e); } } /** * 一个JobThread对应一个JobHandler */ // ---------------------- job handler repository（key为XxlJob注解中的bean名称，value为JobHandler） ---------------------- private static ConcurrentMap<String, IJobHandler> jobHandlerRepository = new ConcurrentHashMap<String, IJobHandler>(); public static IJobHandler registJobHandler(String name, IJobHandler jobHandler){ logger.info(">>>>>>>>>>> xxl-job register jobhandler success, name:{}, jobHandler:{}", name, jobHandler); return jobHandlerRepository.put(name, jobHandler); } public static IJobHandler loadJobHandler(String name){ return jobHandlerRepository.get(name); } // ---------------------- job thread repository（key为任务id，即jobId，value为JobThread对象）---------------------- private static ConcurrentMap<Integer, JobThread> jobThreadRepository = new ConcurrentHashMap<Integer, JobThread>(); public static JobThread registJobThread(int jobId, IJobHandler handler, String removeOldReason){ JobThread newJobThread = new JobThread(jobId, handler); newJobThread.start(); logger.info(">>>>>>>>>>> xxl-job regist JobThread success, jobId:{}, handler:{}", new Object[]{jobId, handler}); JobThread oldJobThread = jobThreadRepository.put(jobId, newJobThread); // putIfAbsent | oh my god, map's put method return the old value!!! if (oldJobThread != null) { oldJobThread.toStop(removeOldReason); oldJobThread.interrupt(); } return newJobThread; } public static void removeJobThread(int jobId, String removeOldReason){ JobThread oldJobThread = jobThreadRepository.remove(jobId); if (oldJobThread != null) { oldJobThread.toStop(removeOldReason); oldJobThread.interrupt(); } } public static JobThread loadJobThread(int jobId){ JobThread jobThread = jobThreadRepository.get(jobId); return jobThread; } }
4.2.3 任务执行回调
任务执行后结果回调：TriggerCallbackThread
package com.xxl.job.core.thread; import com.xxl.job.core.biz.AdminBiz; import com.xxl.job.core.biz.model.HandleCallbackParam; import com.xxl.job.core.biz.model.ReturnT; import com.xxl.job.core.enums.RegistryConfig; import com.xxl.job.core.executor.XxlJobExecutor; import com.xxl.job.core.log.XxlJobFileAppender; import com.xxl.job.core.log.XxlJobLogger; import com.xxl.job.core.util.FileUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.File; import java.util.ArrayList; import java.util.Date; import java.util.List; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; /** * Created by xuxueli on 16/7/22. */ public class TriggerCallbackThread { private static Logger logger = LoggerFactory.getLogger(TriggerCallbackThread.class); private static TriggerCallbackThread instance = new TriggerCallbackThread(); public static TriggerCallbackThread getInstance(){ return instance; } /** * job results callback queue */ private LinkedBlockingQueue<HandleCallbackParam> callBackQueue = new LinkedBlockingQueue<HandleCallbackParam>(); public static void pushCallBack(HandleCallbackParam callback){ getInstance().callBackQueue.add(callback); logger.debug(">>>>>>>>>>> xxl-job, push callback request, logId:{}", callback.getLogId()); } /** * callback thread */ private Thread triggerCallbackThread; private Thread triggerRetryCallbackThread; private volatile boolean toStop = false; public void start() { // valid if (XxlJobExecutor.getAdminBizList() == null) { logger.warn(">>>>>>>>>>> xxl-job, executor callback config fail, adminAddresses is null."); return; } // callback triggerCallbackThread = new Thread(new Runnable() { @Override public void run() { // normal callback while(!toStop){ try { HandleCallbackParam callback = getInstance().callBackQueue.take(); if (callback != null) { // callback list param List<HandleCallbackParam> callbackParamList = new ArrayList<HandleCallbackParam>(); int drainToNum = getInstance().callBackQueue.drainTo(callbackParamList); callbackParamList.add(callback); // callback, will retry if error if (callbackParamList!=null && callbackParamList.size()>0) { doCallback(callbackParamList); } } } catch (Exception e) { if (!toStop) { logger.error(e.getMessage(), e); } } } // last callback try { List<HandleCallbackParam> callbackParamList = new ArrayList<HandleCallbackParam>(); int drainToNum = getInstance().callBackQueue.drainTo(callbackParamList); if (callbackParamList!=null && callbackParamList.size()>0) { doCallback(callbackParamList); } } catch (Exception e) { if (!toStop) { logger.error(e.getMessage(), e); } } logger.info(">>>>>>>>>>> xxl-job, executor callback thread destory."); } }); triggerCallbackThread.setDaemon(true); triggerCallbackThread.setName("xxl-job, executor TriggerCallbackThread"); triggerCallbackThread.start(); // retry triggerRetryCallbackThread = new Thread(new Runnable() { @Override public void run() { while(!toStop){ try { retryFailCallbackFile(); } catch (Exception e) { if (!toStop) { logger.error(e.getMessage(), e); } } try { TimeUnit.SECONDS.sleep(RegistryConfig.BEAT_TIMEOUT); } catch (InterruptedException e) { if (!toStop) { logger.error(e.getMessage(), e); } } } logger.info(">>>>>>>>>>> xxl-job, executor retry callback thread destory."); } }); triggerRetryCallbackThread.setDaemon(true); triggerRetryCallbackThread.start(); } public void toStop(){ toStop = true; // stop callback, interrupt and wait if (triggerCallbackThread != null) { // support empty admin address triggerCallbackThread.interrupt(); try { triggerCallbackThread.join(); } catch (InterruptedException e) { logger.error(e.getMessage(), e); } } // stop retry, interrupt and wait if (triggerRetryCallbackThread != null) { triggerRetryCallbackThread.interrupt(); try { triggerRetryCallbackThread.join(); } catch (InterruptedException e) { logger.error(e.getMessage(), e); } } } /** * do callback, will retry if error * @param callbackParamList */ private void doCallback(List<HandleCallbackParam> callbackParamList){ boolean callbackRet = false; // callback, will retry if error for (AdminBiz adminBiz: XxlJobExecutor.getAdminBizList()) { try { ReturnT<String> callbackResult = adminBiz.callback(callbackParamList); if (callbackResult!=null && ReturnT.SUCCESS_CODE == callbackResult.getCode()) { callbackLog(callbackParamList, "<br>----------- xxl-job job callback finish."); callbackRet = true; break; } else { callbackLog(callbackParamList, "<br>----------- xxl-job job callback fail, callbackResult:" + callbackResult); } } catch (Exception e) { callbackLog(callbackParamList, "<br>----------- xxl-job job callback error, errorMsg:" + e.getMessage()); } } if (!callbackRet) { appendFailCallbackFile(callbackParamList); } } /** * callback log */ private void callbackLog(List<HandleCallbackParam> callbackParamList, String logContent){ for (HandleCallbackParam callbackParam: callbackParamList) { String logFileName = XxlJobFileAppender.makeLogFileName(new Date(callbackParam.getLogDateTim()), callbackParam.getLogId()); XxlJobFileAppender.contextHolder.set(logFileName); XxlJobLogger.log(logContent); } } // ---------------------- fail-callback file ---------------------- private static String failCallbackFilePath = XxlJobFileAppender.getLogPath().concat(File.separator).concat("callbacklog").concat(File.separator); private static String failCallbackFileName = failCallbackFilePath.concat("xxl-job-callback-{x}").concat(".log"); private void appendFailCallbackFile(List<HandleCallbackParam> callbackParamList){ // valid if (callbackParamList==null || callbackParamList.size()==0) { return; } // append file byte[] callbackParamList_bytes = XxlJobExecutor.getSerializer().serialize(callbackParamList); File callbackLogFile = new File(failCallbackFileName.replace("{x}", String.valueOf(System.currentTimeMillis()))); if (callbackLogFile.exists()) { for (int i = 0; i < 100; i++) { callbackLogFile = new File(failCallbackFileName.replace("{x}", String.valueOf(System.currentTimeMillis()).concat("-").concat(String.valueOf(i)) )); if (!callbackLogFile.exists()) { break; } } } FileUtil.writeFileContent(callbackLogFile, callbackParamList_bytes); } private void retryFailCallbackFile(){ // valid File callbackLogPath = new File(failCallbackFilePath); if (!callbackLogPath.exists()) { return; } if (callbackLogPath.isFile()) { callbackLogPath.delete(); } if (!(callbackLogPath.isDirectory() && callbackLogPath.list()!=null && callbackLogPath.list().length>0)) { return; } // load and clear file, retry for (File callbaclLogFile: callbackLogPath.listFiles()) { byte[] callbackParamList_bytes = FileUtil.readFileContent(callbaclLogFile); List<HandleCallbackParam> callbackParamList = (List<HandleCallbackParam>) XxlJobExecutor.getSerializer().deserialize(callbackParamList_bytes, HandleCallbackParam.class); callbaclLogFile.delete(); doCallback(callbackParamList); } } }
4.3 执行器注册到调度中心时序图
4.4 执行器注册到调度中心核心代码解析
4.4.1 执行注册（30秒一次）
将执行器注册到调度中心：ExecutorRegistryThread
package com.xxl.job.core.thread; import com.xxl.job.core.biz.AdminBiz; import com.xxl.job.core.biz.model.RegistryParam; import com.xxl.job.core.biz.model.ReturnT; import com.xxl.job.core.enums.RegistryConfig; import com.xxl.job.core.executor.XxlJobExecutor; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.concurrent.TimeUnit; /** * Created by xuxueli on 17/3/2. * @Description：将执行器注册到调度中心 */ public class ExecutorRegistryThread { private static Logger logger = LoggerFactory.getLogger(ExecutorRegistryThread.class); private static ExecutorRegistryThread instance = new ExecutorRegistryThread(); public static ExecutorRegistryThread getInstance(){ return instance; } private Thread registryThread; private volatile boolean toStop = false; public void start(final String appName, final String address){ // valid if (appName==null || appName.trim().length()==0) { logger.warn(">>>>>>>>>>> xxl-job, executor registry config fail, appName is null."); return; } if (XxlJobExecutor.getAdminBizList() == null) { logger.warn(">>>>>>>>>>> xxl-job, executor registry config fail, adminAddresses is null."); return; } registryThread = new Thread(new Runnable() { @Override public void run() { // registry while (!toStop) { try { RegistryParam registryParam = new RegistryParam(RegistryConfig.RegistType.EXECUTOR.name(), appName, address); for (AdminBiz adminBiz: XxlJobExecutor.getAdminBizList()) { try { ReturnT<String> registryResult = adminBiz.registry(registryParam); if (registryResult!=null && ReturnT.SUCCESS_CODE == registryResult.getCode()) { registryResult = ReturnT.SUCCESS; logger.debug(">>>>>>>>>>> xxl-job registry success, registryParam:{}, registryResult:{}", new Object[]{registryParam, registryResult}); break; } else { logger.info(">>>>>>>>>>> xxl-job registry fail, registryParam:{}, registryResult:{}", new Object[]{registryParam, registryResult}); } } catch (Exception e) { logger.info(">>>>>>>>>>> xxl-job registry error, registryParam:{}", registryParam, e); } } } catch (Exception e) { if (!toStop) { logger.error(e.getMessage(), e); } } try { if (!toStop) { TimeUnit.SECONDS.sleep(RegistryConfig.BEAT_TIMEOUT); } } catch (InterruptedException e) { if (!toStop) { logger.warn(">>>>>>>>>>> xxl-job, executor registry thread interrupted, error msg:{}", e.getMessage()); } } } // registry remove try { RegistryParam registryParam = new RegistryParam(RegistryConfig.RegistType.EXECUTOR.name(), appName, address); for (AdminBiz adminBiz: XxlJobExecutor.getAdminBizList()) { try { ReturnT<String> registryResult = adminBiz.registryRemove(registryParam); if (registryResult!=null && ReturnT.SUCCESS_CODE == registryResult.getCode()) { registryResult = ReturnT.SUCCESS; logger.info(">>>>>>>>>>> xxl-job registry-remove success, registryParam:{}, registryResult:{}", new Object[]{registryParam, registryResult}); break; } else { logger.info(">>>>>>>>>>> xxl-job registry-remove fail, registryParam:{}, registryResult:{}", new Object[]{registryParam, registryResult}); } } catch (Exception e) { if (!toStop) { logger.info(">>>>>>>>>>> xxl-job registry-remove error, registryParam:{}", registryParam, e); } } } } catch (Exception e) { if (!toStop) { logger.error(e.getMessage(), e); } } logger.info(">>>>>>>>>>> xxl-job, executor registry thread destory."); } }); registryThread.setDaemon(true); registryThread.setName("xxl-job, executor ExecutorRegistryThread"); registryThread.start(); } public void toStop() { toStop = true; // interrupt and wait registryThread.interrupt(); try { registryThread.join(); } catch (InterruptedException e) { logger.error(e.getMessage(), e); } } }
旨在为数千万中国开发者提供一个无缝且高效的云端环境，以支持学习、使用和贡献开源项目。
更多推荐
4
0- 0
已为社区贡献2条内容
所有评论(0)