# Python全局字典构建的全面指南与最佳实践

## 引言：全局字典在Python应用中的角色

在Python应用程序开发中，全局字典作为一种共享数据存储机制，扮演着至关重要的角色。它们可以用于配置管理、状态跟踪、缓存实现、服务注册等多个场景。然而，不当的全局字典使用可能导致数据竞争、难以调试和维护问题。本文将深入探讨在Python中构建和管理全局字典的各种方法、设计模式以及最佳实践。

## 基础全局字典实现

### 模块级全局字典

最简单直接的全局字典实现是在模块级别定义字典变量。

```python

# config_store.py

"""模块级全局字典示例"""

# 基础全局字典

global_config = {

    "app_name": "MyApplication",

    "version": "1.0.0",

    "debug": False,

    "max_connections": 100,

    "timeout": 30.0

}

# 只读全局字典（约定俗成，实际并非真正只读）

_readonly_settings = {

    "api_version": "v2",

    "supported_formats": ["json", "xml"]

}

def get_setting(key, default=None):

    """安全获取配置项"""

    return global_config.get(key, default)

def update_settings(updates):

    """安全更新配置"""

    if not isinstance(updates, dict):

        raise TypeError("更新数据必须是字典")

    

    # 验证关键配置

    if "max_connections" in updates:

        if not isinstance(updates["max_connections"], int) or updates["max_connections"] <= 0:

            raise ValueError("max_connections必须是正整数")

    

    global_config.update(updates)

    return True

def get_readonly_setting(key):

    """获取只读配置"""

    return _readonly_settings.get(key)

# 使用示例

if __name__ == "__main__":

    # 访问配置

    print(f"应用名称: {get_setting('app_name')}")

    

    # 更新配置

    update_settings({"debug": True, "timeout": 45.0})

    print(f"调试模式: {get_setting('debug')}")

```

### 类包装的全局字典

使用类包装可以提供更好的封装和控制。

```python

# global_registry.py

"""类封装的全局字典"""

class GlobalRegistry:

    """全局注册表管理器"""

    

    _instance = None

    _registry = {}

    _locks = {}  # 用于线程安全的锁

    

    def __new__(cls):

        """单例模式实现"""

        if cls._instance is None:

            cls._instance = super().__new__(cls)

            cls._instance._initialize()

        return cls._instance

    

    def _initialize(self):

        """初始化注册表"""

        import threading

        self._lock = threading.RLock()

        

        # 初始化默认命名空间

        self._registry = {

            "config": {},

            "cache": {},

            "services": {},

            "state": {}

        }

    

    def set(self, namespace, key, value):

        """

        设置注册表项

        

        Args:

            namespace: 命名空间（如'config', 'cache'）

            key: 键名

            value: 值

        """

        with self._lock:

            if namespace not in self._registry:

                self._registry[namespace] = {}

            self._registry[namespace][key] = value

    

    def get(self, namespace, key, default=None):

        """

        获取注册表项

        

        Args:

            namespace: 命名空间

            key: 键名

            default: 默认值

        """

        with self._lock:

            namespace_dict = self._registry.get(namespace)

            if namespace_dict is None:

                return default

            return namespace_dict.get(key, default)

    

    def delete(self, namespace, key):

        """删除注册表项"""

        with self._lock:

            if namespace in self._registry:

                self._registry[namespace].pop(key, None)

    

    def clear_namespace(self, namespace):

        """清空命名空间"""

        with self._lock:

            if namespace in self._registry:

                self._registry[namespace].clear()

    

    def get_namespace(self, namespace):

        """获取整个命名空间的拷贝"""

        with self._lock:

            if namespace in self._registry:

                return self._registry[namespace].copy()

            return {}

    

    def has_namespace(self, namespace):

        """检查命名空间是否存在"""

        return namespace in self._registry

    

    def list_namespaces(self):

        """列出所有命名空间"""

        return list(self._registry.keys())

    

    def dump(self):

        """导出整个注册表状态（用于调试）"""

        import json

        with self._lock:

            # 转换不可JSON序列化的对象

            serializable = {}

            for ns, data in self._registry.items():

                serializable[ns] = {}

                for k, v in data.items():

                    try:

                        json.dumps(v)

                        serializable[ns][k] = v

                    except:

                        serializable[ns][k] = str(v)

            

            return json.dumps(serializable, indent=2, ensure_ascii=False)

# 便捷访问函数

def get_registry():

    """获取全局注册表实例"""

    return GlobalRegistry()

def set_global(namespace, key, value):

    """便捷设置函数"""

    get_registry().set(namespace, key, value)

def get_global(namespace, key, default=None):

    """便捷获取函数"""

    return get_registry().get(namespace, key, default)

# 使用示例

if __name__ == "__main__":

    # 设置配置

    set_global("config", "database_url", "postgresql://user:pass@localhost/db")

    set_global("config", "max_workers", 4)

    

    # 设置缓存

    set_global("cache", "user_123", {"name": "Alice", "role": "admin"})

    

    # 获取值

    db_url = get_global("config", "database_url")

    print(f"数据库URL: {db_url}")

    

    # 导出状态

    registry = get_registry()

    print("注册表状态:")

    print(registry.dump())

```

## 线程安全的全局字典

### 使用锁机制的线程安全字典

```python

# thread_safe_dict.py

"""线程安全的全局字典实现"""

import threading

import time

from typing import Any, Dict, Optional, TypeVar, Generic

from contextlib import contextmanager

from collections.abc import MutableMapping

T = TypeVar('T')

class ThreadSafeDict(MutableMapping):

    """线程安全的字典实现"""

    

    def __init__(self, *args, **kwargs):

        self._data = dict(*args, **kwargs)

        self._lock = threading.RLock()

        self._condition = threading.Condition(self._lock)

    

    def __getitem__(self, key):

        with self._lock:

            return self._data[key]

    

    def __setitem__(self, key, value):

        with self._lock:

            self._data[key] = value

            # 通知等待的线程

            self._condition.notify_all()

    

    def __delitem__(self, key):

        with self._lock:

            del self._data[key]

    

    def __iter__(self):

        with self._lock:

            # 返回键的拷贝以避免迭代时修改

            return iter(list(self._data.keys()))

    

    def __len__(self):

        with self._lock:

            return len(self._data)

    

    def __repr__(self):

        with self._lock:

            return f"ThreadSafeDict({self._data})"

    

    def get(self, key, default=None):

        """线程安全的get方法"""

        with self._lock:

            return self._data.get(key, default)

    

    def setdefault(self, key, default=None):

        """线程安全的setdefault方法"""

        with self._lock:

            return self._data.setdefault(key, default)

    

    def pop(self, key, default=None):

        """线程安全的pop方法"""

        with self._lock:

            if key in self._data:

                return self._data.pop(key)

            return default

    

    def update(self, other=None, **kwargs):

        """线程安全的update方法"""

        with self._lock:

            if other is not None:

                if hasattr(other, "items"):

                    for k, v in other.items():

                        self._data[k] = v

                else:

                    for k, v in other:

                        self._data[k] = v

            

            for k, v in kwargs.items():

                self._data[k] = v

    

    def clear(self):

        """清空字典"""

        with self._lock:

            self._data.clear()

    

    def copy(self):

        """返回字典的深拷贝"""

        with self._lock:

            import copy

            return copy.deepcopy(self._data)

    

    def wait_for_key(self, key, timeout=None):

        """

        等待键出现

        

        Args:

            key: 等待的键

            timeout: 超时时间（秒）

        

        Returns:

            键的值或None（如果超时）

        """

        with self._condition:

            end_time = time.time() + timeout if timeout else None

            

            while key not in self._data:

                if timeout:

                    remaining = end_time - time.time()

                    if remaining <= 0:

                        return None

                    self._condition.wait(remaining)

                else:

                    self._condition.wait()

            

            return self._data[key]

    

    def atomic_update(self, key, update_func):

        """

        原子更新操作

        

        Args:

            key: 键名

            update_func: 更新函数，接受当前值，返回新值

        """

        with self._lock:

            current = self._data.get(key)

            new_value = update_func(current)

            self._data[key] = new_value

            return new_value

    

    @contextmanager

    def transaction(self):

        """

        事务上下文管理器

        

        用法:

        with dict_instance.transaction():

            dict_instance['key1'] = 'value1'

            dict_instance['key2'] = dict_instance['key1'] + '_suffix'

        """

        with self._lock:

            try:

                yield self

            except Exception:

                # 事务失败，可以选择回滚

                raise

# 全局线程安全字典管理器

class GlobalThreadSafeDict:

    """管理多个线程安全字典的全局管理器"""

    

    _instance = None

    _dicts = {}

    _creation_lock = threading.Lock()

    

    def __new__(cls):

        if cls._instance is None:

            with cls._creation_lock:

                if cls._instance is None:

                    cls._instance = super().__new__(cls)

                    cls._instance._initialize()

        return cls._instance

    

    def _initialize(self):

        self._dicts = ThreadSafeDict()

    

    def get_dict(self, name: str) -> ThreadSafeDict:

        """

        获取或创建命名的线程安全字典

        

        Args:

            name: 字典名称

        

        Returns:

            ThreadSafeDict实例

        """

        if name not in self._dicts:

            with self._dicts.transaction():

                if name not in self._dicts:

                    self._dicts[name] = ThreadSafeDict()

        

        return self._dicts[name]

    

    def remove_dict(self, name: str):

        """移除命名字典"""

        with self._dicts.transaction():

            if name in self._dicts:

                del self._dicts[name]

    

    def list_dicts(self):

        """列出所有字典名称"""

        return list(self._dicts.keys())

# 使用示例

def demo_thread_safe_dict():

    """演示线程安全字典的使用"""

www.ko.zizhitong.cn>www.qm.zizhitong.cn

    

    # 创建全局管理器

    manager = GlobalThreadSafeDict()

    

    # 获取配置字典

    config_dict = manager.get_dict("config")

    

    # 多个线程同时操作

    def worker_thread(thread_id):

        # 原子更新计数器

        def increment_counter(current):

            return (current or 0) + 1

        

        config_dict.atomic_update("request_count", increment_counter)

        

        # 设置线程特定的数据

        config_dict[f"thread_{thread_id}_data"] = {

            "started_at": time.time(),

            "thread_id": thread_id

        }

        

        # 等待共享资源

        if thread_id == 0:

            config_dict["shared_resource"] = "ready"

        else:

            # 其他线程等待资源就绪

            value = config_dict.wait_for_key("shared_resource", timeout=5)

            print(f"线程 {thread_id} 获取到资源: {value}")

    

    # 启动多个线程

    threads = []

    for i in range(5):

        t = threading.Thread(target=worker_thread, args=(i,))

        threads.append(t)

        t.start()

    

    # 等待所有线程完成

    for t in threads:

        t.join()

    

    # 输出结果

    print("最终配置字典:")

    print(config_dict.copy())

    print(f"总请求数: {config_dict.get('request_count', 0)}")

if __name__ == "__main__":

    demo_thread_safe_dict()

```

## 基于装饰器的全局字典访问

### 装饰器实现

```python

# global_dict_decorators.py

"""全局字典的装饰器实现"""

import functools

from typing import Any, Callable, Optional

class GlobalDictDecorators:

    """全局字典装饰器集合"""

    

    _storage = {}

    

    @classmethod

    def initialize(cls):

        """初始化存储"""

        cls._storage = {

            "cache": {},

            "config": {},

            "state": {},

            "metrics": {}

        }

    

    @classmethod

    def get_storage(cls, category="cache"):

        """获取存储字典"""

        if category not in cls._storage:

            cls._storage[category] = {}

        return cls._storage[category]

    

    @classmethod

    def cached(cls, category="cache", ttl=None):

        """

        缓存装饰器

        

        Args:

            category: 缓存类别

            ttl: 缓存存活时间（秒）

        """

        import time

        

        def decorator(func):

            @functools.wraps(func)

            def wrapper(*args, **kwargs):

                # 生成缓存键

                cache_key = f"{func.__module__}.{func.__name__}:{args}:{sorted(kwargs.items())}"

                cache_dict = cls.get_storage(category)

                

                # 检查缓存

                if cache_key in cache_dict:

                    cached_data = cache_dict[cache_key]

                    

                    # 检查TTL

                    if ttl is not None:

                        timestamp, value = cached_data

                        if time.time() - timestamp < ttl:

                            return value

                        else:

                            # 缓存过期，删除

                            del cache_dict[cache_key]

                    else:

                        return cached_data

                

                # 执行函数

                result = func(*args, **kwargs)

                

                # 缓存结果

                if ttl is not None:

                    cache_dict[cache_key] = (time.time(), result)

                else:

                    cache_dict[cache_key] = result

                

                return result

            

            return wrapper

        

        return decorator

    

    @classmethod

    def configurable(cls, config_key, default=None):

        """

        可配置的装饰器

        

        Args:

            config_key: 配置键名

            default: 默认值

        """

        def decorator(func):

            @functools.wraps(func)

            def wrapper(*args, **kwargs):

                # 获取配置

                config_dict = cls.get_storage("config")

                config_value = config_dict.get(config_key, default)

                

                # 将配置作为额外参数传递

                return func(*args, config=config_value, **kwargs)

            

            return wrapper

        

        return decorator

    

    @classmethod

    def stateful(cls, state_key):

        """

        有状态的装饰器

        

        Args:

            state_key: 状态键名

        """

        def decorator(func):

            @functools.wraps(func)

            def wrapper(*args, **kwargs):

                state_dict = cls.get_storage("state")

                

                # 获取或初始化状态

                if state_key not in state_dict:

                    state_dict[state_key] = {"call_count": 0}

                

                state = state_dict[state_key]

                state["call_count"] += 1

                state["last_called"] = time.time()

                

                try:

                    result = func(*args, **kwargs)

                    state["last_success"] = time.time()

                    return result

                except Exception as e:

                    state["last_error"] = str(e)

                    state["error_time"] = time.time()

                    raise

            

            return wrapper

        

        return decorator

    

    @classmethod

    def with_metrics(cls, metric_name):

        """

        带指标收集的装饰器

        

        Args:

            metric_name: 指标名称

        """

        import time

        

        def decorator(func):

            @functools.wraps(func)

            def wrapper(*args, **kwargs):

                start_time = time.time()

                metrics_dict = cls.get_storage("metrics")

                

                if metric_name not in metrics_dict:

                    metrics_dict[metric_name] = {

                        "count": 0,

                        "total_time": 0,

                        "errors": 0

                    }

                

                metrics = metrics_dict[metric_name]

                metrics["count"] += 1

                

                try:

                    result = func(*args, **kwargs)

                    execution_time = time.time() - start_time

                    metrics["total_time"] += execution_time

                    metrics["avg_time"] = metrics["total_time"] / metrics["count"]

                    return result

                except Exception:

                    metrics["errors"] += 1

                    raise

            

            return wrapper

        

        return decorator

# 使用示例

if __name__ == "__main__":

    GlobalDictDecorators.initialize()

    

    # 设置配置

    config_storage = GlobalDictDecorators.get_storage("config")

    config_storage["max_retries"] = 3

    config_storage["timeout"] = 10

    

    # 使用装饰器

    @GlobalDictDecorators.cached(category="api_cache", ttl=300)

    def fetch_user_data(user_id):

        print(f"从API获取用户数据: {user_id}")

        return {"id": user_id, "name": f"User_{user_id}", "active": True}

    

    @GlobalDictDecorators.configurable("max_retries", default=2)

    def retry_operation(operation, config=None):

        print(f"执行操作，最大重试次数: {config}")

        return f"操作完成，重试配置: {config}"

    

    @GlobalDictDecorators.stateful("data_processor")

    def process_data(data):

        print("处理数据...")

        return {"processed": True, "data": data}

    

    @GlobalDictDecorators.with_metrics("api_requests")

    def make_api_request(endpoint):

        print(f"请求API: {endpoint}")

        time.sleep(0.1)  # 模拟延迟

        return {"status": "success", "endpoint": endpoint}

    

    # 测试函数

    print("测试缓存装饰器:")

    print(fetch_user_data(1))

    print(fetch_user_data(1))  # 应该从缓存获取

    

    print("\n测试可配置装饰器:")

    print(retry_operation("重要操作"))

    

    print("\n测试有状态装饰器:")

    print(process_data({"test": "data"}))

    print(process_data({"test": "data2"}))

    

    print("\n测试指标装饰器:")

    print(make_api_request("/users"))

    print(make_api_request("/products"))

    

    # 查看状态和指标

    print("\n状态存储:")

    print(GlobalDictDecorators.get_storage("state"))

    

    print("\n指标存储:")

    print(GlobalDictDecorators.get_storage("metrics"))

```

## 应用场景实现

### 配置管理系统

```python

# configuration_manager.py

"""基于全局字典的配置管理系统"""

import json

import yaml

import os

from pathlib import Path

from typing import Dict, Any, Optional, Union

from threading import Lock

www.rp.zizhitong.cn>www.py.zizhitong.cn

class ConfigurationManager:

    """配置管理器"""

    

    _instance = None

    _config_store = {}

    _lock = Lock()

    _config_versions = {}

    

    # 配置验证规则

    VALIDATION_RULES = {

        "app.name": {"type": str, "required": True},

        "app.version": {"type": str, "required": True},

        "app.debug": {"type": bool, "default": False},

        "database.url": {"type": str, "required": True},

        "database.pool_size": {"type": int, "min": 1, "max": 100, "default": 10},

        "redis.host": {"type": str, "default": "localhost"},

        "redis.port": {"type": int, "min": 1, "max": 65535, "default": 6379},

        "logging.level": {"type": str, "choices": ["DEBUG", "INFO", "WARNING", "ERROR"], "default": "INFO"},

        "api.timeout": {"type": (int, float), "min": 0, "default": 30},

    }

    

    def __new__(cls):

        if cls._instance is None:

            with cls._lock:

                if cls._instance is None:

                    cls._instance = super().__new__(cls)

                    cls._instance._initialize()

        return cls._instance

    

    def _initialize(self):

        """初始化配置存储"""

        self._config_store = {}

        self._config_versions = {}

        self._load_defaults()

    

    def _load_defaults(self):

        """加载默认配置"""

        defaults = {

            "app": {

                "name": "MyApplication",

                "version": "1.0.0",

                "debug": False

            },

            "logging": {

                "level": "INFO",

                "format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s"

            }

        }

        self._merge_config(defaults)

    

    def _merge_config(self, new_config: Dict, prefix: str = ""):

        """合并配置字典"""

        for key, value in new_config.items():

            full_key = f"{prefix}.{key}" if prefix else key

            

            if isinstance(value, dict):

                self._merge_config(value, full_key)

            else:

                self._config_store[full_key] = value

    

    def load_from_file(self, filepath: Union[str, Path]):

        """

        从文件加载配置

        

        Args:

            filepath: 配置文件路径

        

        Raises:

            FileNotFoundError: 文件不存在

            ValueError: 配置文件格式错误

        """

        filepath = Path(filepath)

        if not filepath.exists():

            raise FileNotFoundError(f"配置文件不存在: {filepath}")

        

        with open(filepath, 'r', encoding='utf-8') as f:

            if filepath.suffix.lower() in ['.yaml', '.yml']:

                config_data = yaml.safe_load(f)

            elif filepath.suffix.lower() == '.json':

                config_data = json.load(f)

            else:

                raise ValueError(f"不支持的配置文件格式: {filepath.suffix}")

        

        # 验证配置

        self._validate_config(config_data)

        

        # 保存版本

        version_key = f"config_version_{len(self._config_versions)}"

        self._config_versions[version_key] = self._config_store.copy()

        

        # 合并新配置

        self._merge_config(config_data)

    

    def load_from_env(self, prefix: str = "APP_"):

        """

        从环境变量加载配置

        

        Args:

            prefix: 环境变量前缀

        """

        env_config = {}

        

        for env_key, env_value in os.environ.items():

            if env_key.startswith(prefix):

                # 转换环境变量名：APP_DATABASE_URL -> database.url

                config_key = env_key[len(prefix):].lower().replace('_', '.')

                

                # 尝试转换类型

                try:

                    # 尝试转换为数字

                    if env_value.isdigit():

                        env_value = int(env_value)

                    elif env_value.replace('.', '', 1).isdigit():

                        env_value = float(env_value)

                    elif env_value.lower() in ['true', 'false']:

                        env_value = env_value.lower() == 'true'

                except (ValueError, AttributeError):

                    pass

                

                # 设置配置值

                self._set_dotted_key(config_key, env_value)

    

    def _set_dotted_key(self, dotted_key: str, value: Any):

        """设置点分隔的键"""

        parts = dotted_key.split('.')

        

        current = self._config_store

        for i, part in enumerate(parts[:-1]):

            if part not in current:

                current[part] = {}

            current = current[part]

        

        current[parts[-1]] = value

    

    def get(self, key: str, default: Any = None) -> Any:

        """

        获取配置值

        

        Args:

            key: 配置键（支持点分隔）

            default: 默认值

        

        Returns:

            配置值

        """

        try:

            return self._get_dotted_key(key)

        except KeyError:

            return default

    

    def _get_dotted_key(self, dotted_key: str) -> Any:

        """获取点分隔的键值"""

        parts = dotted_key.split('.')

www.kw.zizhitong.cn>www.dw.zizhitong.cn

        

        current = self._config_store

        for part in parts:

            if not isinstance(current, dict) or part not in current:

                raise KeyError(f"配置键不存在: {dotted_key}")

            current = current[part]

        

        return current

    

    def set(self, key: str, value: Any, validate: bool = True):

        """

        设置配置值

        

        Args:

            key: 配置键

            value: 配置值

            validate: 是否验证

        

        Raises:

            ValueError: 验证失败

        """

        if validate:

            self._validate_single_key(key, value)

        

        self._set_dotted_key(key, value)

    

    def _validate_config(self, config: Dict):

        """验证配置字典"""

        errors = []

        

        for rule_key, rule in self.VALIDATION_RULES.items():

            # 获取配置值

            try:

                value = self._get_dotted_key_from_dict(config, rule_key)

            except KeyError:

                if rule.get("required", False):

                    errors.append(f"缺少必需配置: {rule_key}")

                continue

            

            # 验证类型

            expected_type = rule.get("type")

            if expected_type and not isinstance(value, expected_type):

                errors.append(f"配置 {rule_key} 类型错误: 期望 {expected_type}, 实际 {type(value)}")

            

            # 验证范围

            if "min" in rule and value < rule["min"]:

                errors.append(f"配置 {rule_key} 值太小: 最小 {rule['min']}, 实际 {value}")

            

            if "max" in rule and value > rule["max"]:

                errors.append(f"配置 {rule_key} 值太大: 最大 {rule['max']}, 实际 {value}")

            

            # 验证选项

            if "choices" in rule and value not in rule["choices"]:

                errors.append(f"配置 {rule_key} 值无效: 允许 {rule['choices']}, 实际 {value}")

        

        if errors:

            raise ValueError(f"配置验证失败:\n" + "\n".join(errors))

    

    def _validate_single_key(self, key: str, value: Any):

        """验证单个配置项"""

        # 检查是否有验证规则

        for rule_key, rule in self.VALIDATION_RULES.items():

            if rule_key == key or rule_key.endswith(f".{key}"):

                expected_type = rule.get("type")

                if expected_type and not isinstance(value, expected_type):

                    raise ValueError(f"配置 {key} 类型错误: 期望 {expected_type}, 实际 {type(value)}")

                

                if "min" in rule and value < rule["min"]:

                    raise ValueError(f"配置 {key} 值太小: 最小 {rule['min']}, 实际 {value}")

                

                if "max" in rule and value > rule["max"]:

                    raise ValueError(f"配置 {key} 值太大: 最大 {rule['max']}, 实际 {value}")

                

                if "choices" in rule and value not in rule["choices"]:

                    raise ValueError(f"配置 {key} 值无效: 允许 {rule['choices']}, 实际 {value}")

    

    def _get_dotted_key_from_dict(self, data: Dict, dotted_key: str) -> Any:

        """从字典中获取点分隔的键值"""

        parts = dotted_key.split('.')

        

        current = data

        for part in parts:

            if not isinstance(current, dict) or part not in current:

                raise KeyError(f"键不存在: {dotted_key}")

            current = current[part]

        

        return current

    

    def to_dict(self) -> Dict:

        """导出配置为字典"""

        import copy

        return copy.deepcopy(self._config_store)

    

    def save_to_file(self, filepath: Union[str, Path], format: str = 'yaml'):

        """

        保存配置到文件

        

        Args:

            filepath: 文件路径

            format: 格式 ('yaml' 或 'json')

        """

        filepath = Path(filepath)

        

        with open(filepath, 'w', encoding='utf-8') as f:

            if format.lower() == 'yaml':

                yaml.dump(self.to_dict(), f, default_flow_style=False, allow_unicode=True)

            elif format.lower() == 'json':

                json.dump(self.to_dict(), f, indent=2, ensure_ascii=False)

            else:

                raise ValueError(f"不支持的格式: {format}")

    

    def watch_for_changes(self, filepath: Union[str, Path], interval: int = 60):

        """

        监视配置文件变化（示例）

        

        Args:

            filepath: 配置文件路径

            interval: 检查间隔（秒）

        """

        import threading

        import time

        

        def watcher():

            last_mtime = 0

            while True:

                try:

                    current_mtime = os.path.getmtime(filepath)

                    if current_mtime > last_mtime:

                        print(f"检测到配置文件变化，重新加载...")

                        self.load_from_file(filepath)

                        last_mtime = current_mtime

                except Exception as e:

                    print(f"监视配置文件时出错: {e}")

                

                time.sleep(interval)

        

        thread = threading.Thread(target=watcher, daemon=True)

        thread.start()

        return thread

# 使用示例

def demo_configuration_manager():

    """演示配置管理器的使用"""

    

    config = ConfigurationManager()

    

    # 从文件加载配置

    try:

        config.load_from_file("config.yaml")

    except FileNotFoundError:

        # 创建示例配置

        example_config = {

            "app": {

                "name": "示例应用",

                "version": "2.0.0"

            },

            "database": {

                "url": "postgresql://localhost/mydb",

                "pool_size": 20

            },

            "redis": {

                "host": "redis.local",

                "port": 6379

            }

        }

        

        import yaml

        with open("config.yaml", "w", encoding='utf-8') as f:

            yaml.dump(example_config, f)

        

        config.load_from_file("config.yaml")

    

    # 从环境变量加载

    os.environ["APP_API_TIMEOUT"] = "45"

    os.environ["APP_LOGGING_LEVEL"] = "DEBUG"

    config.load_from_env()

    

    # 获取配置

    print(f"应用名称: {config.get('app.name')}")

    print(f"数据库URL: {config.get('database.url')}")

    print(f"API超时: {config.get('api.timeout')}")

    print(f"日志级别: {config.get('logging.level')}")

    

    # 设置配置

    config.set("app.debug", True)

    print(f"调试模式: {config.get('app.debug')}")

    

    # 保存配置

    config.save_to_file("config_export.yaml")

    

    # 导出为字典

    config_dict = config.to_dict()

    print("\n完整配置:")

    print(json.dumps(config_dict, indent=2, ensure_ascii=False))

if __name__ == "__main__":

    demo_configuration_manager()

```

### 服务注册与发现

```python

# service_registry.py

"""基于全局字典的服务注册表"""

import inspect

import threading

from typing import Dict, Any, Callable, Optional, List

from dataclasses import dataclass, field

from datetime import datetime

from enum import Enum

class ServiceStatus(Enum):

    """服务状态"""

    REGISTERED = "registered"

    HEALTHY = "healthy"

    UNHEALTHY = "unhealthy"

    OFFLINE = "offline"

@dataclass

class ServiceInfo:

    """服务信息"""

    name: str

    instance: Any

    version: str = "1.0.0"

    status: ServiceStatus = ServiceStatus.REGISTERED

    metadata: Dict[str, Any] = field(default_factory=dict)

    registered_at: datetime = field(default_factory=datetime.now)

    last_health_check: Optional[datetime] = None

    

    def to_dict(self) -> Dict[str, Any]:

        """转换为字典"""

        return {

            "name": self.name,

            "version": self.version,

            "status": self.status.value,

            "metadata": self.metadata,

            "registered_at": self.registered_at.isoformat(),

            "last_health_check": self.last_health_check.isoformat() 

                if self.last_health_check else None

        }

class ServiceRegistry:

    """服务注册表"""

    

    _instance = None

    _services: Dict[str, ServiceInfo] = {}

    _service_locks: Dict[str, threading.RLock] = {}

    _registry_lock = threading.RLock()

    

    def __new__(cls):

        if cls._instance is None:

            with threading.Lock():

                if cls._instance is None:

                    cls._instance = super().__new__(cls)

        return cls._instance

    

    def register(self, name: str, instance: Any, 

                version: str = "1.0.0",

                metadata: Optional[Dict] = None) -> bool:

        """

        注册服务

        

        Args:

            name: 服务名称

            instance: 服务实例

            version: 服务版本

            metadata: 元数据

        

        Returns:

            是否注册成功

        """

        with self._registry_lock:

            if name in self._services:

                return False

            

            service_info = ServiceInfo(

                name=name,

                instance=instance,

                version=version,

                metadata=metadata or {},

                status=ServiceStatus.REGISTERED

            )

            

            self._services[name] = service_info

            self._service_locks[name] = threading.RLock()

            

            return True

    

    def unregister(self, name: str) -> bool:

        """

        注销服务

        

        Args:

            name: 服务名称

        

        Returns:

            是否注销成功

        """

        with self._registry_lock:

            if name not in self._services:

                return False

            

            del self._services[name]

            del self._service_locks[name]

            

            return True

    

    def get_service(self, name: str) -> Optional[Any]:

        """

        获取服务实例

        

        Args:

            name: 服务名称

        

        Returns:

            服务实例或None

        """

        with self._registry_lock:

            if name not in self._services:

                return None

            

            service_info = self._services[name]

            

            # 检查服务健康状态

            if service_info.status == ServiceStatus.UNHEALTHY:

                # 可以在这里实现故障转移或重试逻辑

                pass

            

            return service_info.instance

    

    def get_service_info(self, name: str) -> Optional[ServiceInfo]:

        """获取服务信息"""

        with self._registry_lock:

            return self._services.get(name)

    

    def list_services(self, status: Optional[ServiceStatus] = None) -> List[str]:

        """

        列出服务

        

        Args:

            status: 过滤状态

        

        Returns:

            服务名称列表

        """

        with self._registry_lock:

            if status is None:

                return list(self._services.keys())

            

            return [

                name for name, info in self._services.items()

                if info.status == status

            ]

    

    def update_status(self, name: str, status: ServiceStatus) -> bool:

        """

        更新服务状态

        

        Args:

            name: 服务名称

            status: 新状态

        

        Returns:

            是否更新成功

        """

        with self._registry_lock:

            if name not in self._services:

                return False

            

            service_info = self._services[name]

            service_info.status = status

            service_info.last_health_check = datetime.now()

            

            return True

    

    def health_check_all