大家好，我是扣扣。今天聊一个很实用的功能——配置热更新。

什么是配置热更新？

传统的配置更新流程是这样的：

1. 修改配置文件
2. 重启应用
3. 新配置生效

对于需要7x24小时运行的自动化脚本，每次改配置都要重启就很烦了。配置热更新就是让你在不重启程序的情况下，让配置变更生效。

常见场景

1. 定时任务：修改重试次数、间隔时间
2. 监控脚本：调整阈值、告警规则
3. 数据采集：修改采集频率、数据源
4. 爬虫脚本：修改请求间隔、代理列表

一、基础实现：文件监听

最简单的方式：监听配置文件变化。

"""配置文件热更新基础实现"""

import time
import json
import threading
from pathlib import Path
from typing import Any, Optional, Dict
from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class ConfigWatcher:
    """配置文件监听器"""
    
    config_file: str
    poll_interval: float = 1.0  # 轮询间隔（秒）
    _last_modified: float = 0
    _last_content: str = ""
    _callbacks: list = field(default_factory=list)
    _running: bool = False
    _thread: Optional[threading.Thread] = None
    
    def __post_init__(self):
        self._check_and_update()  # 初始化时加载一次
    
    def register_callback(self, callback):
        """注册配置变更回调"""
        self._callbacks.append(callback)
    
    def _check_and_update(self):
        """检查并更新配置"""
        path = Path(self.config_file)
        
        if not path.exists():
            return False
        
        current_mtime = path.stat().st_mtime
        
        # 检查是否修改
        if current_mtime != self._last_modified:
            self._last_modified = current_mtime
            
            # 读取新内容
            with open(self.config_file, 'r', encoding='utf-8') as f:
                new_content = f.read()
            
            # 内容有变化才触发回调
            if new_content != self._last_content:
                old_content = self._last_content
                self._last_content = new_content
                
                # 触发所有回调
                for callback in self._callbacks:
                    try:
                        callback(old_content, new_content)
                    except Exception as e:
                        print(f"回调执行出错: {e}")
                
                return True
        
        return False
    
    def start(self):
        """启动监听"""
        self._running = True
        self._thread = threading.Thread(target=self._watch_loop, daemon=True)
        self._thread.start()
        print(f"开始监听配置文件: {self.config_file}")
    
    def _watch_loop(self):
        """监听循环"""
        while self._running:
            self._check_and_update()
            time.sleep(self.poll_interval)
    
    def stop(self):
        """停止监听"""
        self._running = False
        if self._thread:
            self._thread.join(timeout=2)
        print("停止监听配置文件")


class HotReloadConfig:
    """支持热更新的配置类"""
    
    def __init__(self, config_file: str):
        self.config_file = config_file
        self._config: Dict[str, Any] = {}
        self._lock = threading.RLock()
        self._watcher = ConfigWatcher(config_file)
        self._watcher.register_callback(self._on_config_changed)
    
    def _load_config(self) -> Dict[str, Any]:
        """加载配置文件"""
        path = Path(self.config_file)
        
        if not path.exists():
            return {}
        
        suffix = path.suffix.lower()
        
        if suffix == '.json':
            with open(self.config_file, 'r', encoding='utf-8') as f:
                return json.load(f)
        elif suffix in ('.yaml', '.yml'):
            import yaml
            with open(self.config_file, 'r', encoding='utf-8') as f:
                return yaml.safe_load(f) or {}
        elif suffix == '.toml':
            import tomli
            with open(self.config_file, 'r', encoding='utf-8') as f:
                return tomli.loads(f.read())
        else:
            # 简单key=value格式
            config = {}
            with open(self.config_file, 'r', encoding='utf-8') as f:
                for line in f:
                    line = line.strip()
                    if line and not line.startswith('#'):
                        if '=' in line:
                            key, value = line.split('=', 1)
                            config[key.strip()] = value.strip()
            return config
    
    def _on_config_changed(self, old_content: str, new_content: str):
        """配置变更回调"""
        with self._lock:
            self._config = self._load_config()
        print(f"[{datetime.now()}] 配置已更新: {list(self._config.keys())}")
    
    def start_watching(self):
        """启动监听"""
        with self._lock:
            self._config = self._load_config()
        self._watcher.start()
    
    def stop_watching(self):
        """停止监听"""
        self._watcher.stop()
    
    def get(self, key: str, default: Any = None) -> Any:
        """获取配置值"""
        with self._lock:
            return self._config.get(key, default)
    
    def set(self, key: str, value: Any):
        """设置配置值（仅影响内存）"""
        with self._lock:
            self._config[key] = value
    
    @property
    def all(self) -> Dict[str, Any]:
        """获取所有配置"""
        with self._lock:
            return dict(self._config)


# 使用示例
if __name__ == '__main__':
    # 创建配置
    config_content = {
        "retry_count": 3,
        "retry_interval": 5,
        "log_level": "INFO",
        "max_workers": 4
    }
    
    with open('config.json', 'w') as f:
        json.dump(config_content, f, indent=2)
    
    # 初始化配置
    config = HotReloadConfig('config.json')
    config.start_watching()
    
    print(f"初始配置: {config.all}")
    
    # 模拟修改配置
    print("\n修改配置文件...")
    time.sleep(2)
    config_content['retry_count'] = 10
    config_content['log_level'] = 'DEBUG'
    
    with open('config.json', 'w') as f:
        json.dump(config_content, f, indent=2)
    
    time.sleep(2)
    print(f"更新后配置: {config.all}")
    
    # 使用配置
    print(f"\n当前重试次数: {config.get('retry_count')}")
    
    config.stop_watching()

二、进阶实现：配置项单独监听

有时候我们只关心某些配置项的变化，不需要每次都全量更新：

"""细粒度配置监听"""

import threading
import time
from pathlib import Path
from typing import Any, Callable, Dict, Optional
from dataclasses import dataclass
import json

@dataclass
class ConfigItem:
    """单个配置项"""
    name: str
    value: Any
    validators: list = None
    callbacks: list = None
    
    def __post_init__(self):
        if self.validators is None:
            self.validators = []
        if self.callbacks is None:
            self.callbacks = []


class FineGrainedConfig:
    """细粒度配置管理器"""
    
    def __init__(self, config_file: str):
        self.config_file = config_file
        self._items: Dict[str, ConfigItem] = {}
        self._lock = threading.RLock()
        self._last_mtime: float = 0
        self._running: bool = False
        self._thread: Optional[threading.Thread] = None
    
    def register(self, name: str, default: Any = None, 
                 validator: Callable = None, 
                 on_change: Callable = None) -> 'FineGrainedConfig':
        """注册配置项"""
        with self._lock:
            self._items[name] = ConfigItem(
                name=name,
                value=default,
                validators=[validator] if validator else [],
                callbacks=[on_change] if on_change else []
            )
        return self
    
    def set_validator(self, name: str, validator: Callable):
        """设置验证器"""
        with self._lock:
            if name in self._items:
                self._items[name].validators.append(validator)
    
    def on_change(self, name: str, callback: Callable):
        """设置变更回调"""
        with self._lock:
            if name in self._items:
                self._items[name].callbacks.append(callback)
    
    def _validate(self, name: str, value: Any) -> bool:
        """验证配置值"""
        with self._lock:
            item = self._items.get(name)
            if not item:
                return True
            
            for validator in item.validators:
                try:
                    if not validator(value):
                        print(f"验证失败: {name}={value}")
                        return False
                except Exception as e:
                    print(f"验证器异常: {e}")
                    return False
        return True
    
    def _notify_change(self, name: str, old_value: Any, new_value: Any):
        """通知变更"""
        with self._lock:
            item = self._items.get(name)
            if not item:
                return
            
            for callback in item.callbacks:
                try:
                    callback(old_value, new_value)
                except Exception as e:
                    print(f"回调执行异常: {e}")
    
    def _load_and_update(self):
        """加载并更新配置"""
        path = Path(self.config_file)
        if not path.exists():
            return
        
        current_mtime = path.stat().st_mtime
        if current_mtime == self._last_mtime:
            return
        
        self._last_mtime = current_mtime
        
        # 读取配置
        with open(self.config_file, 'r', encoding='utf-8') as f:
            new_config = json.load(f)
        
        # 更新有变化的配置项
        with self._lock:
            for name, item in self._items.items():
                if name in new_config:
                    new_value = new_config[name]
                    if new_value != item.value:
                        # 验证
                        if self._validate(name, new_value):
                            old_value = item.value
                            item.value = new_value
                            self._notify_change(name, old_value, new_value)
                            print(f"配置更新: {name} = {old_value} -> {new_value}")
    
    def start(self):
        """启动"""
        self._load_and_update()  # 初始加载
        self._running = True
        self._thread = threading.Thread(target=self._loop, daemon=True)
        self._thread.start()
    
    def _loop(self):
        """监听循环"""
        while self._running:
            self._load_and_update()
            time.sleep(1)
    
    def stop(self):
        """停止"""
        self._running = False
    
    def get(self, name: str, default: Any = None) -> Any:
        """获取配置"""
        with self._lock:
            item = self._items.get(name)
            return item.value if item else default
    
    def set(self, name: str, value: Any, persist: bool = True):
        """设置配置"""
        if not self._validate(name, value):
            return False
        
        with self._lock:
            item = self._items.get(name)
            if item:
                old_value = item.value
                item.value = value
                self._notify_change(name, old_value, value)
                
                if persist:
                    self._persist_to_file()
        return True
    
    def _persist_to_file(self):
        """持久化到文件"""
        with self._lock:
            config = {name: item.value for name, item in self._items.items()}
        
        with open(self.config_file, 'w', encoding='utf-8') as f:
            json.dump(config, f, indent=2, ensure_ascii=False)


# 使用示例
if __name__ == '__main__':
    # 创建初始配置
    initial_config = {
        "retry_count": 3,
        "timeout": 30,
        "log_level": "INFO",
        "enabled": True
    }
    
    with open('settings.json', 'w') as f:
        json.dump(initial_config, f, indent=2)
    
    # 创建配置管理器
    config = FineGrainedConfig('settings.json')
    
    # 注册配置项
    config.register(
        'retry_count', 
        default=3,
        validator=lambda x: 1 <= x <= 10,  # 1-10之间
        on_change=lambda old, new: print(f"重试次数变更: {old} -> {new}")
    )
    
    config.register(
        'timeout',
        default=30,
        validator=lambda x: x > 0,
        on_change=lambda old, new: print(f"超时时间变更: {old}s -> {new}s")
    )
    
    config.register('log_level', default='INFO')
    config.register('enabled', default=True)
    
    config.start()
    
    print("当前配置:", {k: config.get(k) for k in ['retry_count', 'timeout']})
    
    # 模拟修改配置
    print("\n修改配置文件...")
    time.sleep(2)
    
    import json
    new_config = {
        "retry_count": 5,  # 有效修改
        "timeout": 60,    # 有效修改
        "log_level": "DEBUG",
        "enabled": True
    }
    
    with open('settings.json', 'w') as f:
        json.dump(new_config, f, indent=2)
    
    time.sleep(2)
    print("当前配置:", {k: config.get(k) for k in ['retry_count', 'timeout']})
    
    config.stop()

三、API方式热更新

有些场景下，配置通过API或命令行修改：

"""API方式更新配置"""

from flask import Flask, request, jsonify
import threading
import time
from typing import Any, Dict
import json

class APIConfigManager:
    """API配置管理器"""
    
    def __init__(self):
        self._config: Dict[str, Any] = {}
        self._lock = threading.RLock()
        self._change_listeners: list = []
        self._history: list = []  # 配置变更历史
        self._max_history = 100
    
    def add_change_listener(self, callback):
        """添加变更监听器"""
        self._change_listeners.append(callback)
    
    def set(self, key: str, value: Any, reason: str = ""):
        """设置配置"""
        with self._lock:
            old_value = self._config.get(key)
            
            if old_value == value:
                return {"status": "unchanged", "key": key, "value": value}
            
            self._config[key] = value
            
            # 记录历史
            self._history.append({
                "time": time.time(),
                "key": key,
                "old_value": old_value,
                "new_value": value,
                "reason": reason
            })
            
            # 限制历史长度
            if len(self._history) > self._max_history:
                self._history = self._history[-self._max_history:]
            
            # 通知监听器
            for listener in self._change_listeners:
                try:
                    listener(key, old_value, value)
                except Exception as e:
                    print(f"监听器异常: {e}")
            
            return {"status": "updated", "key": key, "old": old_value, "new": value}
    
    def get(self, key: str, default: Any = None) -> Any:
        """获取配置"""
        with self._lock:
            return self._config.get(key, default)
    
    def get_all(self) -> Dict[str, Any]:
        """获取所有配置"""
        with self._lock:
            return dict(self._config)
    
    def history(self, key: str = None, limit: int = 20) -> list:
        """获取变更历史"""
        with self._lock:
            if key:
                return [h for h in self._history if h['key'] == key][-limit:]
            return self._history[-limit:]
    
    def rollback(self, key: str, steps: int = 1) -> Dict[str, Any]:
        """回滚配置"""
        with self._lock:
            # 找到目标配置的历史
            key_history = [h for h in reversed(self._history) if h['key'] == key]
            
            if len(key_history) <= steps:
                return {"status": "error", "message": "无法回滚"}
            
            # 获取回滚目标值
            rollback_entry = key_history[steps]
            old_value = rollback_entry['old_value']
            
            return self.set(key, old_value, reason=f"回滚 {steps} 步")


# Flask API
app = Flask(__name__)
config_manager = APIConfigManager()

@app.route('/config', methods=['GET'])
def get_config():
    """获取所有配置"""
    return jsonify(config_manager.get_all())

@app.route('/config/<key>', methods=['GET'])
def get_config_item(key):
    """获取单个配置"""
    value = config_manager.get(key)
    if value is None:
        return jsonify({"error": "配置项不存在"}), 404
    return jsonify({"key": key, "value": value})

@app.route('/config', methods=['POST'])
def set_config():
    """设置配置"""
    data = request.json
    key = data.get('key')
    value = data.get('value')
    reason = data.get('reason', '')
    
    if not key:
        return jsonify({"error": "缺少key"}), 400
    
    result = config_manager.set(key, value, reason)
    return jsonify(result)

@app.route('/config/<key>', methods=['DELETE'])
def delete_config(key):
    """删除配置"""
    return jsonify(config_manager.set(key, None, reason="删除"))

@app.route('/config/history', methods=['GET'])
def get_history():
    """获取变更历史"""
    key = request.args.get('key')
    limit = int(request.args.get('limit', 20))
    return jsonify(config_manager.history(key, limit))

@app.route('/config/<key>/rollback', methods=['POST'])
def rollback_config(key):
    """回滚配置"""
    steps = int(request.args.get('steps', 1))
    return jsonify(config_manager.rollback(key, steps))

# 监听器示例
def on_config_change(key, old_value, new_value):
    """配置变更时执行"""
    print(f"配置变更: {key} = {old_value} -> {new_value}")
    
    # 可以执行各种操作
    if key == 'log_level':
        print("日志级别变更，需要重新配置日志")
    elif key == 'max_workers':
        print("工作线程数变更，需要重建线程池")

config_manager.add_change_listener(on_config_change)


if __name__ == '__main__':
    # 初始化配置
    config_manager.set('retry_count', 3, reason="初始化")
    config_manager.set('timeout', 30, reason="初始化")
    
    print("当前配置:", config_manager.get_all())
    print("\n启动API服务...")
    app.run(host='0.0.0.0', port=5000, debug=True)

四、YAML配置热更新

"""YAML配置热更新"""

import yaml
import time
import threading
from pathlib import Path
from typing import Any, Dict, Optional
from dataclasses import dataclass, field

@dataclass
class YAMLConfig:
    """YAML配置文件管理器"""
    
    config_file: str
    _data: Dict[str, Any] = field(default_factory=dict)
    _lock: threading.RLock = field(default_factory=threading.RLock)
    _observers: Dict[str, list] = field(default_factory=dict)
    _running: bool = False
    _thread: Optional[threading.Thread] = None
    
    def __post_init__(self):
        self._load()
    
    def _load(self):
        """加载配置"""
        path = Path(self.config_file)
        if not path.exists():
            self._data = {}
            return
        
        with open(self.config_file, 'r', encoding='utf-8') as f:
            self._data = yaml.safe_load(f) or {}
    
    def observe(self, key: str, callback):
        """观察配置项变化"""
        if key not in self._observers:
            self._observers[key] = []
        self._observers[key].append(callback)
    
    def _notify(self, key: str, old_value: Any, new_value: Any):
        """通知观察者"""
        if key in self._observers:
            for callback in self._observers[key]:
                try:
                    callback(old_value, new_value)
                except Exception as e:
                    print(f"观察者异常: {e}")
    
    def _watch_loop(self):
        """监听循环"""
        last_mtime = 0
        
        while self._running:
            path = Path(self.config_file)
            if path.exists():
                current_mtime = path.stat().st_mtime
                
                if current_mtime != last_mtime:
                    last_mtime = current_mtime
                    
                    # 加载新配置
                    with self._lock:
                        old_data = dict(self._data)
                        self._load()
                        
                        # 找出变化的项目
                        for key in set(list(old_data.keys()) + list(self._data.keys())):
                            old_value = old_data.get(key)
                            new_value = self._data.get(key)
                            
                            if old_value != new_value:
                                self._notify(key, old_value, new_value)
            
            time.sleep(1)
    
    def start(self):
        """启动监听"""
        self._running = True
        self._thread = threading.Thread(target=self._watch_loop, daemon=True)
        self._thread.start()
    
    def stop(self):
        """停止监听"""
        self._running = False
    
    def get(self, *keys: str, default: Any = None) -> Any:
        """获取嵌套配置 (get('database', 'host'))"""
        with self._lock:
            value = self._data
            for key in keys:
                if isinstance(value, dict):
                    value = value.get(key)
                else:
                    return default
            return value if value is not None else default
    
    def set(self, *keys: str, value: Any):
        """设置嵌套配置"""
        with self._lock:
            # 构建路径
            if len(keys) == 1:
                self._data[keys[0]] = value
            else:
                d = self._data
                for key in keys[:-1]:
                    if key not in d:
                        d[key] = {}
                    d = d[key]
                d[keys[-1]] = value
            
            # 保存
            with open(self.config_file, 'w', encoding='utf-8') as f:
                yaml.dump(self._data, f, default_flow_style=False, allow_unicode=True)


# 使用示例
if __name__ == '__main__':
    # 创建YAML配置
    config_data = {
        'app': {
            'name': 'MyApp',
            'version': '1.0.0'
        },
        'database': {
            'host': 'localhost',
            'port': 3306,
            'pool_size': 10
        },
        'cache': {
            'enabled': True,
            'ttl': 3600
        }
    }
    
    with open('config.yaml', 'w') as f:
        yaml.dump(config_data, f, allow_unicode=True)
    
    # 创建配置管理器
    config = YAMLConfig('config.yaml')
    
    # 添加观察者
    config.observe('database.host', lambda old, new: print(f"数据库主机变更: {new}"))
    config.observe('database.pool_size', lambda old, new: print(f"连接池大小变更: {new}"))
    config.observe('cache.enabled', lambda old, new: print(f"缓存{'启用' if new else '禁用'}"))
    
    config.start()
    
    print("初始配置:", config.get('database'))
    
    # 模拟修改配置
    time.sleep(2)
    config.set('database', 'host', value='db.example.com')
    config.set('database', 'pool_size', value=20)
    
    time.sleep(2)
    print("更新后:", config.get('database'))
    
    config.stop()

总结

1. 文件监听：适合配置文件修改的场景
2. 细粒度监听：只关心特定配置项
3. API方式：通过HTTP API管理配置
4. YAML支持：复杂的嵌套配置

热更新是个很实用的功能，能大大提升自动化脚本的灵活性。我是扣扣，有问题欢迎留言~🙃