深入解析XXL-JOB任务调度执行原理
引言
在分布式系统中，定时任务调度是业务场景中不可或缺的一环。面对海量任务、复杂依赖和高可用性要求，传统单机调度方案逐渐显得力不从心。XXL-JOB作为一款开源的分布式任务调度平台，凭借其轻量级、高扩展性和易用性成为众多企业的选择。本文将深入剖析XXL-JOB的任务执行原理，揭示其背后的设计哲学。
一、架构设计
1、 设计思想
将调度行为抽象形成“调度中心”公共平台，而平台自身并不承担业务逻辑，“调度中心”负责发起调度请求。
将任务抽象成分散的JobHandler，交由“执行器”统一管理，“执行器”负责接收调度请求并执行对应的JobHandler中业务逻辑。
因此，“调度”和“任务”两部分可以相互解耦，提高系统整体稳定性和扩展性；
2、 系统组成
- 调度模块（调度中心）：
负责管理调度信息，按照调度配置发出调度请求，自身不承担业务代码。调度系统与任务解耦，提高了系统可用性和稳定性，同时调度系统性能不再受限于任务模块；
支持可视化、简单且动态的管理调度信息，包括任务新建，更新，删除，GLUE开发和任务报警等，所有上述操作都会实时生效，同时支持监控调度结果以及执行日志，支持执行器Failover。 - 执行模块（执行器）：
负责接收调度请求并执行任务逻辑。任务模块专注于任务的执行等操作，开发和维护更加简单和高效；
接收“调度中心”的执行请求、终止请求和日志请求等。
3、 架构图
4、 quartz的不足
Quartz作为开源作业调度中的佼佼者，是作业调度的首选。但是集群环境中Quartz采用API的方式对任务进行管理，从而可以避免上述问题，但是同样存在以下问题：
- 问题一：调用API的方式操作任务，不人性化；
- 问题二：需要持久化业务QuartzJobBean到底层数据表中，系统侵入性相当严重。
- 问题三：调度逻辑和QuartzJobBean耦合在同一个项目中，这将导致一个问题，在调度任务数量逐渐增多，同时调度任务逻辑逐渐加重的情况下，此时调度系统的性能将大大受限于业务；
- 问题四：quartz底层以“抢占式”获取DB锁并由抢占成功节点负责运行任务，会导致节点负载悬殊非常大；而XXL-JOB通过执行器实现“协同分配式”运行任务，充分发挥集群优势，负载各节点均衡。
XXL-JOB弥补了quartz的上述不足之处。
5、自研调度模块
XXL-JOB最终选择自研调度组件（早期调度组件基于Quartz）；一方面是为了精简系统降低冗余依赖，另一方面是为了提供系统的可控度与稳定性；
XXL-JOB中“调度模块”和“任务模块”完全解耦，调度模块进行任务调度时，将会解析不同的任务参数发起远程调用，调用各自的远程执行器服务。这种调用模型类似RPC调用，调度中心提供调用代理的功能，而执行器提供远程服务的功能。
二、源码解析
1、执行器入口
在配置执行器组件时，会配置一个XxlJobSpringExecutor的Bean
@Bean
public XxlJobSpringExecutor xxlJobExecutor() {
logger.info(">>>>>>>>>>> xxl-job config init.");
XxlJobSpringExecutor xxlJobSpringExecutor = new XxlJobSpringExecutor();
//设置调用中心的连接地址
xxlJobSpringExecutor.setAdminAddresses(adminAddresses);
//设置执行器的名称
xxlJobSpringExecutor.setAppname(appname);
//设置执行器IP
xxlJobSpringExecutor.setIp(ip);
//执行器端口号
xxlJobSpringExecutor.setPort(port);
//调度中心通讯token
xxlJobSpringExecutor.setAccessToken(accessToken);
//任务执行日志存放的目录
xxlJobSpringExecutor.setLogPath(logPath);
//执行器日志文件保存天数
xxlJobSpringExecutor.setLogRetentionDays(logRetentionDays);
return xxlJobSpringExecutor;
}
XxlJobSpringExecutor这个类就是执行器的入口
该类实现了SmartInitializingSingleton接口，经过Bean的生命周期，会调用afterSingletonsInstantiated这个方法
public class XxlJobSpringExecutor extends XxlJobExecutor implements ApplicationContextAware, SmartInitializingSingleton, DisposableBean {
private static final Logger logger = LoggerFactory.getLogger(XxlJobSpringExecutor.class);
@Override
public void afterSingletonsInstantiated() {
// init JobHandler Repository (for method)
initJobHandlerMethodRepository(applicationContext);
GlueFactory.refreshInstance(1);
try {
super.start();
} catch (Exception e) {
throw new RuntimeException(e);
}
}
2、初始化JobHandler
首先来看initJobHandlerMethodRepository这个方法
private void initJobHandlerMethodRepository(ApplicationContext applicationContext) {
if (applicationContext == null) {
return;
}
//getBeanNamesForType获取所有beanNames的集合
String[] beanDefinitionNames = applicationContext.getBeanNamesForType(Object.class, false, true);
for (String beanDefinitionName : beanDefinitionNames) {
Object bean = null;
//如果是懒加载也就是加了@Lazy注解的，直接跳过
Lazy onBean = applicationContext.findAnnotationOnBean(beanDefinitionName, Lazy.class);
if (onBean!=null){
logger.debug("xxl-job annotation scan, skip @Lazy Bean:{}", beanDefinitionName);
continue;
}else {
//获取当前的bean
bean = applicationContext.getBean(beanDefinitionName);
}
//key为Method方法，值为XxlJob注解的集合
Map<Method, XxlJob> annotatedMethods = null;
try {
//通过selectMethods方法返回map对象
annotatedMethods = MethodIntrospector.selectMethods(bean.getClass(),
new MethodIntrospector.MetadataLookup<XxlJob>() {
@Override
public XxlJob inspect(Method method) {
//当前method有没有加XxlJob注解
return AnnotatedElementUtils.findMergedAnnotation(method, XxlJob.class);
}
});
} catch (Throwable ex) {
logger.error("xxl-job method-jobhandler resolve error for bean[" + beanDefinitionName + "].", ex);
}
if (annotatedMethods==null || annotatedMethods.isEmpty()) {
continue;
}
for (Map.Entry<Method, XxlJob> methodXxlJobEntry : annotatedMethods.entrySet()) {
//获取当前Entry的key和value
Method executeMethod = methodXxlJobEntry.getKey();
XxlJob xxlJob = methodXxlJobEntry.getValue();
//注册
registJobHandler(xxlJob, bean, executeMethod);
}
}
}
2.1 MethodIntrospector.selectMethods()
public static <T> Map<Method, T> selectMethods(Class<?> targetType, final MetadataLookup<T> metadataLookup) {
//返回的map集合
final Map<Method, T> methodMap = new LinkedHashMap<>();
Set<Class<?>> handlerTypes = new LinkedHashSet<>();
//添加给定类实现的所有接口的集合，相当于把当前类的上一级全部添加进来
handlerTypes.addAll(ClassUtils.getAllInterfacesForClassAsSet(targetType));
for (Class<?> currentHandlerType : handlerTypes) {
//调用doWithMethods方法，传入函数式接口实现
ReflectionUtils.doWithMethods(currentHandlerType, method -> {
Method specificMethod = ClassUtils.getMostSpecificMethod(method, targetClass);
//调用外层的inspect方法返回结果，也就是XxlJob注解
T result = metadataLookup.inspect(specificMethod);
if (result != null) {
Method bridgedMethod = BridgeMethodResolver.findBridgedMethod(specificMethod);
if (bridgedMethod == specificMethod || metadataLookup.inspect(bridgedMethod) == null) {
//将method和result放入methodMap中
methodMap.put(specificMethod, result);
}
}
}, ReflectionUtils.USER_DECLARED_METHODS);
}
//最后返回map对象
return methodMap;
}
来看一下ReflectionUtils.doWithMethods()这个方法
public static void doWithMethods(Class<?> clazz, MethodCallback mc, @Nullable MethodFilter mf) {
//获取到当前类的声明的所有方法数组
Method[] methods = getDeclaredMethods(clazz, false);
for (Method method : methods) {
if (mf != null && !mf.matches(method)) {
continue;
}
try {
//调用doWith方法，也就是外层的函数式接口实现
mc.doWith(method);
}
catch (IllegalAccessException ex) {
throw new IllegalStateException("Not allowed to access method '" + method.getName() + "': " + ex);
}
}
//对父类或者接口做同样的处理
if (clazz.getSuperclass() != null && (mf != USER_DECLARED_METHODS || clazz.getSuperclass() != Object.class)) {
doWithMethods(clazz.getSuperclass(), mc, mf);
}
else if (clazz.isInterface()) {
for (Class<?> superIfc : clazz.getInterfaces()) {
doWithMethods(superIfc, mc, mf);
}
}
}
2.2 registJobHandler()
protected void registJobHandler(XxlJob xxlJob, Object bean, Method executeMethod){
if (xxlJob == null) {
return;
}
//xxlJob的value值
String name = xxlJob.value();
Class<?> clazz = bean.getClass();
//方法名
String methodName = executeMethod.getName();
if (name.trim().length() == 0) {
throw new RuntimeException("xxl-job method-jobhandler name invalid, for[" + clazz + "#" + methodName + "] .");
}
//如果name存在重复的，则会抛出异常
if (loadJobHandler(name) != null) {
throw new RuntimeException("xxl-job jobhandler[" + name + "] naming conflicts.");
}
executeMethod.setAccessible(true);
//获取init和destroy方法
.....
// registry jobhandler
//这里new了一个MethodJobHandler对象作为参数，后面会详细说明
registJobHandler(name, new MethodJobHandler(bean, executeMethod, initMethod, destroyMethod));
}
//维护的map集合，key是xxljob注解中设置的value值，value是IJobHandler对象
private static ConcurrentMap<String, IJobHandler> jobHandlerRepository = new ConcurrentHashMap<String, IJobHandler>();
public static IJobHandler loadJobHandler(String name){
return jobHandlerRepository.get(name);
}
//注册JobHandler就是放入到jobHandlerRepository这个map集合中
public static IJobHandler registJobHandler(String name, IJobHandler jobHandler){
logger.info(">>>>>>>>>>> xxl-job register jobhandler success, name:{}, jobHandler:{}", name, jobHandler);
return jobHandlerRepository.put(name, jobHandler);
}
2.3 JobHandler
JobHandler其实就是一个定时任务的封装，一个定时任务会对应一个JobHandler对象，注册JobHandler时就是new一个JobHandler对象放入到维护的map集合中
当执行器执行任务的时候，就会调用JobHandler的execute方法
JobHandler有三种实现：
- MethodJobHandler：方法的实现
- GlueJobHandler：支持动态修改任务执行的代码
- ScriptJobHandler：处理脚本任务
public class MethodJobHandler extends IJobHandler {
@Override
public void execute() throws Exception {
Class<?>[] paramTypes = method.getParameterTypes();
//通过反射调用方法
if (paramTypes.length > 0) {
method.invoke(target, new Object[paramTypes.length]);
} else {
method.invoke(target);
}
}
通过上面解析，MethodJobHandler会在项目启动的时候就会创建，GlueJobHandler和ScriptJobHandler都是任务触发时才会创建
3、执行器注册
调用完initJobHandlerMethodRepository方法后，会执行super.start()方法
public void start() throws Exception {
//初始化日志
XxlJobFileAppender.initLogPath(logPath);
//初始化admin集合
initAdminBizList(adminAddresses, accessToken);
//初始化日志清理的线程
JobLogFileCleanThread.getInstance().start(logRetentionDays);
//初始化执行器回调线程
TriggerCallbackThread.getInstance().start();
//初始化执行器服务端
initEmbedServer(address, ip, port, appname, accessToken);
}
private void initEmbedServer(String address, String ip, int port, String appname, String accessToken) throws Exception {
//如果没有指定端口，默认9999端口
port = port>0?port: NetUtil.findAvailablePort(9999);
//没有指定ip，通过IpUtil.getIp()方法获取ip
ip = (ip!=null&&ip.trim().length()>0)?ip: IpUtil.getIp();
//生成注册地址，默认使用指定的address，如果没有指定，则使用ip:port
if (address==null || address.trim().length()==0) {
String ip_port_address = IpUtil.getIpPort(ip, port);
address = "http://{ip_port}/".replace("{ip_port}", ip_port_address);
}
embedServer = new EmbedServer();
//调用start方法
embedServer.start(address, port, appname, accessToken);
}
//创建一个Http服务器,底层是基于Netty实现的,这个Http服务端会接收来自调度中心的请求
public void start(final String address, final int port, final String appname, final String accessToken) {
//当执行器接收到调度中心的请求时，会把请求交给ExecutorBizImpl来处理
//ExecutorBiz还有一个ExecutorBizClient实现类，主要是用来发送http请求，所以这个实现类是在调度中心使用的，用来访问执行器提供的http接口
executorBiz = new ExecutorBizImpl();
thread = new Thread(new Runnable() {
@Override
public void run() {
......
// start registry
startRegistry(appname, address);
......
thread.setDaemon(true);
thread.start();
}
public void startRegistry(final String appname, final String address) {
// start registry
ExecutorRegistryThread.getInstance().start(appname, address);
}
//如果admin是集群部署的话，就会向每台服务器调用registry方法区注册
@Override
public ReturnT<String> registry(RegistryParam registryParam) {
//通过post接口调用
return XxlJobRemotingUtil.postBody(addressUrl + "api/registry", accessToken, timeout, registryParam, String.class);
}
//JobApiController.java
@RequestMapping("/{uri}")
@ResponseBody
@PermissionLimit(limit=false)
public ReturnT<String> api(HttpServletRequest request, @PathVariable("uri") String uri, @RequestBody(required = false) String data) {
......
if ("callback".equals(uri)) {
List<HandleCallbackParam> callbackParamList = GsonTool.fromJson(data, List.class, HandleCallbackParam.class);
return adminBiz.callback(callbackParamList);
} else if ("registry".equals(uri)) {
RegistryParam registryParam = GsonTool.fromJson(data, RegistryParam.class);
//调用registry方法
return adminBiz.registry(registryParam);
} else if ("registryRemove".equals(uri)) {
RegistryParam registryParam = GsonTool.fromJson(data, RegistryParam.class);
return adminBiz.registryRemove(registryParam);
} else {
return new ReturnT<String>(ReturnT.FAIL_CODE, "invalid request, uri-mapping("+ uri +") not found.");
}
}
//最终注册就是往xxl_job_registry表中插入一条记录
public ReturnT<String> registry(RegistryParam registryParam) {
// async execute
registryOrRemoveThreadPool.execute(new Runnable() {
@Override
public void run() {
//先执行更新语句，如果不存在，则执行插入语句
int ret = XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().registryUpdate(registryParam.getRegistryGroup(), registryParam.getRegistryKey(), registryParam.getRegistryValue(), new Date());
if (ret < 1) {
//执行插入方法
XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().registrySave(registryParam.getRegistryGroup(), registryParam.getRegistryKey(), registryParam.getRegistryValue(), new Date());
// fresh
freshGroupRegistryInfo(registryParam);
}
}
});
return ReturnT.SUCCESS;
}
当执行器启动的时候，会启动一个注册线程，这个线程会往调度中心注册当前执行器的信息，包括两部分数据
- 执行器的名字，也就是设置的appname
- 执行器所在机器的ip和端口，这样调度中心就可以访问到这个执行器提供的Http接口
4、任务触发
在xxl-job-admin项目中也会配置一个类似于XxlJobSpringExecutor的Bean，叫XxlJobAdminConfig
//实现了InitializingBean接口，在初始化是会执行afterPropertiesSet方法
@Component
public class XxlJobAdminConfig implements InitializingBean, DisposableBean {
@Override
public void afterPropertiesSet() throws Exception {
adminConfig = this;
xxlJobScheduler = new XxlJobScheduler();
xxlJobScheduler.init();
}
}
public class XxlJobScheduler {
public void init() throws Exception {
// init i18n
initI18n();
//初始化admin的任务触发线程池
JobTriggerPoolHelper.toStart();
//监控和维护注册的执行器实例
JobRegistryHelper.getInstance().start();
//job任务执行失败处理
JobFailMonitorHelper.getInstance().start();
//job任务结果丢失处理
JobCompleteHelper.getInstance().start();
//admin的日志记录
JobLogReportHelper.getInstance().start();
//job任务触发
JobScheduleHelper.getInstance().start();
logger.info(">>>>>>>>> init xxl-job admin success.");
}
}
来看job任务触发的start方法
public void start(){
//开启任务线程
scheduleThread = new Thread(new Runnable() {
@Override
public void run() {
......
//通过数据库实现的分布式锁来保证任务在同一时间只会被其中的一个调度中心触发一次
preparedStatement = conn.prepareStatement( "select * from xxl_job_lock where lock_name = 'schedule_lock' for update" );
preparedStatement.execute();
//查询xxl_job_info这张表中下一次执行的时间 <= 当前时间 + 5s的任务，5s是XxlJob写死的，被称为预读时间，提前读出来，保证任务能准时触发
//查询到任务之后，调度线程会去将这些任务根据执行时间划分为三个部分：
//1、当前时间已经超过任务下一次执行时间5s以上的任务
//2、当前时间已经超过任务下一次执行时间，但是但不足5s的任务
//3、还未到触发时间，但是一定是5s内就会触发执行的
List<XxlJobInfo> scheduleList = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().scheduleJobQuery(nowTime + PRE_READ_MS, preReadCount);
for (XxlJobInfo jobInfo: scheduleList) {
//超过5秒以上会根据任务配置的调度过期策略来选择要不要执行
//调度过期策略就两种,一种直接忽略，一种就是立马执行一次
if (nowTime > jobInfo.getTriggerNextTime() + PRE_READ_MS) {
MisfireStrategyEnum misfireStrategyEnum = MisfireStrategyEnum.match(jobInfo.getMisfireStrategy(), MisfireStrategyEnum.DO_NOTHING);
if (MisfireStrategyEnum.FIRE_ONCE_NOW == misfireStrategyEnum) {
//执行一次
JobTriggerPoolHelper.trigger(jobInfo.getId(), TriggerTypeEnum.MISFIRE, -1, null, null, null);
}
//没有超过5秒则就直接立马执行一次，之后如果判断任务下一次执行时间就在5s内，会直接放到一个时间轮里面，等待下一次触发执行
} else if (nowTime > jobInfo.getTriggerNextTime()) {
JobTriggerPoolHelper.trigger(jobInfo.getId(), TriggerTypeEnum.CRON, -1, null, null, null);
//还没到执行时间，所以不会立马执行，也是直接放到时间轮里面，等待触发执行
} else {
// 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()));
}
}
scheduleThread.setDaemon(true);
scheduleThread.setName("xxl-job, admin JobScheduleHelper#scheduleThread");
scheduleThread.start();
}
}
}
4.1 JobTriggerPoolHelper.trigger()
当任务达到了触发条件，并不是由调度线程直接去触发执行器的任务执行，调度线程会将这个触发的任务交给线程池去执行，并且Xxl-Job为了进一步优化任务的触发，将这个触发任务执行的线程池划分成快线程池和慢线程池两个线程池
在调用执行器的Http接口触发任务执行的时候，Xxl-Job会去记录每个任务的触发所耗费的时间，当任务一次触发的时间超过500ms，那么这个任务的慢次数就会加1，如果这个任务一分钟内触发的慢次数超过10次，接下来就会将触发任务交给慢线程池去执行，所以快慢线程池就是避免那种频繁触发并且每次触发时间还很长的任务阻塞其它任务的触发的情况发生
public static void trigger(int jobId, TriggerTypeEnum triggerType, int failRetryCount, String executorShardingParam, String executorParam, String addressList) {
//执行addTrigger方法
helper.addTrigger(jobId, triggerType, failRetryCount, executorShardingParam, executorParam, addressList);
}
public void addTrigger(final int jobId,
final TriggerTypeEnum triggerType,
final int failRetryCount,
final String executorShardingParam,
final String executorParam,
final String addressList) {
//选择快慢线程池
ThreadPoolExecutor triggerPool_ = fastTriggerPool;
AtomicInteger jobTimeoutCount = jobTimeoutCountMap.get(jobId);
if (jobTimeoutCount!=null && jobTimeoutCount.get() > 10) {
//慢线程池
triggerPool_ = slowTriggerPool;
}
triggerPool_.execute(new Runnable() {
@Override
public void run() {
long start = System.currentTimeMillis();
try {
// do trigger
XxlJobTrigger.trigger(jobId, triggerType, failRetryCount, executorShardingParam, executorParam, addressList);
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
}
});
}
public static void trigger(int jobId,
TriggerTypeEnum triggerType,
int failRetryCount,
String executorShardingParam,
String executorParam,
String addressList) {
......
//如果执行器路由策略是分片广播的话
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 void processTrigger(XxlJobGroup group, XxlJobInfo jobInfo, int finalFailRetryCount, TriggerTypeEnum triggerType, int index, int total){
// 1、save log-id
// 2、init trigger-param
// 3、init address
String address = null;
ReturnT<String> routeAddressResult = 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 {
//选择路由
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
ReturnT<String> triggerResult = null;
if (address != null) {
//远程调用执行器执行任务
triggerResult = runExecutor(triggerParam, address);
} else {
triggerResult = new ReturnT<String>(ReturnT.FAIL_CODE, null);
}
// 5、collection trigger info
// 6、save log trigger-info
}
public static ReturnT<String> runExecutor(TriggerParam triggerParam, String address){
ReturnT<String> runResult = null;
try {
//根据注册地址返回ExecutorBiz对象，这里是用map缓存起来，一个地址对应一个ExecutorBiz，也就是ExecutorBizClient实现类
//private static ConcurrentMap<String, ExecutorBiz> executorBizRepository = new ConcurrentHashMap<String, ExecutorBiz>();
ExecutorBiz executorBiz = XxlJobScheduler.getExecutorBiz(address);
//调用run方法
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));
}
return runResult;
}
@Override
public ReturnT<String> run(TriggerParam triggerParam) {
//通过post请求远程调用
return XxlJobRemotingUtil.postBody(addressUrl + "run", accessToken, timeout, triggerParam, String.class);
}
4.2 路由策略
XxlJob支持多种路由策略，除了分片广播，其余的具体的算法实现都是通过ExecutorRouter的实现类来实现的
第一个、最后一个、轮询、随机都比较好理解
- 一致性Hash：一致性Hash可以理解，就是Hash函数(hashcode%size)的size保持不变，从而保证了Hash函数的前后一致性
- 最不经常使用（LFU：Least Frequently Used）：Xxl-Job内部会有一个缓存，统计每个任务每个地址的使用次数，每次都选择使用次数最少的地址，这个缓存每隔24小时重置一次
- 最近最久未使用（LRU：Least Recently Used）：将地址存到LinkedHashMap中，它利用LinkedHashMap可以根据元素访问（get/put）顺序来给元素排序的特性，快速找到最近最久未使用（未访问）的节点
- 故障转移：调度中心都会去请求每个执行器，只要能接收到响应，说明执行器正常，那么任务就会交给这个执行器去执行
- 忙碌转移：调度中心也会去请求每个执行器，判断执行器是不是正在执行当前需要执行的任务（任务执行时间过长，导致上一次任务还没执行完，下一次又触发了），如果在执行，说明忙碌，不能用，否则就可以用
- 分片广播：XxlJob给每个执行器分配一个编号，从0开始递增，然后向所有执行器触发任务，告诉每个执行器自己的编号和总共执行器的数据，我们可以通过
XxlJobHelper#getShardIndex获取到编号，XxlJobHelper#getShardTotal获取到执行器的总数据量，分片广播就是将任务量分散到各个执行器，每个执行器只执行一部分任务，加快任务的处理
5、任务执行
前面在说执行器注册的时候有提到，当执行器接收到调度中心的请求时，会把请求交给ExecutorBizImpl来处理
@Override
protected void channelRead0(final ChannelHandlerContext ctx, FullHttpRequest msg) throws Exception {
bizThreadPool.execute(new Runnable() {
@Override
public void run() {
Object responseObj = process(httpMethod, uri, requestData, accessTokenReq);
}
});
}
//包括beat、idleBeat、run、kill、log请求都是通过ExecutorBizImpl来实现的
private Object process(HttpMethod httpMethod, String uri, String requestData, String accessTokenReq) {
try {
switch (uri) {
case "/beat":
return executorBiz.beat();
case "/idleBeat":
IdleBeatParam idleBeatParam = GsonTool.fromJson(requestData, IdleBeatParam.class);
return executorBiz.idleBeat(idleBeatParam);
case "/run":
TriggerParam triggerParam = GsonTool.fromJson(requestData, TriggerParam.class);
//最终会调用到ExecutorBizImpl的run方法实现
return executorBiz.run(triggerParam);
case "/kill":
KillParam killParam = GsonTool.fromJson(requestData, KillParam.class);
return executorBiz.kill(killParam);
case "/log":
LogParam logParam = GsonTool.fromJson(requestData, LogParam.class);
return executorBiz.log(logParam);
default:
return new ReturnT<String>(ReturnT.FAIL_CODE, "invalid request, uri-mapping(" + uri + ") not found.");
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
return new ReturnT<String>(ReturnT.FAIL_CODE, "request error:" + ThrowableUtil.toString(e));
}
}
executorBiz.run(TriggerParam triggerParam)，会根据传入的条件构建出IJobHandler和JobThread，最终会放到队列中等待触发
public ReturnT<String> run(TriggerParam triggerParam) {
//根据传入的JobId，加载旧的jobHandler和jobThread（如果一个任务已经执行过一次，会存入jobThreadRepository这个本地的Map里）
JobThread jobThread = XxlJobExecutor.loadJobThread(triggerParam.getJobId());
IJobHandler jobHandler = jobThread!=null?jobThread.getHandler():null;
String removeOldReason = null;
//根据不同的GlueType，构建不同的IJobHandler
GlueTypeEnum glueTypeEnum = GlueTypeEnum.match(triggerParam.getGlueType());
//如果类型是Bean的话，则会构建为MethodJobHandler
if (GlueTypeEnum.BEAN == glueTypeEnum) {
//直接从map缓存中取，在初始化jobHandler的那一步，都已经全部放入到map中
IJobHandler newJobHandler = XxlJobExecutor.loadJobHandler(triggerParam.getExecutorHandler());
//校验jobThread，如果满足以下条件则会终止
if (jobThread!=null && jobHandler != newJobHandler) {
removeOldReason = "change jobhandler or glue type, and terminate the old job thread.";
jobThread = null;
jobHandler = null;
}
//校验jobHandler
if (jobHandler == null) {
jobHandler = newJobHandler;
if (jobHandler == null) {
return new ReturnT<String>(ReturnT.FAIL_CODE, "job handler [" + triggerParam.getExecutorHandler() + "] not found.");
}
}
//如果类型是GLUE的话，则会构建为GlueJobHandler
} else if (GlueTypeEnum.GLUE_GROOVY == glueTypeEnum) {
//如果类型是Script的话，ScriptJobHandler
} else if (glueTypeEnum!=null && glueTypeEnum.isScript()) {
} else {
return new ReturnT<String>(ReturnT.FAIL_CODE, "glueType[" + triggerParam.getGlueType() + "] is not valid.");
}
//如果jobThread不为null，则根据阻塞策略进行相应的操作
if (jobThread != null) {
ExecutorBlockStrategyEnum blockStrategy = ExecutorBlockStrategyEnum.match(triggerParam.getExecutorBlockStrategy(), null);
//如果当前阻塞策略是DISCARD_LATER，也就是丢弃后续调度
if (ExecutorBlockStrategyEnum.DISCARD_LATER == blockStrategy) {
//jobThread正在运行或在触发队列中，则返回错误信息
if (jobThread.isRunningOrHasQueue()) {
return new ReturnT<String>(ReturnT.FAIL_CODE, "block strategy effect："+ExecutorBlockStrategyEnum.DISCARD_LATER.getTitle());
}
//如果当前阻塞策略是COVER_EARLY，也就是覆盖之前调度
} else if (ExecutorBlockStrategyEnum.COVER_EARLY == blockStrategy) {
//jobThread正在运行或在触发队列中，则终止旧的任务线程，并将任务线程设置为 null
if (jobThread.isRunningOrHasQueue()) {
removeOldReason = "block strategy effect：" + ExecutorBlockStrategyEnum.COVER_EARLY.getTitle();
jobThread = null;
}
//如果是单机串行SERIAL_EXECUTION，则放入队列中触发
} else {
// just queue trigger
}
}
//如果jobThread任务线程为null，注册一个新的任务线程
if (jobThread == null) {
jobThread = XxlJobExecutor.registJobThread(triggerParam.getJobId(), jobHandler, removeOldReason);
}
//将触发参数推送到任务线程的触发队列中，等待执行
ReturnT<String> pushResult = jobThread.pushTriggerQueue(triggerParam);
return pushResult;
}
在上面这个方法中，最终会把triggerParam放入TriggerQueue中，那么真正的任务执行是在哪里呢？还是看上面这段代码，XxlJobExecutor.registJobThread这个方法中，注册了一个新的jobThread，并且通过start方法启动了这个线程。
private static ConcurrentMap<Integer, JobThread> jobThreadRepository = new ConcurrentHashMap<Integer, JobThread>();
public static JobThread registJobThread(int jobId, IJobHandler handler, String removeOldReason){
//new了一个JobThread对象，并且调用start方法启动
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);
if (oldJobThread != null) {
oldJobThread.toStop(removeOldReason);
oldJobThread.interrupt();
}
return newJobThread;
}
调用了线程的start方法之后，JobThread的run方法就开始执行。
@Override
public void run() {
//toStop不为true则循环调用
while(!toStop){
running = false;
idleTimes++;
TriggerParam triggerParam = null;
try {
//从队列triggerQueue中取TriggerParam对象，最多等待三秒
triggerParam = triggerQueue.poll(3L, TimeUnit.SECONDS);
if (triggerParam!=null) {
running = true;
idleTimes = 0;
triggerLogIdSet.remove(triggerParam.getLogId());
//如果任务设置了超时时间，则new一个FutureTask对象异步执行任务，通过futureTask.get()方法，设置超时时间，然后捕获TimeoutException异常来实现
if (triggerParam.getExecutorTimeout() > 0) {
Thread futureThread = null;
try {
FutureTask<Boolean> futureTask = new FutureTask<Boolean>(new Callable<Boolean>() {
@Override
public Boolean call() throws Exception {
XxlJobContext.setXxlJobContext(xxlJobContext);
//通过该方法真正执行任务
handler.execute();
return true;
}
});
futureThread = new Thread(futureTask);
futureThread.start();
Boolean tempResult = futureTask.get(triggerParam.getExecutorTimeout(), TimeUnit.SECONDS);
} catch (TimeoutException e) {
XxlJobHelper.log("<br>----------- xxl-job job execute timeout");
XxlJobHelper.log(e);
// handle result
XxlJobHelper.handleTimeout("job execute timeout ");
} finally {
futureThread.interrupt();
}
} else {
//如果没有设置超时时间，直接执行任务
handler.execute();
}
//如果任务执行失败或结果丢失，调用XxlJobHelper.handleFail()方法进行处理
//如果任务执行成功，记录执行结果信息
if (XxlJobContext.getXxlJobContext().getHandleCode() <= 0) {
XxlJobHelper.handleFail("job handle result lost.");
} else {
String tempHandleMsg = XxlJobContext.getXxlJobContext().getHandleMsg();
tempHandleMsg = (tempHandleMsg!=null&&tempHandleMsg.length()>50000)
?tempHandleMsg.substring(0, 50000).concat("...")
:tempHandleMsg;
XxlJobContext.getXxlJobContext().setHandleMsg(tempHandleMsg);
}
XxlJobHelper.log("<br>----------- xxl-job job execute end(finish) -----------<br>----------- Result: handleCode="
+ XxlJobContext.getXxlJobContext().getHandleCode()
+ ", handleMsg = "
+ XxlJobContext.getXxlJobContext().getHandleMsg()
);
} else {
//空闲次数计数器idleTimes。如果idleTimes达到阈值（30 次），相当于有30次都没有执行对应的任务，就从执行器缓存中删除这个任务
if (idleTimes > 30) {
if(triggerQueue.size() == 0) {
XxlJobExecutor.removeJobThread(jobId, "excutor idel times over limit.");
}
}
}
} catch (Throwable e) {
if (toStop) {
XxlJobHelper.log("<br>----------- JobThread toStop, stopReason:" + stopReason);
}
//处理异常信息
StringWriter stringWriter = new StringWriter();
e.printStackTrace(new PrintWriter(stringWriter));
String errorMsg = stringWriter.toString();
XxlJobHelper.handleFail(errorMsg);
XxlJobHelper.log("<br>----------- JobThread Exception:" + errorMsg + "<br>----------- xxl-job job execute end(error) -----------");
} finally {
if(triggerParam != null) {
//如果任务线程未被停止，将推送回调信息push到回调线程TriggerCallbackThread
if (!toStop) {
TriggerCallbackThread.pushCallBack(new HandleCallbackParam(
triggerParam.getLogId(),
triggerParam.getLogDateTime(),
XxlJobContext.getXxlJobContext().getHandleCode(),
XxlJobContext.getXxlJobContext().getHandleMsg() )
);
} else {
//如果任务线程被停止，则推送线程停止信息到回调线程TriggerCallbackThread
TriggerCallbackThread.pushCallBack(new HandleCallbackParam(
triggerParam.getLogId(),
triggerParam.getLogDateTime(),
XxlJobContext.HANDLE_CODE_FAIL,
stopReason + " [job running, killed]" )
);
}
}
}
}
//如果线程被停止后触发队列不为空，则推送线程停止信息到回调线程
while(triggerQueue !=null && triggerQueue.size()>0){
TriggerParam triggerParam = triggerQueue.poll();
if (triggerParam!=null) {
// is killed
TriggerCallbackThread.pushCallBack(new HandleCallbackParam(
triggerParam.getLogId(),
triggerParam.getLogDateTime(),
XxlJobContext.HANDLE_CODE_FAIL,
stopReason + " [job not executed, in the job queue, killed.]")
);
}
}
}
- JobThread会将任务执行的结果发送到一个内存队列中
- 执行器启动的时候会开启一个处发送任务执行结果的线程：TriggerCallbackThread
- 这个线程会不停地从队列中获取所有的执行结果，将执行结果批量发送给调度中心
- 调用中心接收到请求时，会根据执行的结果修改这次任务的执行状态和进行一些后续的事，比如失败了是否需要重试，是否有子任务需要触发等等
三、推荐阅读
XXL-JOB官网
xxl-job架构原理讲解
XXL-JOB调度算法
xxl-job源码解读