XXL-JOB任务执行的过程源码分析

2024-12-01 技术教程

xxl-job介绍xxl-job是一个分布式任务调度系统，基于quartz实现调度器。

1、quartz是基于数据库for update实现锁，来保证同一个任务同一时间只会执行一次。2、最新版本的xxl-job已经摒弃了quartz.xxl-job核心模块1、调度中心，也就是任务的管理系统2、执行器，任务真正的执行服务,一般是分布式的服务。任务执行过程1、调度中心1.1 点击执行的时候，入口JobInfoController.triggerJob方法：

进入JobTriggerPoolHelper.trigger方法，调用了JobTriggerPoolHelper.addTrigger方法，那么看一下addTrigger方法：

/** * add trigger */ public void addTrigger(final int jobId, final TriggerTypeEnum triggerType, final int failRetryCount, final String executorShardingParam, final String executorParam) { // choose thread pool //这里区分了快慢线程池，1分钟内超过500毫秒的请求大于10次，放入慢线程池处理 //快线程池默认8个核心线程，最大线程200，任务队列1000 //慢线程池默认0个核心线程，最大线程100，任务队列2000 ThreadPoolExecutor triggerPool_ = fastTriggerPool; AtomicInteger jobTimeoutCount = jobTimeoutCountMap.get(jobId); if (jobTimeoutCount!=null && jobTimeoutCount.get() > 10) { // job-timeout 10 times in 1 min triggerPool_ = slowTriggerPool; } // trigger triggerPool_.execute(new Runnable() { @Override public void run() { long start = System.currentTimeMillis(); try { //触发执行任务 XxlJobTrigger.trigger(jobId, triggerType, failRetryCount, executorShardingParam, executorParam); } catch (Exception e) { logger.error(e.getMessage(), e); } finally { // check timeout-count-map long minTim_now = System.currentTimeMillis()/60000; if (minTim != minTim_now) { minTim = minTim_now; jobTimeoutCountMap.clear(); } // incr timeout-count-map long cost = System.currentTimeMillis()-start; if (cost > 500) { //耗时超过500ms就计算为慢请求，加入慢线程池。 AtomicInteger timeoutCount = jobTimeoutCountMap.put(jobId, new AtomicInteger(1)); if (timeoutCount != null) { timeoutCount.incrementAndGet(); } } } } }); }

进入执行任务：XxlJobTrigger.trigger

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

进入processTrigger方法，组装任务参数，选择路由和阻塞策略

private static void processTrigger(XxlJobGroup group, XxlJobInfo jobInfo, int finalFailRetryCount, TriggerTypeEnum triggerType, int index, int total){ //阻塞策略 ExecutorBlockStrategyEnum blockStrategy = ExecutorBlockStrategyEnum.match(jobInfo.getExecutorBlockStrategy(), ExecutorBlockStrategyEnum.SERIAL_EXECUTION); // block strategy //路由策略，官方一共10中路由策略 ExecutorRouteStrategyEnum executorRouteStrategyEnum = ExecutorRouteStrategyEnum.match(jobInfo.getExecutorRouteStrategy(), null); // route strategy 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 初始化任务参数 //这里要增加2个参数; // 1、固定IP， // 2、任务为空时，默认轮训队列次数 TriggerParam triggerParam = new TriggerParam(); triggerParam.setJobId(jobInfo.getId()); triggerParam.setExecutorHandler(jobInfo.getExecutorHandler()); triggerParam.setExecutorParams(jobInfo.getExecutorParam()); triggerParam.setExecutorBlockStrategy(jobInfo.getExecutorBlockStrategy()); triggerParam.setExecutorTimeout(jobInfo.getExecutorTimeout()); triggerParam.setLogId(jobLog.getId()); triggerParam.setLogDateTim(jobLog.getTriggerTime().getTime()); triggerParam.setGlueType(jobInfo.getGlueType()); triggerParam.setGlueSource(jobInfo.getGlueSource()); triggerParam.setGlueUpdatetime(jobInfo.getGlueUpdatetime().getTime()); triggerParam.setBroadcastIndex(index); triggerParam.setBroadcastTotal(total); triggerParam.setAssignAddress(jobInfo.getAssignAddress()); triggerParam.setJobEmptyLoopNum(jobInfo.getJobEmptyLoopNum()); // 3、init address 选择执行器服务地址 String address = null; ReturnT<String> routeAddre***esult = null; if (group.getRegistryList()!=null && !group.getRegistryList().isEmpty()) { if (ExecutorRouteStrategyEnum.SHARDING_BROADCAST == executorRouteStrategyEnum) { if (index < group.getRegistryList().size()) { address = group.getRegistryList().get(index); } else { address = group.getRegistryList().get(0); } } else { //根据路由策略，选择合适的执行器服务地址来执行任务 routeAddre***esult = executorRouteStrategyEnum.getRouter().route(triggerParam, group.getRegistryList()); if (routeAddre***esult.getCode() == ReturnT.SUCCESS_CODE) { address = routeAddre***esult.getContent(); } } } else { routeAddre***esult = new ReturnT<String>(ReturnT.FAIL_CODE, I18nUtil.getString("jobconf_trigger_address_empty")); } // 4、trigger remote executor 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(" ").append(I18nUtil.getString("jobconf_trigger_admin_adress")).append("：").append(IpUtil.getIp()); triggerMsgSb.append(" ").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(" ").append(I18nUtil.getString("jobconf_trigger_exe_regaddress")).append("：").append(group.getRegistryList()); triggerMsgSb.append(" ").append(I18nUtil.getString("jobinfo_field_executorRouteStrategy")).append("：").append(executorRouteStrategyEnum.getTitle()); if (shardingParam != null) { triggerMsgSb.append("("+shardingParam+")"); } triggerMsgSb.append(" ").append(I18nUtil.getString("jobinfo_field_executorBlockStrategy")).append("：").append(blockStrategy.getTitle()); triggerMsgSb.append(" ").append(I18nUtil.getString("jobinfo_field_timeout")).append("：").append(jobInfo.getExecutorTimeout()); triggerMsgSb.append(" ").append(I18nUtil.getString("jobinfo_field_executorFailRetryCount")).append("：").append(finalFailRetryCount); triggerMsgSb.append(" >>>>>>>>>>>"+ I18nUtil.getString("jobconf_trigger_run") +"<<<<<<<<<<< ") .append((routeAddre***esult!=null&&routeAddre***esult.getMsg()!=null)?routeAddre***esult.getMsg()+" ":"").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()); }

由于某些场景，某个任务必须保证只能执行一次，或者宁愿不执行也不允许重复执行，比方说发放优惠券，可以少发或者不发，但是不能多发，这种情况下，让任务固定到一个执行器服务IP上执行，所以在原来的基础上增加了一个策略，固定IP策略

组装好参数，选择了执行任务地址，进入runExecutor方法

创建好了RPC的客户端对象，在创建对象过程中使用了NettyHttp协议，HESSIAN序列化，就可以发起RPC请求了

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的代理对象，重点在这个里面。 executorBiz = (ExecutorBiz) new XxlRpcReferenceBean( NetEnum.NETTY_HTTP, //nettyHttp Serializer.SerializeEnum.HESSIAN.getSerializer(),//序列化 CallType.SYNC,//同步 LoadBalance.ROUND, ExecutorBiz.class, null, 5000, address, XxlJobAdminConfig.getAdminConfig().getAccessToken(), null, null).getObject(); executorBizRepository.put(address, executorBiz); //对象放入缓存 return executorBiz; }

然后发起RPC请求：executorBiz.run

public ReturnT<String> run(TriggerParam triggerParam) { // load old：jobHandler + jobThread // 通过参数中的JobID，从本地线程库里面获取线程 ( 第一次进来是没有线程的，jobThread为空， // 本地线程库，本质上就是一个ConcurrentHashMap<Integer, JobThread> JobThread jobThread = XxlJobExecutor.loadJobThread(triggerParam.getJobId()); IJobHandler jobHandler = jobThread!=null?jobThread.getHandler():null; String removeOldReason = null; // valid：jobHandler + jobThread //运行模式，这里看一下java模式就可以了 GlueTypeEnum glueTypeEnum = GlueTypeEnum.match(triggerParam.getGlueType()); if (GlueTypeEnum.BEAN == glueTypeEnum) { // new jobhandler // 通过参数中的handlerName从本地内存中获取handler实例 // （在执行器启动的时候，是把所有带有@JobHandler的实例通过name放入到一个map中的） IJobHandler newJobHandler = XxlJobExecutor.loadJobHandler(triggerParam.getExecutorHandler()); // valid old jobThread // 如果修改了任务的handler， name此处会默认把以前老的handler清空，后面会以最新的newJobHandler为准 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) // 如果jobThread为空，那么这个时候，就要注册一个线程到本地线程库里面去。 // 然后启动这个线程，线程会轮训任务队列开始执行，可以查看JobThread.run方法 if (jobThread == null) { jobThread = XxlJobExecutor.registJobThread(triggerParam.getJobId(), jobHandler, removeOldReason, triggerParam.getJobEmptyLoopNum()); } // push data to queue // 任务线程已经存在了，将任务参数放入任务队列，每个任务线程有一个任务队列，任务线程去轮询这个任务 ReturnT<String> pushResult = jobThread.pushTriggerQueue(triggerParam); return pushResult; }

至此调度中心，基本处理完成。然后看执行器的操作流程。

执行器的流程1、执行器部署启动会执行XxlJobSpringExecutor.start方法：

然后执行XxlJobExecutor。start方法:

public void start() throws Exception { // init logpath 初始化本地日志路径 XxlJobFileAppender.initLogPath(logPath); // init invoker, admin-client // 初始化调度中心的地址列表，创建好adminBiz实例，调度中心客户端 initAdminBizList(adminAddresses, accessToken); // init JobLogFileCleanThread JobLogFileCleanThread.getInstance().start(logRetentionDays); // init TriggerCallbackThread TriggerCallbackThread.getInstance().start(); // init executor-server port = port>0?port: NetUtil.findAvailablePort(9999); ip = (ip!=null&&ip.trim().length()>0)?ip: IpUtil.getIp(); //启动执行器服务，默认开端口9999 initRpcProvider(ip, port, appName, accessToken); }

RPC服务启动了，就可以正常提供执行器服务了。

任务线程如何处理任务。JobThread就是一个线程

重点看一下线程的run方法，会循环获取队列里的任务，每次获取超时时间是3秒，默认30次，如果没有任务就停止线程，这里的30次已经进行了定制化修改。

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 { //获取队列里的任务，设置3秒钟超时 triggerParam = triggerQueue.poll(3L, TimeUnit.SECONDS); if (triggerParam!=null) { running = true; idleTimes = 0; triggerLogIdSet.remove(triggerParam.getLogId()); // log filename, like "logPath/yyyy-MM-dd/9999.log" String logFileName = XxlJobFileAppender.makeLogFileName(new Date(triggerParam.getLogDateTim()), triggerParam.getLogId()); XxlJobFileAppender.contextHolder.set(logFileName); ShardingUtil.setShardingVo(new ShardingUtil.ShardingVO(triggerParam.getBroadcastIndex(), triggerParam.getBroadcastTotal())); // execute XxlJobLogger.log(" ----------- xxl-job job execute start ----------- ----------- 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(" ----------- 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(" ----------- xxl-job job execute end(finish) ----------- ----------- ReturnT:" + executeResult); } else { //超过一定次数，清空线程，并设置JobThread的stop停止标识位，终止轮询。也就是3*jobEmptyLoopNum秒空轮询 XxlJobLogger.log(" ----------- xxl-job loop num diy set Param:" + jobEmptyLoopNum); if (idleTimes > jobEmptyLoopNum) { XxlJobExecutor.removeJobThread(jobId, "excutor idel times over limit."); } } } catch (Throwable e) { if (toStop) { XxlJobLogger.log(" ----------- 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(" ----------- JobThread Exception:" + errorMsg + " ----------- xxl-job job execute end(error) -----------"); } finally { if(triggerParam != null) { // callback handler info if (!toStop) { // commonm TriggerCallbackThread.pushCallBack(new HandleCallbackParam(triggerParam.getLogId(), triggerParam.getLogDateTim(), executeResult)); } else { // is killed ReturnT<String> stopResult = new ReturnT<String>(ReturnT.FAIL_CODE, stopReason + " [job running，killed]"); TriggerCallbackThread.pushCallBack(new HandleCallbackParam(triggerParam.getLogId(), triggerParam.getLogDateTim(), 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.getLogDateTim(), stopResult)); } } // destroy try { handler.destroy(); } catch (Throwable e) { logger.error(e.getMessage(), e); } logger.info(">>>>>>>>>>> xxl-job JobThread stoped, hashCode:{}", Thread.currentThread()); }