Python脚本如何热更新配置内容

本文目录导读：

1. 使用配置文件 + 定时检查
2. 使用 watchdog 库（更高效）
3. 使用环境变量（适合容器化部署）
4. 基于Web的管理接口
5. 推荐的最佳实践
6. 注意事项

在Python中实现配置热更新（无需重启应用即可更新配置），主要有以下几种常见方法：

使用配置文件 + 定时检查

最简单的实现方式

import time
import json
import os
from threading import Thread, Lock
class ConfigManager:
    def __init__(self, config_path, check_interval=5):
        self.config_path = config_path
        self.config = {}
        self.last_mtime = 0
        self.check_interval = check_interval
        self._lock = Lock()
        self._load_config()
        # 启动后台检查线程
        self._watch_thread = Thread(target=self._watch_config, daemon=True)
        self._watch_thread.start()
    def _load_config(self):
        """加载配置文件"""
        try:
            with open(self.config_path, 'r', encoding='utf-8') as f:
                new_config = json.load(f)
            with self._lock:
                self.config = new_config
            self.last_mtime = os.path.getmtime(self.config_path)
            print(f"配置已加载: {self.config}")
        except Exception as e:
            print(f"加载配置失败: {e}")
    def _watch_config(self):
        """监控配置文件变化"""
        while True:
            try:
                if os.path.exists(self.config_path):
                    current_mtime = os.path.getmtime(self.config_path)
                    if current_mtime > self.last_mtime:
                        print("检测到配置文件变更，重新加载...")
                        self._load_config()
            except Exception as e:
                print(f"监控配置出错: {e}")
            time.sleep(self.check_interval)
    def get(self, key, default=None):
        with self._lock:
            return self.config.get(key, default)
    def get_all(self):
        with self._lock:
            return self.config.copy()
# 使用示例
config = ConfigManager('config.json', check_interval=2)
def main():
    while True:
        # 始终获取最新配置
        db_host = config.get('db_host', 'localhost')
        debug_mode = config.get('debug', False)
        print(f"当前配置 - 数据库: {db_host}, 调试模式: {debug_mode}")
        time.sleep(3)
# 运行主程序
# main()

使用 watchdog 库（更高效）

pip install watchdog

import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import json
class ConfigWatcher(FileSystemEventHandler):
    def __init__(self, callback):
        self.callback = callback
        self.last_modified = 0
    def on_modified(self, event):
        if event.is_directory:
            return
        if event.src_path.endswith('.json'):
            current_time = time.time()
            if current_time - self.last_modified > 1:  # 防抖
                self.last_modified = current_time
                print(f"配置变更: {event.src_path}")
                self.callback(event.src_path)
class HotConfig:
    def __init__(self, config_path):
        self.config_path = config_path
        self.config = {}
        self._load_config()
        # 设置文件监控
        self._setup_watcher()
    def _load_config(self):
        """加载配置"""
        try:
            with open(self.config_path, 'r', encoding='utf-8') as f:
                self.config = json.load(f)
            print(f"配置已更新: {self.config}")
        except Exception as e:
            print(f"配置加载失败: {e}")
    def _setup_watcher(self):
        """设置文件监控"""
        import os
        self.observer = Observer()
        handler = ConfigWatcher(lambda path: self._load_config())
        # 监控配置文件的目录
        watch_dir = os.path.dirname(os.path.abspath(self.config_path))
        self.observer.schedule(handler, watch_dir, recursive=False)
        self.observer.start()
    def get(self, key, default=None):
        return self.config.get(key, default)
    def update(self, key, value):
        """动态更新配置并保存到文件"""
        self.config[key] = value
        self._save_config()
    def _save_config(self):
        """保存配置到文件"""
        try:
            with open(self.config_path, 'w', encoding='utf-8') as f:
                json.dump(self.config, f, indent=2, ensure_ascii=False)
        except Exception as e:
            print(f"保存配置失败: {e}")
# 使用示例
config = HotConfig('config.json')
# 动态更新配置
config.update('max_connections', 100)
config.update('api_timeout', 30)
# 获取配置
print(config.get('max_connections'))

使用环境变量（适合容器化部署）

import os
import json
from typing import Any
class EnvConfig:
    def __init__(self, prefix="APP_"):
        self.prefix = prefix
        self.config = {}
        self._load_from_env()
    def _load_from_env(self):
        """从环境变量加载配置"""
        for key, value in os.environ.items():
            if key.startswith(self.prefix):
                config_key = key[len(self.prefix):].lower()
                # 尝试转换类型
                self.config[config_key] = self._parse_value(value)
    def _parse_value(self, value: str) -> Any:
        """智能解析配置值"""
        # 尝试转换为数字
        try:
            if '.' in value:
                return float(value)
            return int(value)
        except ValueError:
            pass
        # 布尔值
        if value.lower() in ('true', 'yes', '1'):
            return True
        if value.lower() in ('false', 'no', '0'):
            return False
        return value
    def get(self, key, default=None):
        return self.config.get(key, default)
    def refresh(self):
        """手动刷新配置"""
        self._load_from_env()
    def __getattr__(self, name):
        return self.config.get(name, None)
# 使用示例
config = EnvConfig()
# 设置环境变量（实际生产中在部署环境设置）
os.environ['APP_DB_HOST'] = 'localhost'
os.environ['APP_DEBUG'] = 'true'
os.environ['APP_PORT'] = '8080'
# 获取配置
print(config.get('db_host'))    # localhost
print(config.get('debug'))      # True
print(config.get('port'))       # 8080

基于Web的管理接口

from flask import Flask, request, jsonify
import threading
import json
class WebConfig:
    def __init__(self, config_path, web_port=5000):
        self.config_path = config_path
        self.config = {}
        self.lock = threading.Lock()
        self._load_config()
        # 启动Web服务
        self.app = Flask(__name__)
        self._setup_routes()
        # 在后台线程启动Flask
        thread = threading.Thread(target=self._run_web, daemon=True)
        thread.start()
    def _load_config(self):
        """加载配置"""
        try:
            with open(self.config_path, 'r', encoding='utf-8') as f:
                with self.lock:
                    self.config = json.load(f)
        except Exception as e:
            print(f"配置加载失败: {e}")
    def _save_config(self):
        """保存配置"""
        with self.lock:
            with open(self.config_path, 'w', encoding='utf-8') as f:
                json.dump(self.config, f, indent=2, ensure_ascii=False)
    def _setup_routes(self):
        @self.app.route('/config', methods=['GET'])
        def get_config():
            with self.lock:
                return jsonify(self.config)
        @self.app.route('/config', methods=['POST'])
        def update_config():
            data = request.json
            if not data:
                return jsonify({"error": "No data provided"}), 400
            with self.lock:
                self.config.update(data)
                self._save_config()
            return jsonify({"message": "配置更新成功", "config": self.config})
        @self.app.route('/config/<key>', methods=['PUT'])
        def update_single_config(key):
            data = request.json
            if not data or 'value' not in data:
                return jsonify({"error": "需要提供value字段"}), 400
            with self.lock:
                self.config[key] = data['value']
                self._save_config()
            return jsonify({"message": f"配置 {key} 已更新"})
    def _run_web(self):
        self.app.run(host='0.0.0.0', port=5000, debug=False)
    def get(self, key, default=None):
        with self.lock:
            return self.config.get(key, default)
# 使用示例
config = WebConfig('config.json', web_port=5000)
# 通过API更新配置
# curl -X POST http://localhost:5000/config -H "Content-Type: application/json" -d '{"debug": true, "max_connections": 200}'
# 或直接代码更新
def update_config_externally():
    import requests
    requests.post('http://localhost:5000/config', 
                  json={'debug': True, 'max_connections': 200})

推荐的最佳实践

结合多种方式的热更新管理器

import json
import os
import time
import hashlib
from threading import Thread, Lock
from typing import Any, Dict, Optional
class AdvancedHotConfig:
    """高级热更新配置管理器"""
    def __init__(self, 
                 config_path: str,
                 check_interval: int = 5,
                 backup: bool = True):
        self.config_path = config_path
        self.check_interval = check_interval
        self._config: Dict[str, Any] = {}
        self._file_hash = ''
        self._lock = Lock()
        self._callbacks = []
        self._backup = backup
        self._init_config()
        self._start_watching()
    def _init_config(self):
        """初始化配置"""
        if not os.path.exists(self.config_path):
            # 创建默认配置
            default_config = {
                'debug': False,
                'max_connections': 50,
                'timeout': 30,
                'log_level': 'INFO'
            }
            self._save_config(default_config)
        else:
            self._load_config()
    def _load_config(self):
        """加载配置"""
        try:
            with open(self.config_path, 'r', encoding='utf-8') as f:
                content = f.read()
                new_hash = hashlib.md5(content.encode()).hexdigest()
                if new_hash != self._file_hash:
                    new_config = json.loads(content)
                    with self._lock:
                        old_config = self._config.copy()
                        self._config = new_config
                        self._file_hash = new_hash
                    # 通知监听器
                    self._notify_listeners(old_config, new_config)
                    print(f"[{time.strftime('%H:%M:%S')}] 配置已更新")
        except Exception as e:
            print(f"加载配置失败: {e}")
    def _save_config(self, config: Dict[str, Any] = None):
        """保存配置"""
        if config is None:
            config = self._config
        try:
            # 备份旧配置
            if self._backup and os.path.exists(self.config_path):
                backup_path = f"{self.config_path}.bak"
                import shutil
                shutil.copy2(self.config_path, backup_path)
            with open(self.config_path, 'w', encoding='utf-8') as f:
                json.dump(config, f, indent=2, ensure_ascii=False)
            # 更新hash
            content = json.dumps(config, indent=2, ensure_ascii=False)
            self._file_hash = hashlib.md5(content.encode()).hexdigest()
        except Exception as e:
            print(f"保存配置失败: {e}")
    def _start_watching(self):
        """启动配置监控"""
        def watch_loop():
            while True:
                try:
                    if os.path.exists(self.config_path):
                        self._load_config()
                except Exception as e:
                    print(f"监控配置出错: {e}")
                time.sleep(self.check_interval)
        thread = Thread(target=watch_loop, daemon=True)
        thread.start()
    def _notify_listeners(self, old_config: Dict, new_config: Dict):
        """通知配置变更监听器"""
        changes = {}
        for key in set(list(old_config.keys()) + list(new_config.keys())):
            if old_config.get(key) != new_config.get(key):
                changes[key] = {
                    'old': old_config.get(key),
                    'new': new_config.get(key)
                }
        if changes:
            for callback in self._callbacks:
                try:
                    callback(changes)
                except Exception as e:
                    print(f"回调执行失败: {e}")
    def on_change(self, callback):
        """注册配置变更回调"""
        self._callbacks.append(callback)
        return callback
    def get(self, key: str, default: Any = None) -> Any:
        """获取配置值"""
        with self._lock:
            return self._config.get(key, default)
    def update(self, key: str, value: Any):
        """更新单个配置"""
        with self._lock:
            self._config[key] = value
        self._save_config()
    def batch_update(self, updates: Dict[str, Any]):
        """批量更新配置"""
        with self._lock:
            self._config.update(updates)
        self._save_config()
    def get_all(self) -> Dict[str, Any]:
        """获取所有配置"""
        with self._lock:
            return self._config.copy()
    def reload(self):
        """强制重新加载配置"""
        self._load_config()
# 配置变更回调示例
def on_config_change(changes):
    print(f"配置变更: {changes}")
    for key, change in changes.items():
        if key == 'debug':
            print(f"调试模式: {change['new']}")
        elif key == 'max_connections':
            print(f"最大连接数已调整为: {change['new']}")
# 使用示例
config = AdvancedHotConfig('config.json', check_interval=3)
# 注册配置变更监听器
config.on_change(on_config_change)
# 主程序逻辑
def main():
    while True:
        debug_mode = config.get('debug', False)
        max_conn = config.get('max_connections', 50)
        if debug_mode:
            print(f"[调试模式] 当前配置: {config.get_all()}")
        else:
            # 正常业务逻辑
            pass
        time.sleep(10)
# 运行主程序
if __name__ == '__main__':
    # main()
    pass

注意事项

1. 线程安全：使用锁机制保护配置读写
2. 异常处理：配置加载失败时使用默认值
3. 性能考虑：避免频繁读取配置文件
4. 配置文件格式：推荐使用JSON或YAML
5. 配置验证：加载配置后验证必要字段
6. 版本控制：考虑配置文件的版本管理
7. 监控告警：配置变更通知和日志记录

选择哪种方式取决于你的具体需求：

• 简单应用：定时检查文件修改
• 生产环境：watchdog + 配置文件
• 容器部署：环境变量
• 需要远程管理：Web接口