Graceful shutdown with MSE XXL-JOB

Logic 1: Executor registration

• After your application starts with the XXL-JOB SDK, it initializes the ExecutorRegistryThread to continuously send heartbeats to the scheduling center.

• The scheduling center uses JobRegistryHelper to write the registered executor information into the xxl_job_registry database table.

• JobRegistryHelper runs a background thread that periodically updates the address_list field (the actual list used) in the xxl_job_group table.

// 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;
}

registryMonitorThread = new Thread(new Runnable() {
    @Override
    public void run() {
        while (!toStop) {
            try {
                // auto registry group
                List<XxlJobGroup> groupList = XxlJobAdminConfig.getAdminConfig().getXxlJobGroupDao().findByAddressType(0);
                if (groupList!=null && !groupList.isEmpty()) {
                    // remove dead address (admin/executor)
                    List<Integer> ids = XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().findDead(RegistryConfig.DEAD_TIMEOUT, new Date());
                    if (ids!=null && ids.size()>0) {...}
                    // fresh online address (admin/executor)
                    HashMap&lt;String, List&lt;String&gt;> appAddressMap = new HashMap<>();
                    List<XxlJobRegistry> list = XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().findAll(RegistryConfig.DEAD_TIMEOUT, new Date());
                    if (list != null) {...}
                    // fresh group address
                    for (XxlJobGroup group: groupList) {
                        List<String> registryList = appAddressMap.get(group.getAppname());
                        String addressListStr = null;
                        if (registryList!=null && !registryList.isEmpty()) {...}
                        group.setAddressList(addressListStr);
                        group.setUpdateTime(new Date());
                        XxlJobAdminConfig.getAdminConfig().getXxlJobGroupDao().update(group);
                    }
                }
            } catch (Exception e) {...}
            try {
                TimeUnit.SECONDS.sleep(RegistryConfig.BEAT_TIMEOUT);
            } catch (InterruptedException e) {
                if (!toStop) {...}
            }
        }
        logger.info(">>>>>>>>>>> xxl-job, job registry monitor thread stop");
    }
});

Logic 2: Selecting an online executor

• When triggered by the scheduler thread, a task is handed to XxlJobTrigger for execution.

• Before execution, XxlJobTrigger reads the available executor list from the address_list field in the xxl_job_group table.

• ExecutorRouter selects one machine from this list based on the configured routing policy.

• After selection, the task is dispatched via RPC to the chosen IP node. If the selected node is already shutting down, the trigger fails.

public static void trigger(int jobId,
                           TriggerTypeEnum triggerType,
                           int failRetryCount,
                           String executorShardingParam,
                           String executorParam,
                           String addressList) {
    // load data
    XxlJobInfo jobInfo = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().loadById(jobId);
    if (jobInfo == null) {
        logger.warn(">>>>>>>>>>> trigger fail, jobId invalid, jobId={}", jobId);
        return;
    }
    if (executorParam != null) {
        jobInfo.setExecutorParam(executorParam);
    }
    int finalFailRetryCount = failRetryCount>=0?failRetryCount:jobInfo.getExecutorFailRetryCount();
    XxlJobGroup group = XxlJobAdminConfig.getAdminConfig().getXxlJobGroupDao().load(jobInfo.getJobGroup()); // IDE flags an error in this load method
    // cover addressList
    if (addressList!=null && addressList.trim().length()>0) {
        group.setAddressType(1);
        group.setAddressList(addressList.trim());
    }
}

Logic 1: Dispatching tasks to executors

• When an executor receives a task, it creates a JobThread for that task ID to handle execution.

• The current trigger request is added to the task thread’s execution queue. Different blocking strategies yield different behaviors.

• The JobThread continuously loops, reading trigger records from the queue and executing the corresponding JobHandler to process business logic.

• After execution, the result is submitted to the TriggerCallbackThread’s callback queue, and the thread proceeds to the next execution.

• During shutdown, XxlJobExecutor.destroy interrupts running threads and clears pending scheduling requests from the queue.

Logic 2: Reporting execution results

• TriggerCallbackThread continuously processes the result queue and batches results to the scheduling center.

• If sending results fails, they are written to a local file for retry.

• The scheduling center updates execution records in the database upon receiving results.

public void destroy(){
    // destroy executor-server
    stopEmbedServer();
    // destroy jobThreadRepository
    if (jobThreadRepository.size() > 0) {
        for (Map.Entry<Integer, JobThread> item: jobThreadRepository.entrySet()) {
            JobThread oldJobThread = removeJobThread(item.getKey(), "web container destroy and kill the job.");
            // wait for job thread push result to callback queue
            if (oldJobThread != null) {
                try {
                    oldJobThread.join();
                } catch (InterruptedException e) {
                    logger.error(">>>>>>>>>>> xxl-job, JobThread destroy(join) error, jobId:{}", item.getKey(), e);
                }
            }
        }
        jobThreadRepository.clear();
    }
    jobHandlerRepository.clear();
    // destroy JobLogFileCleanThread
    JobLogFileCleanThread.getInstance().toStop();
    // destroy TriggerCallbackThread
    TriggerCallbackThread.getInstance().toStop();
}