xxjob代码执行过程

文章目录

- 简介
- - xxl-job定时任务的种类
  - xxl-job相关的数据表
- 调度服务
- - JobScheduleHelper.start().run任务执行列表获取
  - 执行任务
- 执行器
- - 执行方法
- 参考资料

简介

调度中心：
负责管理调度信息，按照调度配置发出调度请求，自身不承担业务代码。调度系统与任务解耦，提高了系统可用性和稳定性，同时调度系统性能不再受限于任务模块；
支持可视化、简单且动态的管理调度信息，包括任务新建，更新，删除，GLUE开发和任务报警等，所有上述操作都会实时生效，同时支持监控调度结果以及执行日志，支持执行器Failover，支持创建执行器等功能。
执行模块（执行器）：
负责接收调度请求并执行任务逻辑。任务模块专注于任务的执行等操作，开发和维护更加简单和高效；接收“调度中心”的执行请求、终止请求和日志请求等。

特性
xxl-job的特性有很多，官网上有详细的介绍，这里我会介绍几个重要的特性：

不分片：使用数据库锁在集群模式下同一时刻只有一个调度中心处理任务调度
简单：支持通过Web页面对任务进行CRUD操作，操作简单，一分钟上手；
动态：支持动态修改任务状态、启动/停止任务，以及终止运行中的任务，都是即时生效的。
调度中心HA（中心式）：调度采用中心式设计，“调度中心”自研调度组件并支持集群部署，可保证调度中心HA；
执行器HA（分布式）：任务分布式执行，任务”执行器”支持集群部署，可保证任务执行HA；
调度过期策略：调度中心错过调度时间的补偿处理策略：包括：忽略，立即补偿触发一次等；
阻塞处理策略：调度过于密集执行器来不及处理时的处理策略，策略包括：单机串行（默认）、丢弃后续调度、覆盖之前的调用。
任务超时控制：支持自定义任务超时时间，任务运行超时将会主动中断任务；

最大的特点就是不分片，相对其他分布式任务组件，这个组件是最简单解决方式同时又能保证调度任务的一致性

xxl-job定时任务的种类

xxll0job支持java、groovy、脚本（Shell、Python、PHP、NodeJs、PowerShell）的定时任务

xxl-job相关的数据表

xxl-job将任务信息以及日志信息持久化到数据表中，这个就保证了可以动态的添加删除任务。

xxl_job_lock：任务调度锁表，在线程查询任务信息时会调用上锁。
xxl_job_group：执行器信息表，维护任务执行器信息；
xxl_job_info：调度扩展信息表：用于保存XXL-JOB调度任务的扩展信息，如任务分组、任务名、机器地址、执行器、执行入参和报警邮件等等；
xxl_job_log：调度日志表：用于保存XXL-JOB任务调度的历史信息，如调度结果、执行结果、调度入参、调度机器和执行器等等；
xxl_job_log_report：调度日志报表：用户存储XXL-JOB任务调度日志的报表，调度中心报表功能页面会用到；
xxl_job_logglue：任务GLUE日志：用于保存GLUE更新历史，用于支持GLUE的版本回溯功能；
xxl_job_registry：执行器注册表，维护在线的执行器和调度中心机器地址信息；
xxl_job_user：系统用户表；

调度服务

xxl-job集群部署时，如何避免多个服务器同时调度任务？
定时任务是如何实现的？

JobScheduleHelper.start().run任务执行列表获取

JobScheduleHelper.start().run()使用while()循环不断的查询数据库中<5秒的任务列表

public class JobScheduleHelper {
 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(){
        //扫描任务列表线程
        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.");

                    //循环扫描任务：重点部分
                while (!scheduleThreadToStop) {
                     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();


                        // 1、pre read
                        long nowTime = System.currentTimeMillis();
                        //关键部分，预读出 当前时间+5s(PRE_READ_MS=5s) 的所有触发任务的时间小于这个时间预读出时间的任务
                        List<XxlJobInfo> scheduleList = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().scheduleJobQuery(nowTime + PRE_READ_MS, preReadCount);
                        if (scheduleList!=null && scheduleList.size()>0) {
                            // 2、push time-ring
                                // time-ring jump
                                //遍历所有任务，过滤出所有过期任务
                                //过期任务的判断方式 当前时间>任务下一次触发时间+空闲间隔周期5s (在这段时间内任务还没触发的话说明任务超时过期了，既在5s的误差内还没出发任务),并按照策略执行
                                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());

                                    //2.1-1、misfire match  查询当前任务的过期策略
                                    MisfireStrategyEnum misfireStrategyEnum = MisfireStrategyEnum.match(jobInfo.getMisfireStrategy(), MisfireStrategyEnum.DO_NOTHING);
                                    if (MisfireStrategyEnum.FIRE_ONCE_NOW == misfireStrategyEnum) {
                                        // FIRE_ONCE_NOW策略则过期立即执行一次
                                        JobTriggerPoolHelper.trigger(jobInfo.getId(), TriggerTypeEnum.MISFIRE, -1, null, null, null);
                                        logger.debug(">>>>>>>>>>> xxl-job, schedule push trigger : jobId = " + jobInfo.getId() );
                                    }

                                    // 2.1-2、fresh 刷新下次执行时间到数据库中
                                    refreshNextValidTime(jobInfo, new Date());
                                //否则不是过期任务  任务下一次触发时间+空闲间隔周期5s>当前时间>下一次任务触发时间  （在允许的误差之内都不算过期任务）
                                } else if (nowTime > jobInfo.getTriggerNextTime()) {
                                    // 2.2、trigger-expire < 5s：direct-trigger && make next-trigger-time
                                    // 2.2-1、trigger
                                    JobTriggerPoolHelper.trigger(jobInfo.getId(), TriggerTypeEnum.CRON, -1, null, null, null);
                                    logger.debug(">>>>>>>>>>> xxl-job, schedule push trigger : jobId = " + jobInfo.getId() );

                                    // 2.1-2、fresh  刷新下次执行时间到数据库中
                                    refreshNextValidTime(jobInfo, new Date());

                                    // 任务状态有效&&执行时间在5秒以内的任务放入时间环中
                                    if (jobInfo.getTriggerStatus()==1 && nowTime + PRE_READ_MS > jobInfo.getTriggerNextTime()) {

                                        // 1.获取剩余秒数
                                        int ringSecond = (int)((jobInfo.getTriggerNextTime()/1000)%60);

                                        // 2、把剩余秒数作为key放入时间轮中
                                        pushTimeRing(ringSecond, jobInfo.getId());

                                        // 3、刷新任务的下次执行时间
                                        refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime()));

                                    }

                                } else {
                                    // 2.3 即将促发的任务放入时间轮中5秒钟

                                    // 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 {

                        // 最后释放数据库锁，重置自动提交，关闭连接
                        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

                    }
                    long cost = System.currentTimeMillis()-start;

                }

                logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper#scheduleThread stop");
            }
        });
        scheduleThread.setDaemon(true);
        scheduleThread.setName("xxl-job, admin JobScheduleHelper#scheduleThread");
        scheduleThread.start();



        //执行任务线程：从时间轮中获取任务并执行
        ringThread = new Thread(new Runnable() {
            @Override
            public void run() {

                //休息到下一秒=一秒-当前毫秒=剩下的毫秒数：只是启动时执行一次，以确保while是整秒执行
                try {
                    TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis()%1000 );
                } catch (InterruptedException e) {
                    if (!ringThreadToStop) {
                        logger.error(e.getMessage(), e);
                    }
                }

                //轮询时间轮
                while (!ringThreadToStop) {
                    try {
                        synchronized (ringData){
                            if(ringData.isEmpty()){
                                     ringData.wait();
                            }
                        }

                        // 处理前2秒的数据
                        List<Integer> ringItemData = new ArrayList<>();
                        int nowSecond = Calendar.getInstance().get(Calendar.SECOND); //当前秒数  
                        // 避免处理耗时太长，跨过刻度，向前校验一个刻度=移除前一个刻度的k-v值并jobId放入执行列表
                        for (int i = 0; i < 2; i++) {
                            List<Integer> tmpData = ringData.remove( (nowSecond+60-i)%60 );
                            if (tmpData != null) {
                                ringItemData.addAll(tmpData);
                            }
                        }

                        // 促发执行任务
                        logger.debug(">>>>>>>>>>> xxl-job, time-ring beat : " + nowSecond + " = " + Arrays.asList(ringItemData) );
                        if (ringItemData.size() > 0) {
                            // do trigger
                            for (int jobId: ringItemData) {
                                // 执行任务
                                JobTriggerPoolHelper.trigger(jobId, TriggerTypeEnum.CRON, -1, null, null, null);
                            }
                            // 清空需要执行任务的列表
                            ringItemData.clear();
                        }
                    } catch (Exception e) {
                        if (!ringThreadToStop) {
                            logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#ringThread error:{}", e);
                        }
                    }

                    //以上内容执行完可能还没到达下一秒钟整点时间，那就休息到到达下一秒整点时的毫秒数时间
                    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();


    }//start()方法结束

}

任务调度器工作流程：
1.任务线程scheduleThread通过数据库的for update行锁来保证多个调度中心集群在同一时间内只有一个调度中心在调度任务
2.周期性的遍历所有的jobInfo这个表，读取触发时间小于nowtime+5s这个时间之前的所有任务，然后进行引入以下触发机制判断

三种触发任务机制：

1）过了执行时间5S多还没执行的nowtime-TriggerNextTime()>PRE_READ_MS(5s) 既超过有误差5S外的，则查看当前任务的失效调度策略，若为立即重试一次，则立即触发调度任务，且触发类型为misfire
2)过了执行时间5S内还没执行的nowtime-TriggerNextTime()<PRE_READ_MS(5s) 既没有超过有效误差5S内，则立即调度调度任务
3)执行时间即将来5S内(因为他列表查的就是将来5S内的)还没执行的nowtime<TriggerNextTime() 则说明这个任务马上就要触发了，放到一个时间轮上（https://blog.csdn.net/zalu9810/article/details/113396131），

3.随后将快要触发的任务放到时间轮上（Map<int,List>结构实现固定60个key），时间轮由key(将要触发的时间s)，value(在当前触发s的所有任务id集合)，然后更新这个任务的下一次触发时间到数据库中

4.第二个线程ringThread对这个时间轮的任务遍历，周期在1s之内周期的扫描这个时间轮，然后执行调度任务

原文链接：https://blog.csdn.net/s6056826a/article/details/113446126

xxl-job通过mysql悲观锁实现分布式锁，从而避免多个服务器同时调度任务：

通过setAutoCommit(false)，关闭自动提交
通过select lock for update语句，其他事务无法获取到锁，显示排她锁。
最后在finally块中commit()提交事务释放for update的排他锁，并且setAutoCommit

执行任务

在上面的调度任务中最终调用的是执行器执行任务方法，具体代码如下

//在调度代码中我们看到:执行任务
JobTriggerPoolHelper.trigger(jobId, TriggerTypeEnum.CRON, -1, null, null, null);

//JobTriggerPoolHelper.trigger方法最终调在线程池中调用的是XxlJobTrigger.trigger
JobTriggerPoolHelper.trigger(){
ThreadPoolExecutor triggerPool_ = fastTriggerPool;
 triggerPool_.execute(new Runnable() {
 //构建参数后执行 XxlJobTrigger.trigger
 XxlJobTrigger.trigger(jobId, triggerType, failRetryCount, executorShardingParam, executorParam, addressList);
 }
}

//XxlJobTrigger.trigger
XxlJobTrigger.trigger(){
 // load data，加载任务信息
 XxlJobInfo jobInfo = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().loadById(jobId);
 if (executorParam != null) {
 jobInfo.setExecutorParam(executorParam);
 }
 processTrigger(group, jobInfo, finalFailRetryCount, triggerType, shardingParam[0], shardingParam[1]);

}

//核心逻辑在processTrigger中
//代码位置： com.xxl.job.admin.core.trigger.XxlJobTrigger#processTrigger
private static void processTrigger(XxlJobGroup group, XxlJobInfo jobInfo, int finalFailRetryCount, TriggerTypeEnum triggerType, int index, int total){

//初始化trigger-param
 TriggerParam triggerParam = new TriggerParam();
 triggerParam.setJobId(jobInfo.getId());
//执行路由策略：index\分片\路由策略\获取任务地址
routeAddressResult = executorRouteStrategyEnum.getRouter().route(triggerParam, group.getRegistryList());
//运行执行器
 triggerResult = runExecutor(triggerParam, address);
 //日志处理，代码省略
}

//代码位置：com.xxl.job.admin.core.trigger.XxlJobTrigger#runExecutor
public static ReturnT<String> runExecutor(TriggerParam triggerParam, String address){
        ReturnT<String> runResult = null;
        try {
            //获取执行器:通过地址从executorBizRepository（map）获取ExecutorBiz
            ExecutorBiz executorBiz = XxlJobScheduler.getExecutorBiz(address);
            //执行任务并返回结果:executorBiz是抽象类，根据具体类型执行本地任务/执行器任务
            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;
}

ExecutorBiz接口有两个实现，分别是ExecutorBizClient（执行器客户端）、ExecutorBizImpl（执行器服务端），

ExecutorBizClien类就是客户端操作远程服务的任务
ExecutorBizImpl就是服务端操作本地任务

ExecutorBiz接口有beat（心跳检测）、idleBeat（空闲检测）、run（执行任务）、kill（停止任务）、log（打印日志）这些方法。

我们看看ExecutorBizClien的run方法：

//代码位置：com.xxl.job.core.biz.client.ExecutorBizClient#run
public ReturnT<String> run(TriggerParam triggerParam) {
//调用http的POST请求发送触发参数触发服务端的任务执行，然后将结果返回给客户端。请求的地址为addressUrl + "run"，当客户端发送请求以后，ExecutorBizImpl的run方法将会接收请求处理，然后将处理的结果返回
        return XxlJobRemotingUtil.postBody(addressUrl + "run", accessToken, timeout, triggerParam, String.class);
}

客户端EmbedServer类的内部类EmbedHttpServerHandler的process方法会调用ExecutorBizImpl类的run方法通过Netty Http调用执行器的EmbedServer的process方法

执行器

根据提供的样例，可以发布为单独的服务，和调度器分开部署。

执行器启动时，会初始化一个EmbedServer类，该类的start方法会启动netty服务器。netty服务器会接收客户端发送过来的http请求，当接收到触发请求（请求路径是/run）会交给EmbedServer类的process方法处理，process方法将会调用ExecutorBizImpl的run方法处理客户端发送的触发请求，process方法接收触发请求的代码如下图所示:

private Object process(HttpMethod httpMethod, String uri, String requestData, String accessTokenReq){

if(){//....
}else if ("/run".equals(uri)) {
                    TriggerParam triggerParam = GsonTool.fromJson(requestData, TriggerParam.class);
                    return executorBiz.run(triggerParam);
                }
 }

ExecutorBizImpl的run方法处理流程大致如下：

加载任务处理器与任务执行线程，校验任务处理器与任务执行线程。
执行阻塞策略
注册任务
保存触发参数到缓存

run方法首先根据任务id从缓存jobThreadRepository（map）中获取任务执行线程jobThread，任务执行线程jobThread保存着任务处理器jobHandler，然后进行校验任务执行线程以及任务处理器。

接下来检验任务执行线程以及任务处理器，就是按照Java、groovy、脚本分别进行校验

xxl-job有三种阻塞策略，分别为SERIAL_EXECUTION（并行）、DISCARD_LATER（丢弃）、COVER_EARLY（覆盖之前的）。当阻塞策略为丢弃，则判断该执行线程是否正在执行，如果是则直接返回结果，不再往下执行任务了。当阻塞策略为覆盖之前的，则判断执行线程是否正在执行，如果是则杀掉原来的执行线程。如果阻塞策略是并行，则不做什么

注册任务线程：如果任务线程等于null，注册任务线程并启动线程。registJobThread方法首先新建一个任务线程，并调用newJobThread的start方法启动任务线程。然后加入jobThreadRepository进行缓存，当旧的oldJobThread不等于null，则停止掉旧的任务线程。

在服务端执行的流程中，将任务交给任务线程池JobThread执行，JobThread的run方法主要做了几件事：

处理器的初始化
任务的执行
销毁清理工作

while()循环从triggerQueue队列中弹出触发参数，如果存在执行超时时间并大于0，则在规定的时间异步调用handler的execute方法执行任务，否则立即调用handler的execute方法执行任务。

while()循环如果任务停止了，需要将队列中所有的触发删除（所有定时任务删除）

执行方法

定时任务执行过程调用处理器的init（初始化方法）、execute（执行方法）、destroy（销毁）方法，这些方法是由IJobHandler抽象类的实现类实现的：

IJobHandler抽象类有三个子类，

GlueJobHandler是执行groovy的处理器，MethodJobHandler是执行java的处理器
ScriptJobHandler是执行脚本（Pyhotn、PHP、NodeJS、Shell等）的处理器。
MethodJobHandler是处理java定时任务的方法，当我们用java开发了定时任务方法，然后用@XxlJob注解修饰方法，就可以调度该定时任务方法了

我们看看MethodJobHandler的execute方法是如何执行定时任务方法的。

//代码位置：com.xxl.job.core.handler.impl.MethodJobHandler#execute
public void execute() throws Exception {
        Class<?>[] paramTypes = method.getParameterTypes();
        if (paramTypes.length > 0) {
            method.invoke(target, new Object[paramTypes.length]);       // method-param can not be primitive-types
        } else {
            method.invoke(target);
        }
}

MethodJobHandler的execute方法利用反射，获取定时任务的method，然后利用invoke执行定时任务方法。

GlueJobHandler是执行groovy的处理器，在admin界面的idea界面上写好groovy保存在数据库，会调用GlueJobHandler类的execute方法执行，groovy是一种基于JVM的开发语言，groovy 代码能够与 Java 代码很好地结合，也能用于扩展现有代码。GlueFactory类的loadNewInstance方法将写好的groovy加载解析为写好的groovy代码，并返回IJobHandler，然后将IJobHandler传进GlueJobHandler构造器中新建GlueJobHandler对象。

GlueJobHandler的execute方法就是调用GlueFactory工厂类创建的IJobHandler的execute方法。

ScriptJobHandler的execute方法有些不重要的代码被省略。主要有几个重要的流程：获取脚本执行命令、保存脚本命令到文件、执行脚本。

ScriptUtil的markScriptFile方法，将脚本定时任务代码保存在名字为jobId_glueUpdatetime_suffix中，jobId为任务id，glueUpdatetime为脚本更新时间、suffix为脚本后缀，如采用python写定时任务时，保存在类似666-123456789.py文件中。

执行脚本ScriptUtil的execToFile方法，execToFile方法是利用Runtime.getRuntime().exec()方法在java程序里运行脚本程序。Runtime.getRuntime().exec()方法会将执行命令发送给操作系统，然后等待操作系统运行程序的结果返回，如Runtime.getRuntime().exec()方法给操作系统发送pyhton hello.py命令，这样就会执行python脚本，并等待python脚本的运行结果的返回

这里直接参考：xxl-job定时任务执行流程分析-任务执行

https://zhuanlan.zhihu.com/p/438464389