【xxl-job】源码分析 - 任务触发

太过爱你忘了你带给我的痛 2022-05-10 13:16 151阅读 0赞

【引言】

上篇博客简单分析了xxl-job项目的启动流程，本篇博客分析的内容是调度任务的触发。在之前的博客中，也了解到xxl-job是基于quartz实现的，下面就从源码学习下是如何实现的。

【实现】

任务调度的核心类是RemoteHttpJobBean，它继承了QuartzJobBean，默认是允许并行机制的，代码如下：

/**
     * http job bean
     * “@DisallowConcurrentExecution” diable concurrent, thread size can not be only one, better given more
     * @author xuxueli 2015-12-17 18:20:34
     */
    //@DisallowConcurrentExecution
    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, -1, TriggerTypeEnum.CRON);
    	}
    
    }

以上核心方法是JobTriggerPoolHelper.trigger()，在此方法中，根据路由策略区分，分片广播策略是一种处理方法，其他策略归并为另一种处理方法，策略不同，调用不同的实现。

总结一下，处理动作分为五步，1、save log-id - > 2、prepare trigger-info - > 3.1、trigger-param - > 3.2、trigger-run (route run / trigger remote executor) - > 4、save trigger-info - > 5、monitor trigger 。核心是3.2中的runExecutor()，向执行器中发送指令，源码如下：

/**
         * run executor
         * @param triggerParam
         * @param address
         * @return  ReturnT.content: final address
         */
        public static ReturnT<String> runExecutor(TriggerParam triggerParam, String address){
            ReturnT<String> runResult = null;
            try {
            //创建一个ExcutorBiz 的对象，重点在这个方法里面
                ExecutorBiz executorBiz = XxlJobDynamicScheduler.getExecutorBiz(address);
            // 这个run 方法不会最终执行，仅仅只是为了触发 proxy object 的 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, ""+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());
            runResult.setContent(address);
            return runResult;
        }

XxlJobDynamicScheduler.getExecutorBiz （address） , 通过机器地址，获取一个executor , 从缓存中查是否存在，缓存中没有的话，创建代理对象，(ExecutorBiz) new NetComClientProxy(ExecutorBiz.class, address, accessToken).getObject()，源码如下：

/**
     * rpc proxy
     * @author xuxueli 2015-10-29 20:18:32
     */
    public class NetComClientProxy implements FactoryBean<Object> {
    	private static final Logger logger = LoggerFactory.getLogger(NetComClientProxy.class);
    
    	// ---------------------- config ----------------------
    	private Class<?> iface;
    	private String serverAddress;
    	private String accessToken;
    	private JettyClient client = new JettyClient();
    	public NetComClientProxy(Class<?> iface, String serverAddress, String accessToken) {
    		this.iface = iface;
    		this.serverAddress = serverAddress;
    		this.accessToken = accessToken;
    	}
    
    	@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
    						// 创建request信息， 发送HTTP请求到执行器服务器上。
    						RpcRequest request = new RpcRequest();
    	                    request.setServerAddress(serverAddress); // 服务器地址
    	                    request.setCreateMillisTime(System.currentTimeMillis()); // 创建时间， 用于判断请求是否超时
    	                    request.setAccessToken(accessToken); // 数据校验
    	                    request.setClassName(method.getDeclaringClass().getName()); // 将目标类的class名称传给执行器，让那边来创建对象，并执行逻辑代码
    	                    request.setMethodName(method.getName()); // 方法名称为run 
    	                    request.setParameterTypes(method.getParameterTypes()); // 参数类型
    	                    request.setParameters(args); // 参数
    
    	                    // send
    	                    RpcResponse response = client.send(request); // 发送HTTP请求
    	                    
    	                    // 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(); // 返回请求结果
    	                    }
    	                   
    					}
    				});
    	}
    	@Override
    	public Class<?> getObjectType() {
    		return iface;
    	}
    	@Override
    	public boolean isSingleton() {
    		return false;
    	}
    
    }

【总结】

以上就是调度中心，触发任务之后执行的核心代码，重点就是最后的发送http请求给执行器，由执行器去执行代码，下篇博客将分析执行器接受到任务的处理过程。