Inside XXL-JOB: A Deep Dive into Its Distributed Scheduling Architecture

Architecture Diagram

Source Code Analysis

Project Structure

XXL‑JOB consists of three modules:

xxl-job-admin : the scheduling center that manages jobs, executors, and logs.

xxl-job-core : shared dependencies for the admin and executors.

xxl-job-executor-samples : sample executors that run actual tasks.

Scheduling Center

Scheduling Center Flow

The admin module is a Spring Boot application. When XxlJobAdminApplication starts, Spring creates beans; XxlJobAdminConfig implements InitializingBean, so its afterPropertiesSet method launches a series of internal threads for the scheduler.

Code Walk‑through

1. Internationalization initialization

private void initI18n(){
    for (ExecutorBlockStrategyEnum item : ExecutorBlockStrategyEnum.values()) {
        item.setTitle(I18nUtil.getString("jobconf_block_".concat(item.name())));
    }
}

The method loads language‑specific resource files and updates the display titles of executor block‑strategy enums.

2. Trigger thread‑pool initialization JobTriggerPoolHelper.start() creates two pools: a fast pool ( fastTriggerPool) and a slow pool ( slowTriggerPool). The fast pool is used by default; if a job times out more than ten times within a minute, the slow pool is selected.

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;
    }
    XxlJobTrigger.trigger(jobId, triggerType, failRetryCount,
                         executorShardingParam, executorParam, addressList);
}

The core trigger logic resides in XxlJobTrigger.trigger (shown later).

3. Registry and monitoring thread‑pool initialization JobRegistryHelper.start() creates a pool registryOrRemoveThreadPool and a daemon thread registryMonitorThread.

registryOrRemoveThreadPool handles executor registration and removal.

public ReturnT<String> registry(RegistryParam registryParam) {
    if (!StringUtils.hasText(registryParam.getRegistryGroup()) ||
        !StringUtils.hasText(registryParam.getRegistryKey()) ||
        !StringUtils.hasText(registryParam.getRegistryValue())) {
        return new ReturnT<String>(ReturnT.FAIL_CODE, "Illegal Argument.");
    }
    registryOrRemoveThreadPool.execute(() -> {
        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());
            freshGroupRegistryInfo(registryParam);
        }
    });
    return ReturnT.SUCCESS;
}

When an executor starts, it registers itself via this method, persisting data to the xxl_job_registry table.

public ReturnT<String> registryRemove(RegistryParam registryParam) {
    if (!StringUtils.hasText(registryParam.getRegistryGroup()) ||
        !StringUtils.hasText(registryParam.getRegistryKey()) ||
        !StringUtils.hasText(registryParam.getRegistryValue())) {
        return new ReturnT<String>(ReturnT.FAIL_CODE, "Illegal Argument.");
    }
    registryOrRemoveThreadPool.execute(() -> {
        int ret = XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao()
            .registryDelete(registryParam.getRegistryGroup(),
                             registryParam.getRegistryKey(),
                             registryParam.getRegistryValue());
        if (ret > 0) {
            freshGroupRegistryInfo(registryParam);
        }
    });
    return ReturnT.SUCCESS;
}

When an executor shuts down, it calls registryRemove to delete its entry. registryMonitorThread continuously scans registered executors, removes entries that have not reported for 90 seconds (three missed 30‑second heartbeats), and refreshes the address_list field of xxl_job_group with currently alive executors. 4. Job‑failure monitoring thread JobFailMonitorHelper.start() launches a daemon thread that repeatedly fetches up to 1,000 failed job logs, locks each log (setting alarm_status to –1), retries if the job’s retry count is positive, and finally triggers an alarm via the configured JobAlarmer .

public void start(){
    monitorThread = new Thread(() -> {
        while (!toStop) {
            try {
                List<Long> failLogIds = XxlJobAdminConfig.getAdminConfig()
                    .getXxlJobLogDao().findFailJobLogIds(1000);
                if (failLogIds != null && !failLogIds.isEmpty()) {
                    for (long failLogId : failLogIds) {
                        int lockRet = XxlJobAdminConfig.getAdminConfig()
                            .getXxlJobLogDao().updateAlarmStatus(failLogId, 0, -1);
                        if (lockRet < 1) continue;
                        XxlJobLog log = XxlJobAdminConfig.getAdminConfig()
                            .getXxlJobLogDao().load(failLogId);
                        XxlJobInfo info = XxlJobAdminConfig.getAdminConfig()
                            .getXxlJobInfoDao().loadById(log.getJobId());
                        if (log.getExecutorFailRetryCount() > 0) {
                            JobTriggerPoolHelper.trigger(log.getJobId(), TriggerTypeEnum.RETRY,
                                log.getExecutorFailRetryCount() - 1,
                                log.getExecutorShardingParam(), log.getExecutorParam(), null);
                        }
                        int newAlarmStatus = 0;
                        if (info != null) {
                            boolean alarmResult = XxlJobAdminConfig.getAdminConfig()
                                .getJobAlarmer().alarm(info, log);
                            newAlarmStatus = alarmResult ? 2 : 3;
                        } else {
                            newAlarmStatus = 1;
                        }
                        XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao()
                            .updateAlarmStatus(failLogId, -1, newAlarmStatus);
                    }
                }
            } catch (Exception e) {
                logger.error("xxxxx job fail monitor thread error:{}", e);
            }
            try { TimeUnit.SECONDS.sleep(10); } catch (Exception ignored) {}
        }
        logger.info("xxxxx job fail monitor thread stop");
    });
    monitorThread.setDaemon(true);
    monitorThread.setName("xxl-job, admin JobFailMonitorHelper");
    monitorThread.start();
}

5. Job‑completion monitoring thread JobCompleteHelper.start() creates a callback thread pool for executor callbacks and a daemon thread that marks jobs as failed when they stay in the “running” state for more than 10 minutes while the executor’s heartbeat is missing.

public void start(){
    // callback thread pool
    callbackThreadPool = new ThreadPoolExecutor(2, 20, 30L, TimeUnit.SECONDS,
        new LinkedBlockingQueue<Runnable>(3000),
        r -> new Thread(r, "callback-" + r.hashCode()),
        (r, exec) -> { r.run(); logger.warn("callback too fast, run now"); });
    // monitor thread
    monitorThread = new Thread(() -> {
        try { TimeUnit.MILLISECONDS.sleep(50); } catch (InterruptedException e) { logger.error(e.getMessage(), e); }
        while (!toStop) {
            try {
                Date losedTime = DateUtil.addMinutes(new Date(), -10);
                List<Long> losedJobIds = XxlJobAdminConfig.getAdminConfig()
                    .getXxlJobLogDao().findLostJobIds(losedTime);
                if (losedJobIds != null && !losedJobIds.isEmpty()) {
                    for (Long logId : losedJobIds) {
                        XxlJobLog jobLog = new XxlJobLog();
                        jobLog.setId(logId);
                        jobLog.setHandleTime(new Date());
                        jobLog.setHandleCode(ReturnT.FAIL_CODE);
                        jobLog.setHandleMsg(I18nUtil.getString("joblog_lost_fail"));
                        XxlJobCompleter.updateHandleInfoAndFinish(jobLog);
                    }
                }
            } catch (Exception e) { logger.error("xxxxx job fail monitor thread error:{}", e); }
            try { TimeUnit.SECONDS.sleep(60); } catch (Exception ignored) {}
        }
        logger.info("xxxxx JobLosedMonitorHelper stop");
    });
    monitorThread.setDaemon(true);
    monitorThread.setName("xxl-job, admin JobLosedMonitorHelper");
    monitorThread.start();
}

6. Log‑reporting thread JobLogReportHelper.start() launches a thread that aggregates the last three days of successful and failed job logs into xxl_job_log_report and purges old logs. 7. Job‑scheduling thread The scheduler periodically locks the xxl_job_lock row (using FOR UPDATE ) to ensure only one node in a cluster performs scheduling. It pre‑reads jobs whose next trigger time is within the next 5 seconds, calculates a “pre‑read count” based on thread‑pool capacities, and then processes each job according to three cases:

If now - TriggerNextTime > PRE_READ_MS (5 s) and the misfire strategy is “fire once now”, it triggers the job immediately with type MISFIRE and refreshes the next trigger time.

If now - TriggerNextTime is within 5 seconds, it triggers the job directly with type CRON and refreshes the next time.

If now < TriggerNextTime, the job is placed into a second‑level time‑wheel (0‑59 seconds). When the wheel reaches the appropriate second, the ring thread fires the job.

The scheduler aligns its start to the next whole second (sleeping 5000 - System.currentTimeMillis()%1000 ms) and repeats the loop, committing the transaction after each batch.

// schedule thread (simplified excerpt)
while (!scheduleThreadToStop) {
    long start = System.currentTimeMillis();
    // acquire lock, pre‑read jobs, push to time‑ring or trigger directly
    // ... (SQL and business logic omitted for brevity) ...
    long cost = System.currentTimeMillis() - start;
    if (cost < 1000) {
        TimeUnit.MILLISECONDS.sleep((preReadSuc ? 1000 : PRE_READ_MS) - System.currentTimeMillis()%1000);
    }
}

Time‑wheel thread The ring thread wakes every second, extracts tasks scheduled for the current and previous second from ringData , and triggers them via JobTriggerPoolHelper.trigger . After processing, it clears the temporary list and repeats.

while (!ringThreadToStop) {
    TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis() % 1000);
    int nowSecond = Calendar.getInstance().get(Calendar.SECOND);
    List<Integer> ringItemData = new ArrayList<>();
    for (int i = 0; i < 2; i++) {
        List<Integer> tmp = ringData.remove((nowSecond + 60 - i) % 60);
        if (tmp != null) ringItemData.addAll(tmp);
    }
    for (int jobId : ringItemData) {
        JobTriggerPoolHelper.trigger(jobId, TriggerTypeEnum.CRON, -1, null, null, null);
    }
    ringItemData.clear();
}

The time‑wheel spans seconds 0‑59, ensuring that jobs whose trigger time is within the next second are executed promptly, while also handling any leftover tasks from the previous second.