XXL-JOB原理--任务执行时间轮（六）

太过爱你忘了你带给我的痛 2023-01-05 14:16 209阅读 0赞

#### 1、介绍 ####

之前 XXL-JOB 任务执行是通过 Quartz来进行任务管理触发的，在之前的博客 [《Quartz任务调度框架–任务执行流程》][Quartz] 我们进行了任务执行的流程介绍，目前 XXL-JOB 任务执行已经摒弃 Quartz 框架，目前通过时间轮方式来管理任务触发任务。

#### 2、任务执行 ####

XXL-JOB 任务执行中启动了两个线程：  
（1）线程 scheduleThread 运行中不断的从任务表中查询 查询近 5000 毫秒(5秒)中要执行的任务，如果当前时间大于任务接下来要执行的时间则立即执行，否则将任务执行时间除以 1000 变为秒之后再与 60 求余添加到时间轮中。  
（2）XXL-JOB 时间轮实现方式比较简单，就是一个 Map 结构数据，key值0-60，value是任务ID列表  
Map<Integer, List> ringData  
（3）线程 ringThread 运行中不断根据当前时间求余从 时间轮 ringData 中获取任务列表，取出任务之后执行任务。

public class JobScheduleHelper {
        private static Logger logger = LoggerFactory.getLogger(JobScheduleHelper.class);
    
        private static JobScheduleHelper instance = new JobScheduleHelper();
        public static JobScheduleHelper getInstance(){
            return instance;
        }
        // 任务间隔大小
        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;
        // 时间轮，key 0-60，value 任务ID列表，两个线程同时处理这个对象
        private volatile static Map<Integer, List<Integer>> ringData = new ConcurrentHashMap<>();
    
        public void start(){
    
            // schedule thread
            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.");
    
                    // pre-read count: treadpool-size * trigger-qps (each trigger cost 50ms, qps = 1000/50 = 20)
                    // 每次读取到任务个数
                    int preReadCount = (XxlJobAdminConfig.getAdminConfig().getTriggerPoolFastMax() + XxlJobAdminConfig.getAdminConfig().getTriggerPoolSlowMax()) * 20;
    
                    while (!scheduleThreadToStop) {
    
                        // Scan Job
                        long start = System.currentTimeMillis();
    
                        Connection conn = null;
                        Boolean connAutoCommit = null;
                        PreparedStatement preparedStatement = null;
    
                        boolean preReadSuc = true;
                        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();
    
                            // tx start
    
                            // 1、pre read
                            long nowTime = System.currentTimeMillis();
                            // 查询 当前时间 + 5000 毫秒，就是接下来 5 秒要执行到任务
                            List<XxlJobInfo> scheduleList = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().scheduleJobQuery(nowTime + PRE_READ_MS, preReadCount);
                            if (scheduleList!=null && scheduleList.size()>0) {
                                // 2、push time-ring
                                for (XxlJobInfo jobInfo: scheduleList) {
    
                                    // time-ring jump
                                    // 如果当前时间大于要接下来执行到时间 + 5 秒
                                    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());
    
                                        // 1、misfire match
                                        // 如果每次都有执行则立即触发执行
                                        MisfireStrategyEnum misfireStrategyEnum = MisfireStrategyEnum.match(jobInfo.getMisfireStrategy(), MisfireStrategyEnum.DO_NOTHING);
                                        if (MisfireStrategyEnum.FIRE_ONCE_NOW == misfireStrategyEnum) {
                                            // FIRE_ONCE_NOW 》 trigger
                                            JobTriggerPoolHelper.trigger(jobInfo.getId(), TriggerTypeEnum.MISFIRE, -1, null, null, null);
                                            logger.debug(">>>>>>>>>>> xxl-job, schedule push trigger : jobId = " + jobInfo.getId() );
                                        }
    
                                        // 2、fresh next
                                        // 刷新接下来要执行时间
                                        refreshNextValidTime(jobInfo, new Date());
    
                                    } else if (nowTime > jobInfo.getTriggerNextTime()) {
                                        // 2.2、trigger-expire < 5s：direct-trigger && make next-trigger-time
    
                                        // 1、trigger
                                        // 如果当前时间大于接下来要执行到时间则立即触发执行
                                        JobTriggerPoolHelper.trigger(jobInfo.getId(), TriggerTypeEnum.CRON, -1, null, null, null);
                                        logger.debug(">>>>>>>>>>> xxl-job, schedule push trigger : jobId = " + jobInfo.getId() );
    
                                        // 2、fresh next
                                        // 刷新下次执行时间
                                        refreshNextValidTime(jobInfo, new Date());
    
                                        // next-trigger-time in 5s, pre-read again
                                        if (jobInfo.getTriggerStatus()==1 && nowTime + PRE_READ_MS > jobInfo.getTriggerNextTime()) {
    
                                            // 1、make ring second
                                            // 如果接下来 5 秒内还执行则直接放到时间轮中
                                            int ringSecond = (int)((jobInfo.getTriggerNextTime()/1000)%60);
    
                                            // 2、push time ring
                                            pushTimeRing(ringSecond, jobInfo.getId());
    
                                            // 3、fresh next
                                            // 刷新下次执行时间
                                            refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime()));
    
                                        }
    
                                    } else {
                                        // 2.3、trigger-pre-read：time-ring trigger && make next-trigger-time
    
                                        // 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 {
    
                            // commit
                            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
                            if (null != preparedStatement) {
                                try {
                                    preparedStatement.close();
                                } catch (SQLException e) {
                                    if (!scheduleThreadToStop) {
                                        logger.error(e.getMessage(), e);
                                    }
                                }
                            }
                        }
                        long cost = System.currentTimeMillis()-start;
    
    
                        // Wait seconds, align second
                        if (cost < 1000) {  // scan-overtime, not wait
                            try {
                                // pre-read period: success > scan each second; fail > skip this period;
                                TimeUnit.MILLISECONDS.sleep((preReadSuc?1000:PRE_READ_MS) - System.currentTimeMillis()%1000);
                            } catch (InterruptedException e) {
                                if (!scheduleThreadToStop) {
                                    logger.error(e.getMessage(), e);
                                }
                            }
                        }
    
                    }
    
                    logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper#scheduleThread stop");
                }
            });
            scheduleThread.setDaemon(true);
            scheduleThread.setName("xxl-job, admin JobScheduleHelper#scheduleThread");
            scheduleThread.start();
    
    
            // ring thread
            ringThread = new Thread(new Runnable() {
                @Override
                public void run() {
    
                    // align second
                    try {
                        TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis()%1000 );
                    } catch (InterruptedException e) {
                        if (!ringThreadToStop) {
                            logger.error(e.getMessage(), e);
                        }
                    }
    
                    while (!ringThreadToStop) {
    
                        try {
                            // second data
                            List<Integer> ringItemData = new ArrayList<>();
                            int nowSecond = Calendar.getInstance().get(Calendar.SECOND);   // 避免处理耗时太长，跨过刻度，向前校验一个刻度；
                            // 获取最近1秒和 2秒要执行到任务
                            for (int i = 0; i < 2; i++) {
                                List<Integer> tmpData = ringData.remove( (nowSecond+60-i)%60 );
                                if (tmpData != null) {
                                    ringItemData.addAll(tmpData);
                                }
                            }
    
                            // ring trigger
                            logger.debug(">>>>>>>>>>> xxl-job, time-ring beat : " + nowSecond + " = " + Arrays.asList(ringItemData) );
                            if (ringItemData.size() > 0) {
                                // do trigger
                                for (int jobId: ringItemData) {
                                    // do trigger
                                    // 执行任务
                                    JobTriggerPoolHelper.trigger(jobId, TriggerTypeEnum.CRON, -1, null, null, null);
                                }
                                // clear
                                ringItemData.clear();
                            }
                        } catch (Exception e) {
                            if (!ringThreadToStop) {
                                logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#ringThread error:{}", e);
                            }
                        }
    
                        // next second, align second
                        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();
        }
    
        private void refreshNextValidTime(XxlJobInfo jobInfo, Date fromTime) throws Exception {
            Date nextValidTime = generateNextValidTime(jobInfo, fromTime);
            if (nextValidTime != null) {
                jobInfo.setTriggerLastTime(jobInfo.getTriggerNextTime());
                jobInfo.setTriggerNextTime(nextValidTime.getTime());
            } else {
                jobInfo.setTriggerStatus(0);
                jobInfo.setTriggerLastTime(0);
                jobInfo.setTriggerNextTime(0);
                logger.warn(">>>>>>>>>>> xxl-job, refreshNextValidTime fail for job: jobId={}, scheduleType={}, scheduleConf={}",
                        jobInfo.getId(), jobInfo.getScheduleType(), jobInfo.getScheduleConf());
            }
        }
    
        private void pushTimeRing(int ringSecond, int jobId){
            // push async ring
            List<Integer> ringItemData = ringData.get(ringSecond);
            if (ringItemData == null) {
                ringItemData = new ArrayList<Integer>();
                ringData.put(ringSecond, ringItemData);
            }
            ringItemData.add(jobId);
    
            logger.debug(">>>>>>>>>>> xxl-job, schedule push time-ring : " + ringSecond + " = " + Arrays.asList(ringItemData) );
        }
    
        public void toStop(){
    
            // 1、stop schedule
            scheduleThreadToStop = true;
            try {
                TimeUnit.SECONDS.sleep(1);  // wait
            } catch (InterruptedException e) {
                logger.error(e.getMessage(), e);
            }
            if (scheduleThread.getState() != Thread.State.TERMINATED){
                // interrupt and wait
                scheduleThread.interrupt();
                try {
                    scheduleThread.join();
                } catch (InterruptedException e) {
                    logger.error(e.getMessage(), e);
                }
            }
    
            // if has ring data
            boolean hasRingData = false;
            if (!ringData.isEmpty()) {
                for (int second : ringData.keySet()) {
                    List<Integer> tmpData = ringData.get(second);
                    if (tmpData!=null && tmpData.size()>0) {
                        hasRingData = true;
                        break;
                    }
                }
            }
            if (hasRingData) {
                try {
                    TimeUnit.SECONDS.sleep(8);
                } catch (InterruptedException e) {
                    logger.error(e.getMessage(), e);
                }
            }
    
            // stop ring (wait job-in-memory stop)
            ringThreadToStop = true;
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                logger.error(e.getMessage(), e);
            }
            if (ringThread.getState() != Thread.State.TERMINATED){
                // interrupt and wait
                ringThread.interrupt();
                try {
                    ringThread.join();
                } catch (InterruptedException e) {
                    logger.error(e.getMessage(), e);
                }
            }
    
            logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper stop");
        }
    
    
        // ---------------------- tools ----------------------
        public static Date generateNextValidTime(XxlJobInfo jobInfo, Date fromTime) throws Exception {
            ScheduleTypeEnum scheduleTypeEnum = ScheduleTypeEnum.match(jobInfo.getScheduleType(), null);
            if (ScheduleTypeEnum.CRON == scheduleTypeEnum) {
                Date nextValidTime = new CronExpression(jobInfo.getScheduleConf()).getNextValidTimeAfter(fromTime);
                return nextValidTime;
            } else if (ScheduleTypeEnum.FIX_RATE == scheduleTypeEnum /*|| ScheduleTypeEnum.FIX_DELAY == scheduleTypeEnum*/) {
                return new Date(fromTime.getTime() + Integer.valueOf(jobInfo.getScheduleConf())*1000 );
            }
            return null;
        }
    
    }

通过以上分析，我们发现 XXL-JOB 通过一个简单到时间轮就可以完成一个任务到倒计时执行操作。

[Quartz]: https://tianjunwei.blog.csdn.net/article/details/82972167