XXL-JOB原理--任务执行（五）

ゝ一世哀愁。 2022-05-12 07:34 314阅读 0赞

一、任务调度中心发送任务执行请求

任务发送执行的操作有两种：

（1）根据配置的cron表达式周期性执行相关任务

（2）在任务调度中心主动执行任务

在注册quartz定时任务时已经注册执行类为RemoteHttpJobBean，所以周期性执行定时任务会调用RemoteHttpJobBean的executeInternal方法，在executeInternal中会调用JobTriggerPoolHelper.trigger(jobId)，通过任务调度中心主动执行任务时也是会调用JobTriggerPoolHelper.trigger(jobId)方法，所以接下来我们要看的是JobTriggerPoolHelper.trigger(jobId)中做的逻辑处理就好。

public class RemoteHttpJobBean extends QuartzJobBean {
    	private static Logger logger = LoggerFactory.getLogger(RemoteHttpJobBean.class);
    
    	@Override
    	protected void executeInternal(JobExecutionContext context)
    			throws JobExecutionException {
    
    		// load jobId
    		JobKey jobKey = context.getTrigger().getJobKey();
    		Integer jobId = Integer.valueOf(jobKey.getName());
    
    		// trigger
    		//XxlJobTrigger.trigger(jobId);
    		JobTriggerPoolHelper.trigger(jobId);
    	}
    
    }

JobTriggerPoolHelper.trigger所做的操作是将任务提交给一个线程池（任务调度中心默认开启50个线程），在线程池中调用XxlJobTrigger.trigger。

public void addTrigger(final int jobId){
            triggerPool.execute(new Runnable() {
                @Override
                public void run() {
                    XxlJobTrigger.trigger(jobId);
                }
            });
        }

在XxlJobTrigger.trigger中会根据jobId获取任务的基本配置信息（阻塞策略、路由策略、失败重试测试、分组服务器列表等等），然后根据路由策略选择是广播还是单播等，接下来就是组装消息体调用runExecutor方法发送http请求到任务执行器。

public static void trigger(int jobId) {
 
 		//获取任务信息
 // load data
 XxlJobInfo jobInfo = XxlJobDynamicScheduler.xxlJobInfoDao.loadById(jobId); // job info
 if (jobInfo == null) {
 logger.warn(">>>>>>>>>>>> trigger fail, jobId invalid，jobId={}", jobId);
 return;
 }
 		//根据任务的分组信息找到分组，分组中存在服务器的IP和端口地址等
 XxlJobGroup group = XxlJobDynamicScheduler.xxlJobGroupDao.load(jobInfo.getJobGroup()); // group info
 		
 		//阻塞策略
 ExecutorBlockStrategyEnum blockStrategy = ExecutorBlockStrategyEnum.match(jobInfo.getExecutorBlockStrategy(), ExecutorBlockStrategyEnum.SERIAL_EXECUTION); // block strategy
 //失败策略
 		ExecutorFailStrategyEnum failStrategy = ExecutorFailStrategyEnum.match(jobInfo.getExecutorFailStrategy(), ExecutorFailStrategyEnum.NULL); // fail strategy
 //执行路由测试
 		ExecutorRouteStrategyEnum executorRouteStrategyEnum = ExecutorRouteStrategyEnum.match(jobInfo.getExecutorRouteStrategy(), null); // route strategy
 		//服务器地址
 ArrayList<String> addressList = (ArrayList<String>) group.getRegistryList();
 
 		//广播模式
 // broadcast
 if (ExecutorRouteStrategyEnum.SHARDING_BROADCAST == executorRouteStrategyEnum && CollectionUtils.isNotEmpty(addressList)) {
 			//依次调用所有的服务器
 for (int i = 0; i < addressList.size(); i++) {
 String address = addressList.get(i);
 
 // 1、save log-id
 XxlJobLog jobLog = new XxlJobLog();
 jobLog.setJobGroup(jobInfo.getJobGroup());
 jobLog.setJobId(jobInfo.getId());
 XxlJobDynamicScheduler.xxlJobLogDao.save(jobLog);
 logger.debug(">>>>>>>>>>> xxl-job trigger start, jobId:{}", jobLog.getId());
 
 // 2、prepare trigger-info
 //jobLog.setExecutorAddress(executorAddress);
 jobLog.setGlueType(jobInfo.getGlueType());
 jobLog.setExecutorHandler(jobInfo.getExecutorHandler());
 jobLog.setExecutorParam(jobInfo.getExecutorParam());
 jobLog.setTriggerTime(new Date());
 
 ReturnT<String> triggerResult = new ReturnT<String>(null);
 StringBuffer triggerMsgSb = new StringBuffer();
 triggerMsgSb.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()).append("("+i+"/"+addressList.size()+")"); // update01
 triggerMsgSb.append(" ").append(I18nUtil.getString("jobinfo_field_executorBlockStrategy")).append("：").append(blockStrategy.getTitle());
 triggerMsgSb.append(" ").append(I18nUtil.getString("jobinfo_field_executorFailStrategy")).append("：").append(failStrategy.getTitle());
 triggerMsgSb.append(" ").append(I18nUtil.getString("jobinfo_field_timeout")).append("：").append(jobInfo.getExecutorTimeout());
 
 // 3、trigger-valid
 if (triggerResult.getCode()==ReturnT.SUCCESS_CODE && CollectionUtils.isEmpty(addressList)) {
 triggerResult.setCode(ReturnT.FAIL_CODE);
 triggerMsgSb.append(" ---------------------- ").append(I18nUtil.getString("jobconf_trigger_address_empty"));
 }
 
 if (triggerResult.getCode() == ReturnT.SUCCESS_CODE) {
 // 4.1、trigger-param
 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(i);
 triggerParam.setBroadcastTotal(addressList.size()); // update02
 
 // 4.2、trigger-run (route run / trigger remote executor)
 					//远程调用服务接口，执行任务
 triggerResult = runExecutor(triggerParam, address); // update03
 triggerMsgSb.append(" >>>>>>>>>>>"+ I18nUtil.getString("jobconf_trigger_run") +"<<<<<<<<<<< ").append(triggerResult.getMsg());
 
 // 4.3、trigger (fail retry)
 if (triggerResult.getCode()!=ReturnT.SUCCESS_CODE && failStrategy == ExecutorFailStrategyEnum.FAIL_TRIGGER_RETRY) {
 triggerResult = runExecutor(triggerParam, address); // update04
 triggerMsgSb.append(" >>>>>>>>>>>"+ I18nUtil.getString("jobconf_fail_trigger_retry") +"<<<<<<<<<<< ").append(triggerResult.getMsg());
 }
 }
 
 // 5、save trigger-info
 jobLog.setExecutorAddress(triggerResult.getContent());
 jobLog.setTriggerCode(triggerResult.getCode());
 jobLog.setTriggerMsg(triggerMsgSb.toString());
 XxlJobDynamicScheduler.xxlJobLogDao.updateTriggerInfo(jobLog);
 
 // 6、monitor trigger
 JobFailMonitorHelper.monitor(jobLog.getId());
 logger.debug(">>>>>>>>>>> xxl-job trigger end, jobId:{}", jobLog.getId());
 
 }
 } else {
 			//单播模式
 // 1、save log-id
 XxlJobLog jobLog = new XxlJobLog();
 jobLog.setJobGroup(jobInfo.getJobGroup());
 jobLog.setJobId(jobInfo.getId());
 XxlJobDynamicScheduler.xxlJobLogDao.save(jobLog);
 logger.debug(">>>>>>>>>>> xxl-job trigger start, jobId:{}", jobLog.getId());
 
 // 2、prepare trigger-info
 //jobLog.setExecutorAddress(executorAddress);
 jobLog.setGlueType(jobInfo.getGlueType());
 jobLog.setExecutorHandler(jobInfo.getExecutorHandler());
 jobLog.setExecutorParam(jobInfo.getExecutorParam());
 jobLog.setTriggerTime(new Date());
 
 ReturnT<String> triggerResult = new ReturnT<String>(null);
 StringBuffer triggerMsgSb = new StringBuffer();
 triggerMsgSb.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());
 triggerMsgSb.append(" ").append(I18nUtil.getString("jobinfo_field_executorBlockStrategy")).append("：").append(blockStrategy.getTitle());
 triggerMsgSb.append(" ").append(I18nUtil.getString("jobinfo_field_executorFailStrategy")).append("：").append(failStrategy.getTitle());
 triggerMsgSb.append(" ").append(I18nUtil.getString("jobinfo_field_timeout")).append("：").append(jobInfo.getExecutorTimeout());
 
 // 3、trigger-valid
 if (triggerResult.getCode()==ReturnT.SUCCESS_CODE && CollectionUtils.isEmpty(addressList)) {
 triggerResult.setCode(ReturnT.FAIL_CODE);
 triggerMsgSb.append(" ---------------------- ").append(I18nUtil.getString("jobconf_trigger_address_empty"));
 }
 
 if (triggerResult.getCode() == ReturnT.SUCCESS_CODE) {
 // 4.1、trigger-param
 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(0);
 triggerParam.setBroadcastTotal(1);
 
 // 4.2、trigger-run (route run / trigger remote executor)
 				//路由后远程调用服务接口，执行任务
 triggerResult = executorRouteStrategyEnum.getRouter().routeRun(triggerParam, addressList);
 triggerMsgSb.append(" >>>>>>>>>>>"+ I18nUtil.getString("jobconf_trigger_run") +"<<<<<<<<<<< ").append(triggerResult.getMsg());
 
 // 4.3、trigger (fail retry)
 if (triggerResult.getCode()!=ReturnT.SUCCESS_CODE && failStrategy == ExecutorFailStrategyEnum.FAIL_TRIGGER_RETRY) {
 triggerResult = executorRouteStrategyEnum.getRouter().routeRun(triggerParam, addressList);
 triggerMsgSb.append(" >>>>>>>>>>>"+ I18nUtil.getString("jobconf_fail_trigger_retry") +"<<<<<<<<<<< ").append(triggerResult.getMsg());
 }
 }
 
 // 5、save trigger-info
 jobLog.setExecutorAddress(triggerResult.getContent());
 jobLog.setTriggerCode(triggerResult.getCode());
 jobLog.setTriggerMsg(triggerMsgSb.toString());
 XxlJobDynamicScheduler.xxlJobLogDao.updateTriggerInfo(jobLog);
 
 // 6、monitor trigger
 JobFailMonitorHelper.monitor(jobLog.getId());
 logger.debug(">>>>>>>>>>> xxl-job trigger end, jobId:{}", jobLog.getId());
 }
 
 }

runExecutor方法中根据address服务器地址，XxlJobDynamicScheduler.getExecutorBiz中会获取代理类最终调用JettyClient的send方法。

public static ReturnT<String> runExecutor(TriggerParam triggerParam, String address){
 ReturnT<String> runResult = null;
 try {
 			//获取代理对象
 ExecutorBiz executorBiz = XxlJobDynamicScheduler.getExecutorBiz(address);
 			//最终调用执行
 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, ""+e );
 }
 
 StringBuffer runResultSB = new StringBuffer(I18nUtil.getString("jobconf_trigger_run") + "：");
 runResultSB.append(" address：").append(address);
 runResultSB.append(" code：").append(runResult.getCode());
 runResultSB.append(" msg：").append(runResult.getMsg());
 
 runResult.setMsg(runResultSB.toString());
 runResult.setContent(address);
 return runResult;
 }

在XxlJobDynamicScheduler的getExecutorBiz中会通过NetComClientProxy生成代理对象，在执行时会调用其方法。

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) new NetComClientProxy(ExecutorBiz.class, address, accessToken).getObject();
            executorBizRepository.put(address, executorBiz);
            return executorBiz;
        }

getObject中生成代理对象，执行会执行JettyClient的send方法。

@Override
    	public Object getObject() throws Exception {
    		return Proxy.newProxyInstance(Thread.currentThread()
    				.getContextClassLoader(), new Class[] { iface },
    				new InvocationHandler() {
    					@Override
    					public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
    
    						// filter method like "Object.toString()"
    						if (Object.class.getName().equals(method.getDeclaringClass().getName())) {
    							logger.error(">>>>>>>>>>> xxl-rpc proxy class-method not support [{}.{}]", method.getDeclaringClass().getName(), method.getName());
    							throw new RuntimeException("xxl-rpc proxy class-method not support");
    						}
    						
    						// request
    						RpcRequest request = new RpcRequest();
    	                    request.setServerAddress(serverAddress);
    	                    request.setCreateMillisTime(System.currentTimeMillis());
    	                    request.setAccessToken(accessToken);
    	                    request.setClassName(method.getDeclaringClass().getName());
    	                    request.setMethodName(method.getName());
    	                    request.setParameterTypes(method.getParameterTypes());
    	                    request.setParameters(args);
    						
    						//发起http调用，执行任务
    	                    // send
    	                    RpcResponse response = client.send(request);
    	                    
    	                    // valid response
    						if (response == null) {
    							throw new Exception("Network request fail, response not found.");
    						}
    	                    if (response.isError()) {
    	                        throw new RuntimeException(response.getError());
    	                    } else {
    	                        return response.getResult();
    	                    }
    	                   
    					}
    				});
    	}

任务调度中心向任务执行器发送的任务请求数据如下。

![70][]

二、任务执行器接收任务执行

在任务执行器会根据内置的jetty提供web服务，提供请求处理器JettyServerHandler接收处理任务调度中心发送过来的任务

public class JettyServerHandler extends AbstractHandler {
    	private static Logger logger = LoggerFactory.getLogger(JettyServerHandler.class);
    
    	//接收请求，处理任务
    	@Override
    	public void handle(String target, Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException {
    		
    		// invoke
    		//调用任务执行
            RpcResponse rpcResponse = doInvoke(request);
    
            // serialize response
            byte[] responseBytes = HessianSerializer.serialize(rpcResponse);
    		
    		response.setContentType("text/html;charset=utf-8");
    		response.setStatus(HttpServletResponse.SC_OK);
    		baseRequest.setHandled(true);
    		
    		OutputStream out = response.getOutputStream();
    		out.write(responseBytes);
    		out.flush();
    		
    	}
    
    	private RpcResponse doInvoke(HttpServletRequest request) {
    		try {
    			// deserialize request
    			byte[] requestBytes = HttpClientUtil.readBytes(request);
    			if (requestBytes == null || requestBytes.length==0) {
    				RpcResponse rpcResponse = new RpcResponse();
    				rpcResponse.setError("RpcRequest byte[] is null");
    				return rpcResponse;
    			}
    			//反序列化数据
    			RpcRequest rpcRequest = (RpcRequest) HessianSerializer.deserialize(requestBytes, RpcRequest.class);
    
    			// invoke
    			//通过反射调用任务
    			RpcResponse rpcResponse = NetComServerFactory.invokeService(rpcRequest, null);
    			return rpcResponse;
    		} catch (Exception e) {
    			logger.error(e.getMessage(), e);
    
    			RpcResponse rpcResponse = new RpcResponse();
    			rpcResponse.setError("Server-error:" + e.getMessage());
    			return rpcResponse;
    		}
    	}
    
    }

在invokeService根据发送过来的类名com.xxl.job.core.biz.ExecutorBiz和方法run，通过反射机制调用

public static RpcResponse invokeService(RpcRequest request, Object serviceBean) {
 		if (serviceBean==null) {
 			serviceBean = serviceMap.get(request.getClassName());
 		}
 		if (serviceBean == null) {
 			// TODO
 		}
 
 		RpcResponse response = new RpcResponse();
 
 		if (System.currentTimeMillis() - request.getCreateMillisTime() > 180000) {
 			response.setResult(new ReturnT<String>(ReturnT.FAIL_CODE, "The timestamp difference between admin and executor exceeds the limit."));
 			return response;
 		}
 		if (accessToken!=null && accessToken.trim().length()>0 && !accessToken.trim().equals(request.getAccessToken())) {
 			response.setResult(new ReturnT<String>(ReturnT.FAIL_CODE, "The access token[" + request.getAccessToken() + "] is wrong."));
 			return response;
 		}
 
 		try {
 			Class<?> serviceClass = serviceBean.getClass();
 			String methodName = request.getMethodName();
 			Class<?>[] parameterTypes = request.getParameterTypes();
 			Object[] parameters = request.getParameters();
 
 			FastClass serviceFastClass = FastClass.create(serviceClass);
 			FastMethod serviceFastMethod = serviceFastClass.getMethod(methodName, parameterTypes);
 
 			Object result = serviceFastMethod.invoke(serviceBean, parameters);
 
 			response.setResult(result);
 		} catch (Throwable t) {
 			t.printStackTrace();
 			response.setError(t.getMessage());
 		}
 
 		return response;
 	}

我们看看com.xxl.job.core.biz.ExecutorBiz的run方法中做了什么处理操作。

在ExecutorBiz中根据发送过来的消息，根据demoJobHandler找到接口的实现类，接下来就可以新起线程去执行实现类DemoJobHandler了。

@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;
 }

总结：任务执行器提供web服务，任务调度中心根据任务分组及分组服务器发送http请求，任务执行器收到请求，根据请求中的数据调用对应的任务。

[70]: /images/20220512/2682563189274f3b966e422ad6613cb7.png