基于MSE XXL-JOB实现优雅下线

逻辑处理一：执行器注册

• 业务应用在依赖XXL-JOB SDK启动后，会初始化ExecutorRegistryThread线程持续向调度中心汇报心跳。

• 调度中心在接收到后，会通过JobRegistryHelper，将注册上来的执行器信息写入数据库表xxl_job_registry。

• JobRegistryHelper中存在一个线程，定期查询更新xxl_job_group表的address_list（实际使用的列表）。

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

逻辑处理二：选择在线执行机器

• 任务在通过调度线程触发后，会交给XxlJobTrigger完成触发执行动作。

• 在触发执行前，会从xxl_job_group表address_list读取可用的执行器列表。

• 通过ExecutorRouter在上述机器列表中，按对应的路由策略选择一台机器。

• 完成机器选择后，会通过RPC请求将任务分发至对应IP节点运行；此时如选择一个已下线节点任务触发会失败。

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 标记此处 load 方法存在错误
    // cover addressList
    if (addressList!=null && addressList.trim().length()>0) {
        group.setAddressType(1);
        group.setAddressList(addressList.trim());
    }
}

逻辑处理一：将任务分发至对应机器执行

• 业务应用执行器接收到对应任务后，会根据任务ID为每个任务创建一个JobThread线程用于任务执行。

• 该任务本次触发执行请求，会被添加到当前任务线程待执行队列中，任务不同阻塞策略会有不同处理。

• JobThread该线程会持续循环读取，队列中的触发记录并执行对应的JobHandler完成业务逻辑处理。

• 任务执行结束后，会将本次执行结果信息提交给TriggerCallbackThread的执行应答队列，继续下一次执行。

• 执行器在停止时会执行XxlJobExecutor.destroy方法，该方法会中断运行的线程及清理掉队列中等待执行的调度请求。

逻辑处理二：任务执行结果反馈

• TriggerCallbackThread会持续运行加载当前执行结果队列，批量分发执行结果给调度中心。

• 如向调度中心发送应答结果失败，则会写入本地文件落盘，会安排重试。

• 调度中心在接收到执行结果后会进行执行记录的更新写入库。

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();
}