Python脚本配置中心如何动态更新

本文目录导读：

1. 文件监听方案 (简单直接)
2. Redis配置中心 (分布式方案)
3. 基于etcd的方案 (生产级)
4. 基于Consul的方案 (云原生)
5. 自定义观察者模式 (轻量级)
6. 最佳实践建议

对于Python脚本配置中心的动态更新,有几种成熟方案：

文件监听方案 (简单直接)

import time
import json
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class Config:
    def __init__(self, config_path):
        self.config_path = config_path
        self.config = {}
        self._load_config()
    def _load_config(self):
        with open(self.config_path, 'r') as f:
            self.config = json.load(f)
    def get(self, key, default=None):
        return self.config.get(key, default)
class ConfigWatcher(FileSystemEventHandler):
    def __init__(self, config):
        self.config = config
    def on_modified(self, event):
        if event.src_path.endswith('config.json'):
            print("配置已更新，重新加载...")
            self.config._load_config()
# 使用
config = Config('config.json')
observer = Observer()
observer.schedule(ConfigWatcher(config), path='.')
observer.start()

Redis配置中心 (分布式方案)

import redis
import json
import threading
class RedisConfigCenter:
    def __init__(self, redis_client, namespace='app_config'):
        self.redis = redis_client
        self.namespace = namespace
        self.config = {}
        self._callbacks = []
        self._load_config()
        self._start_listener()
    def _load_config(self):
        raw = self.redis.get(self.namespace)
        if raw:
            self.config = json.loads(raw)
    def _start_listener(self):
        pubsub = self.redis.pubsub()
        pubsub.subscribe(**{f'{self.namespace}:update': self._on_update})
        thread = threading.Thread(target=pubsub.run_in_thread, daemon=True)
        thread.start()
    def _on_update(self, message):
        self._load_config()
        for callback in self._callbacks:
            callback(self.config)
    def watch(self, key, callback):
        """监听特定配置变化"""
        self._callbacks.append(lambda config: 
            callback(config.get(key)) if key in config else None)
# 使用
r = redis.Redis(host='localhost', port=6379)
config_center = RedisConfigCenter(r)
# 监听特定配置变化
config_center.watch('timeout', lambda timeout: 
    print(f"超时配置更新为: {timeout}"))

基于etcd的方案 (生产级)

import etcd3
import json
class EtcdConfigClient:
    def __init__(self, host='localhost', port=2379):
        self.client = etcd3.client(host=host, port=port)
        self.config = {}
        self._watch_prefix('/app/config/')
    def _watch_prefix(self, prefix):
        events_iterator, cancel = self.client.watch_prefix(prefix)
        for event in events_iterator:
            key = event.key.decode()
            if event.type == 'PUT':
                value = json.loads(event.value.decode())
                config_key = key.split('/')[-1]
                self.config[config_key] = value
                self._notify(config_key, value)
            elif event.type == 'DELETE':
                config_key = key.split('/')[-1]
                self.config.pop(config_key, None)
    def _notify(self, key, value):
        # 触发回调
        print(f"配置 {key} 已更新: {value}")
    def get_config(self, key):
        result = self.client.get(f'/app/config/{key}')
        if result[0]:
            return json.loads(result[0])
        return None
# 使用
config = EtcdConfigClient()
timeout = config.get_config('timeout')

基于Consul的方案 (云原生)

import consul
import json
class ConsulConfigCenter:
    def __init__(self, host='localhost', port=8500):
        self.client = consul.Consul(host=host, port=port)
        self.config = {}
        self._watch_config()
    def _watch_config(self):
        index = None
        while True:
            index, data = self.client.kv.get('app/config/', 
                                             index=index, 
                                             wait='10s')
            if data:
                self._parse_config(data)
    def _parse_config(self, data):
        for item in data:
            key = item['Key'].split('/')[-1]
            value = json.loads(item['Value'])
            self.config[key] = value
    def get_config(self, key):
        return self.config.get(key)
# 使用
config_center = ConsulConfigCenter()
database_url = config_center.get_config('database_url')

自定义观察者模式 (轻量级)

class ObservableConfig:
    def __init__(self):
        self._config = {}
        self._observers = {}
    def set_config(self, key, value):
        old_value = self._config.get(key)
        self._config[key] = value
        if key in self._observers and old_value != value:
            self._notify(key, old_value, value)
    def observe(self, key, callback):
        if key not in self._observers:
            self._observers[key] = []
        self._observers[key].append(callback)
    def _notify(self, key, old_value, new_value):
        for callback in self._observers[key]:
            callback(key, old_value, new_value)
    def get(self, key, default=None):
        return self._config.get(key, default)
# 使用
config = ObservableConfig()
# 注册观察者
config.observe('timeout', lambda key, old, new: 
    print(f"超时从 {old} 更新为 {new}"))
# 模拟配置更新
config.set_config('timeout', 30)  # 触发回调
config.set_config('timeout', 60)  # 再次触发

最佳实践建议

选择指南：

• 小型项目/简单场景: 文件监听方案
• 分布式系统: Redis/etcd/Consul
• 云原生/K8s: Consul或ConfigMap
• 微服务架构: etcd或Nacos

关键注意事项：

# 1. 配置验证
def validate_config(config):
    required_keys = ['host', 'port', 'timeout']
    for key in required_keys:
        if key not in config:
            raise ValueError(f"缺少必要配置: {key}")
# 2. 优雅回退
def safe_get_config(key, default=None):
    try:
        return config_center.get_config(key)
    except ConnectionError:
        return default
# 3. 配置变更日志
import logging
logging.basicConfig(level=logging.INFO)
def config_change_logger(key, old_val, new_val):
    logging.info(f"配置变更: {key}: {old_val} -> {new_val}")
# 4. 并发安全
import threading
class ThreadSafeConfig:
    def __init__(self):
        self._lock = threading.Lock()
        self._config = {}
    def update(self, key, value):
        with self._lock:
            old = self._config.get(key)
            self._config[key] = value
            return old

推荐组合：

• 开发环境: 文件监听 + 观察者模式
• 生产环境: etcd/Consul + Redis作为缓存层
• 容器化环境: 使用ConfigMap + 侧车容器同步

选择方案时需要考虑：团队技术栈、基础设施、性能要求、运维复杂度等因素。