您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 相關知識

Java 定時任務進階實戰：基於 ScheduledExecutorService + 反射的動態可配置任務調度框架

編輯：Java 相關知識

前言

生產系統常需支援 **動態增減任務、調整周期、啟停控制、參數化配置** 而無需重新部署。本文構建一個輕量級定時任務框架：核心調度用 **`ScheduledExecutorService`**，任務解耦用 **反射機制**，配置驅動用 **XML/資料庫/配置中心**，實現任務類與調度邏輯完全分離，支援 Spring 集成與分布式擴展。

一、架構設計

┌─────────────────┐     ┌──────────────┐     ┌──────────────────┐
│  配置源          │────▶│  配置解析器   │────▶│  TaskModel 列表   │
│ (XML/DB/Nacos)   │     │ (XmlReader)   │     │ (類名/方法/參數)  │
└─────────────────┘     └──────────────┘     └────────┬─────────┘
                                                      │
                                    ┌─────────────────▼─────────────────┐
                                    │      SchedulerEngine              │
                                    │  ScheduledExecutorService         │
                                    │  + 動態增減/修改/暫停/恢復 API     │
                                    └─────────────────┬─────────────────┘
                                                      │
                                    ┌─────────────────▼─────────────────┐
                                    │        MyTask (Runnable)          │
                                    │  反射調用 targetClass.targetMethod│
                                    │  + 異常隔離 + 執行日誌 + 指標收集  │
                                    └───────────────────────────────────┘

二、核心領域模型

// 任務元數據模型 (可擴展: cron表達式、分片參數、優先級、描述...)
public class TaskModel implements Serializable {
    private String id;              // 唯一標識
    private String className;       // 目標類全限定名
    private String methodName;      // 目標方法名 (無參靜態方法)
    private long initialDelay;      // 首次延遲 (秒)
    private long period;            // 執行周期 (秒) - 固定頻率
    // private String cronExpression; // 進階: 支持 Cron 表達式
    private boolean enabled = true; // 是否啟用
    private Map params; // 方法參數 (擴展: 非靜態/帶參)
    // getter/setter/toString 省略
}

三、配置解析層 (支援熱加載)

1. XML 配置文件 (tasks-config.xml)

<?xml version="1.0" encoding="UTF-8"?>
<task-config>
    <task id="syncUser" enabled="true">
        <class>com.example.task.SyncUserTask</class>
        <method>execute</method>
        <initialDelay>10</initialDelay>
        <period>300</period> <!-- 5分鐘 -->
        <params>
            <param key="batchSize">1000</param>
        </params>
    </task>
    <task id="cleanLog" enabled="true">
        <class>com.example.task.CleanLogTask</class>
        <method>run</method>
        <initialDelay>60</initialDelay>
        <period>3600</period> <!-- 1小時 -->
    </task>
</task-config>

2. 通用解析器介面與 XML 實現 (JDOM2 / DOM / JAXB 均可)

public interface TaskConfigReader {
    List loadTasks() throws Exception;
    void watchChanges(Consumer> onChange); // 熱加載回調
}

public class XmlTaskConfigReader implements TaskConfigReader {
    private final String configPath;
    private final WatchService watchService; // JDK NIO 文件監聽

    public XmlTaskConfigReader(String configPath) throws IOException {
        this.configPath = configPath;
        this.watchService = FileSystems.getDefault().newWatchService();
        Paths.get(configPath).getParent().register(watchService, StandardWatchEventKinds.ENTRY_MODIFY);
    }

    @Override
    public List loadTasks() {
        List tasks = new ArrayList<>();
        try (InputStream is = new FileInputStream(configPath)) {
            DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();
            Document doc = factory.newDocumentBuilder().parse(is);
            NodeList taskNodes = doc.getElementsByTagName("task");
            for (int i = 0; i < taskNodes.getLength(); i++) {
                Element taskEl = (Element) taskNodes.item(i);
                TaskModel tm = new TaskModel();
                tm.setId(taskEl.getAttribute("id"));
                tm.setEnabled(Boolean.parseBoolean(taskEl.getAttribute("enabled")));
                tm.setClassName(getChildText(taskEl, "class"));
                tm.setMethodName(getChildText(taskEl, "method"));
                tm.setInitialDelay(Long.parseLong(getChildText(taskEl, "initialDelay", "0")));
                tm.setPeriod(Long.parseLong(getChildText(taskEl, "period", "60")));
                // 解析 params...
                tasks.add(tm);
            }
        } catch (Exception e) {
            throw new RuntimeException("加載任務配置失敗", e);
        }
        return tasks;
    }

    @Override
    public void watchChanges(Consumer> onChange) {
        new Thread(() -> {
            while (true) {
                try {
                    WatchKey key = watchService.take();
                    for (WatchEvent event : key.pollEvents()) {
                        if (event.context().toString().endsWith(configPath)) {
                            onChange.accept(loadTasks());
                            break;
                        }
                    }
                    key.reset();
                } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; }
                catch (Exception e) { log.error("配置熱加載異常", e); }
            }
        }, "TaskConfig-Watcher").start();
    }

    private String getChildText(Element parent, String tagName, String defaultVal) { ... }
    private String getChildText(Element parent, String tagName) { return getChildText(parent, tagName, null); }
}

四、調度引擎核心實現

@Slf4j
public class SchedulerEngine {
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(
        Runtime.getRuntime().availableProcessors(), 
        new ThreadFactoryBuilder().setNameFormat("TaskScheduler-%d").build()
    );
    private final Map> runningTasks = new ConcurrentHashMap<>();
    private final TaskConfigReader configReader;

    public SchedulerEngine(TaskConfigReader configReader) {
        this.configReader = configReader;
    }

    // 啟動：加載配置並調度
    public void start() {
        List tasks = configReader.loadTasks();
        scheduleTasks(tasks);
        
        // 註冊熱加載
        configReader.watchChanges(this::reloadTasks);
        log.info("調度引擎啟動完成，共載入 {} 個任務", tasks.size());
    }

    // 核心調度邏輯
    private void scheduleTasks(List tasks) {
        for (TaskModel tm : tasks) {
            if (!tm.isEnabled()) continue;
            try {
                // 反射驗證類/方法存在 (提前失敗)
                Class clazz = Class.forName(tm.getClassName());
                Method method = clazz.getMethod(tm.getMethodName()); // 僅支持無參靜態方法
                if (!Modifier.isStatic(method.getModifiers())) {
                    log.warn("任務 {} 方法非靜態，跳過", tm.getId());
                    continue;
                }
                
                // 提交調度
                ScheduledFuture future = scheduler.scheduleAtFixedRate(
                    new MyTask(tm), 
                    tm.getInitialDelay(), 
                    tm.getPeriod(), 
                    TimeUnit.SECONDS
                );
                runningTasks.put(tm.getId(), future);
                log.info("任務已調度: {} -> {}.{}() [delay={}s, period={}s]", 
                    tm.getId(), tm.getClassName(), tm.getMethodName(), 
                    tm.getInitialDelay(), tm.getPeriod());
            } catch (Exception e) {
                log.error("任務調度失敗: {}", tm.getId(), e);
            }
        }
    }

    // 熱加載：對比新舊配置，增量增減
    private void reloadTasks(List newTasks) {
        Map newMap = newTasks.stream().collect(Collectors.toMap(TaskModel::getId, t -> t));
        Map oldMap = runningTasks.keySet().stream()
            .collect(Collectors.toMap(k -> k, k -> null)); // 簡化：實際需保存舊 TaskModel
        
        // 1. 新增/修改
        for (TaskModel newTm : newTasks) {
            if (!newTm.isEnabled()) { cancelTask(newTm.getId()); continue; }
            if (!runningTasks.containsKey(newTm.getId())) {
                scheduleTasks(List.of(newTm)); // 新增
            } else if (isChanged(newTm, oldMap.get(newTm.getId()))) { // 需實現 isChanged
                cancelTask(newTm.getId());
                scheduleTasks(List.of(newTm)); // 修改：取消舊 + 重新調度
            }
        }
        // 2. 刪除
        for (String oldId : runningTasks.keySet()) {
            if (!newMap.containsKey(oldId)) cancelTask(oldId);
        }
        log.info("任務配置熱加載完成，當前運行任務數: {}", runningTasks.size());
    }

    // 取消任務
    public void cancelTask(String taskId) {
        ScheduledFuture future = runningTasks.remove(taskId);
        if (future != null) {
            future.cancel(true); // 中斷正在執行的任務
            log.info("任務已取消: {}", taskId);
        }
    }

    // 動態 API (可暴露為 HTTP/Actuator 端點)
    public void enableTask(String taskId) { /* 更新配置源並觸發 reload */ }
    public void disableTask(String taskId) { /* 同上 */ }
    public void updatePeriod(String taskId, long newPeriod) { /* 同上 */ }
    public Map getAllTaskStatus() { /* 返回運行狀態、上次執行時間、耗時、成功/失敗計數 */ }

    // 關閉引擎
    public void shutdown() {
        scheduler.shutdown();
        try { if (!scheduler.awaitTermination(60, TimeUnit.SECONDS)) scheduler.shutdownNow(); }
        catch (InterruptedException e) { scheduler.shutdownNow(); Thread.currentThread().interrupt(); }
    }
}

五、任務執行器：反射調用 + 異常隔離 + 指標

public class MyTask implements Runnable {
    private final TaskModel taskModel;
    private static final MeterRegistry METRICS = Metrics.globalRegistry; // Micrometer

    public MyTask(TaskModel tm) { this.taskModel = tm; }

    @Override
    public void run() {
        String taskId = taskModel.getId();
        Timer.Sample sample = Timer.start(METRICS);
        long start = System.currentTimeMillis();
        try {
            log.info("⏰ 任務開始執行: {} @ {}", taskId, new Date());
            
            Class clazz = Class.forName(taskModel.getClassName());
            Method method = clazz.getMethod(taskModel.getMethodName());
            
            // 安全檢查: 僅允許 public static 方法
            if (!Modifier.isPublic(method.getModifiers()) || !Modifier.isStatic(method.getModifiers())) {
                throw new IllegalStateException("任務方法必須為 public static");
            }
            
            method.invoke(null); // 靜態方法 invoke(null)
            
            sample.stop(Timer.builder("task.execution.duration").tag("task", taskId).tag("status", "success").register(METRICS));
            log.info("✅ 任務執行成功: {} 耗時 {}ms", taskId, System.currentTimeMillis() - start);
        } catch (InvocationTargetException e) {
            // 解包真實業務異常
            Throwable cause = e.getCause();
            sample.stop(Timer.builder("task.execution.duration").tag("task", taskId).tag("status", "error").register(METRICS));
            log.error("❌ 任務業務異常: {} - {}", taskId, cause.getMessage(), cause);
        } catch (Exception e) {
            sample.stop(Timer.builder("task.execution.duration").tag("task", taskId).tag("status", "error").register(METRICS));
            log.error("❌ 任務反射調用失敗: {}", taskId, e);
        }
    }
}

六、業務任務類示例 (POJO, 無框架依賴)


package com.example.task;

public class SyncUserTask {
    public static void execute() {
        // 業務邏輯: 同步用戶數據
        System.out.println("同步用戶數據任務執行...");
        // 可注入 Spring Bean? 需結合 ApplicationContextAware 或靜態持有上下文
    }
}

public class CleanLogTask {
    public static void run() {
        System.out.println("清理過期日誌任務執行...");
    }
}

七、Spring Boot 整合啟動 (自動裝配)

@Configuration
@EnableScheduling // 可選，若需同時支援 @Scheduled
public class SchedulerAutoConfig {

    @Bean
    @ConditionalOnMissingBean
    public TaskConfigReader taskConfigReader(@Value("${task.config.path:classpath:tasks-config.xml}") Resource resource) throws IOException {
        return new XmlTaskConfigReader(resource.getFile().getAbsolutePath());
    }

    @Bean(destroyMethod = "shutdown")
    public SchedulerEngine schedulerEngine(TaskConfigReader reader) {
        SchedulerEngine engine = new SchedulerEngine(reader);
        engine.start();
        return engine;
    }
    
    // 暴露 Actuator 端點管理任務
    @Bean
    public TaskSchedulerEndpoint taskSchedulerEndpoint(SchedulerEngine engine) {
        return new TaskSchedulerEndpoint(engine);
    }
}

@Component
@Endpoint(id = "tasks")
class TaskSchedulerEndpoint {
    private final SchedulerEngine engine;
    // @ReadOperation, @WriteOperation 等暴露 HTTP API
}

八、生產級擴展建議

能力	實現方案
分布式協調 (單點執行)	集成 ShedLock (基於 DB/Redis/ZK 鎖) 或 XXL-JOB / PowerJob 成熟平台
Cron 表達式支援	引入 `CronExpression` (Spring/Quartz) + `scheduleWithFixedDelay` 計算下次執行時間
任務參數注入	反射獲取方法參數類型，從 TaskModel.params 轉換注入 (支援基本類型、String、JSON 轉對象)
非靜態方法/Bean 方法調用	集成 Spring `ApplicationContext.getBean(className)` 獲取實例再反射調用
執行日誌持久化	異步寫入 DB/ES，包含 taskId, startTime, endTime, status, errorMsg, traceId
監控告警	Micrometer + Prometheus + Grafana 監控 `task.execution.duration`, `task.success.total`, `task.error.total`
優雅停機	實現 `SmartLifecycle` / `DisposableBean`，`shutdown()` 等待運行中任務完成 (配置超時)

九、總結

基於 `ScheduledExecutorService` + 反射 + 配置驅動的輕量級定時任務框架，核心優勢在於：**零重啟動態變更、業務代碼零框架侵入、反射解耦調度邏輯、熱加載配置、指標可觀測**。適合中小型系統、微服務單體模塊、不想引入重型調度平台的場景。若需分布式高可用、分片、工作流編排，請直接接入 **XXL-JOB / PowerJob / ElasticJob** 等成熟平台，本框架可作為其輕量級 Client Side 補充或過渡方案。