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

wen 实用脚本 2026-07-19 1

本文目录导读：

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

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

对于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