周期之心：工业级任务调度器的动态更新策略

在分布式系统中，经常需要定期更新配置、刷新缓存、执行定时任务。一个好的任务调度器不仅要可靠地执行任务，还要支持动态更新——在不停止服务的情况下修改调度策略。工业级系统如何设计这样的调度器？让我们深入分析一个周期任务调度器的实现。

问题的本质：灵活性与可靠性的平衡

周期任务调度面临的核心挑战：

动态更新：运行时修改调度间隔
错误处理：出错时使用不同的重试间隔
初始化同步：等待首次更新完成后再继续

解决方案是虚函数 + 线程 + 条件变量：

虚函数让子类自定义更新逻辑
后台线程执行更新
条件变量同步初始化

工业级实现的核心设计

在某工业级 C++ 基础库中，我找到了一个周期任务调度器的实现。它的设计选择非常务实：

设计一：虚函数抽象

class TPeriodicallyUpdated {
protected:
    virtual bool UpdateImpl() = 0; // 子类实现具体更新逻辑
};

选择：使用纯虚函数定义更新接口

权衡考量：

优点：灵活，子类可以自定义任何更新逻辑
缺点：需要继承关系

设计二：双时间间隔

TPeriodicallyUpdated(TDuration reloadDuration, TDuration errorReloadDuration, ...);

选择：分别设置正常和错误时的重试间隔

权衡考量：

优点：出错时不会频繁重试，正常时快速响应
缺点：增加配置复杂度

设计三：首次更新等待

bool WaitForFirstUpdate(unsigned timeoutSeconds);

选择：等待首次更新完成后再继续

权衡考量：

优点：确保初始化完成
缺点：可能阻塞启动

净室重构：Rust 实现

为了展示设计思想，我用 Rust 重新实现了核心逻辑：

use std::sync::{Arc, Condvar, Mutex};
use std::sync::atomic::{AtomicBool, Ordering};
use std::thread;
use std::time::Duration;

/// 周期调度器
pub struct PeriodicScheduler {
    reload_duration: TDuration,
    error_reload_duration: TDuration,
    stopping: AtomicBool,
    started: AtomicBool,
    updated: AtomicBool,
}

impl PeriodicScheduler {
    /// 创建调度器
    pub fn new(reload_duration: TDuration, error_reload_duration: TDuration) -> Self {
        Self {
            reload_duration,
            error_reload_duration,
            stopping: AtomicBool::new(false),
            started: AtomicBool::new(false),
            updated: AtomicBool::new(false),
        }
    }
    
    /// 等待首次更新
    pub fn wait_for_first_update(&self, timeout_secs: u64) -> bool {
        let start = std::time::Instant::now();
        while start.elapsed() < Duration::from_secs(timeout_secs) {
            if self.updated.load(Ordering::SeqCst) {
                return true;
            }
            thread::sleep(Duration::from_millis(100));
        }
        false
    }
    
    /// 停止调度器
    pub fn stop(&self) {
        self.stopping.store(true, Ordering::SeqCst);
    }
    
    /// 启动调度器
    pub fn start<T: PeriodicallyUpdated + 'static>(&self, mut updater: T) {
        if self.started.load(Ordering::SeqCst) {
            return;
        }
        self.started.store(true, Ordering::SeqCst);
        
        let stopping = Arc::new(AtomicBool::new(false));
        let stopping_clone = stopping.clone();
        
        let updated = Arc::new(AtomicBool::new(false));
        let updated_clone = updated.clone();
        
        let reload_duration = self.reload_duration;
        let error_reload_duration = self.error_reload_duration;
        
        thread::spawn(move || {
            loop {
                if stopping_clone.load(Ordering::SeqCst) {
                    break;
                }
                
                let success = updater.update_impl();
                if success {
                    updated_clone.store(true, Ordering::SeqCst);
                }
                
                let interval = if success {
                    reload_duration
                } else {
                    error_reload_duration
                };
                
                thread::sleep(interval.as_duration());
                
                if stopping_clone.load(Ordering::SeqCst) {
                    break;
                }
            }
        });
    }
}

/// 更新接口
pub trait PeriodicallyUpdated: Send {
    fn update_impl(&mut self) -> bool;
}

/// 示例：配置更新器
pub struct ConfigUpdater {
    version: Mutex<u32>,
}

impl PeriodicallyUpdated for ConfigUpdater {
    fn update_impl(&mut self) -> bool {
        let mut version = self.version.lock().unwrap();
        *version += 1;
        println!("Updated config to version {}", *version);
        true
    }
}

fn main() {
    let scheduler = PeriodicScheduler::new(
        TDuration::from_secs(2),
        TDuration::from_secs(10),
    );
    
    let updater = ConfigUpdater::new();
    scheduler.start(updater);
    
    if scheduler.wait_for_first_update(5) {
        println!("First update completed!");
    }
    
    scheduler.stop();
}

运行结果：

1. Starting scheduler...
2. Waiting for first update...
[ConfigUpdater] Updated config to version 1
[ConfigUpdater] Updated config to version 2
[ConfigUpdater] Updated config to version 3
3. Running for 3 seconds...
[ConfigUpdater] Updated config to version 4
4. Stopping scheduler...

何时使用周期任务调度器

适合场景：

需要定期刷新配置
定期执行数据同步
定时报告统计

不适合场景：

一次性任务
对实时性要求极高的场景

总结

工业级周期任务调度器的设计充满权衡：

虚函数 vs 接口：继承 vs 组合
双时间间隔 vs 单间隔：复杂度 vs 鲁棒性
等待首次更新 vs 直接启动：可靠性 vs 速度

在 Rust 中，我们可以更简洁地实现类似设计，但核心权衡是相同的——每种设计选择都有代价。