Python设计模式工程实践中的模式应用1. 技术分析1.1 设计模式分类类型模式复杂度适用场景创建型单例、工厂、建造者低~中对象创建结构型适配器、装饰器、代理低~中接口适配行为型观察者、策略、命令中算法封装1.2 Python实现特点Python的动态特性和内置机制使得某些传统模式可以更简洁地实现单例模式 → 模块级实现装饰器模式 → decorator语法上下文管理器 → with语句2. 核心功能实现2.1 单例模式# 方法1: 模块级单例推荐 # database.py class Database: def __init__(self): self.connection None def connect(self): if self.connection is None: print(建立数据库连接) self.connection conn_123 return self.connection # 全局实例 _db_instance Database() def get_database(): return _db_instance # 方法2: 元类单例 class SingletonMeta(type): _instances {} def __call__(cls, *args, **kwargs): if cls not in cls._instances: cls._instances[cls] super().__call__(*args, **kwargs) return cls._instances[cls] class DatabaseMeta(metaclassSingletonMeta): def __init__(self): self.connection None def connect(self): if self.connection is None: print(建立数据库连接) self.connection conn_456 return self.connection # 方法3: 装饰器单例 from functools import wraps def singleton(cls): instances {} wraps(cls) def get_instance(*args, **kwargs): if cls not in instances: instances[cls] cls(*args, **kwargs) return instances[cls] return get_instance singleton class CacheService: def __init__(self): self.cache {} def get(self, key): return self.cache.get(key) def set(self, key, value): self.cache[key] value2.2 工厂模式from abc import ABC, abstractmethod # 抽象产品 class Model(ABC): abstractmethod def predict(self, data): pass # 具体产品 class LinearRegressionModel(Model): def __init__(self, lr0.01): self.lr lr self.weights None def predict(self, data): return sum(w * x for w, x in zip(self.weights, data)) def fit(self, X, y): # 简化实现 self.weights [0.5] * len(X[0]) class DecisionTreeModel(Model): def __init__(self, max_depth10): self.max_depth max_depth self.tree {} def predict(self, data): return 0.5 # 简化实现 # 工厂 class ModelFactory: _registry {} classmethod def register(cls, model_type, model_class): cls._registry[model_type] model_class classmethod def create(cls, model_type, **kwargs): if model_type not in cls._registry: raise ValueError(fUnknown model type: {model_type}) return cls._registry[model_type](**kwargs) # 注册模型 ModelFactory.register(linear, LinearRegressionModel) ModelFactory.register(tree, DecisionTreeModel) # 使用 model ModelFactory.create(linear, lr0.001) result model.predict([1.0, 2.0, 3.0])2.3 策略模式from abc import ABC, abstractmethod class OptimizationStrategy(ABC): abstractmethod def optimize(self, data): pass class GradientDescent(OptimizationStrategy): def __init__(self, lr0.01): self.lr lr def optimize(self, data): return GD优化, lr str(self.lr) class AdamOptimizer(OptimizationStrategy): def __init__(self, lr0.001, beta10.9, beta20.999): self.lr lr self.beta1 beta1 self.beta2 beta2 def optimize(self, data): return fAdam优化, lr{self.lr} class SGDOptimizer(OptimizationStrategy): def __init__(self, lr0.01, momentum0.9): self.lr lr self.momentum momentum def optimize(self, data): return fSGD优化, lr{self.lr}, momentum{self.momentum} class Trainer: def __init__(self, strategy: OptimizationStrategy): self.strategy strategy def set_strategy(self, strategy: OptimizationStrategy): self.strategy strategy def train(self, data): return self.strategy.optimize(data) # 使用 trainer Trainer(GradientDescent(lr0.01)) print(trainer.train(data)) trainer.set_strategy(AdamOptimizer(lr0.001)) print(trainer.train(data))2.4 观察者模式from collections import defaultdict class EventObserver: def __init__(self): self._handlers defaultdict(list) def subscribe(self, event, handler): self._handlers[event].append(handler) return lambda: self.unsubscribe(event, handler) def unsubscribe(self, event, handler): if event in self._handlers: self._handlers[event].remove(handler) def emit(self, event, *args, **kwargs): for handler in self._handlers[event]: handler(*args, **kwargs) class TrainingMonitor: def __init__(self): self.observer EventObserver() self.history {loss: [], accuracy: []} def on_epoch_end(self, epoch, loss, accuracy): self.history[loss].append(loss) self.history[accuracy].append(accuracy) self.observer.emit(epoch_end, epochepoch, lossloss, accuracyaccuracy) # 检查早停 if len(self.history[loss]) 5: if all(self.history[loss][-i] self.history[loss][-i-1] for i in range(1, 5)): self.observer.emit(early_stop, epochepoch) # 使用 monitor TrainingMonitor() def save_model(epoch, **kwargs): print(f保存模型: epoch{epoch}) def send_notification(epoch, loss, **kwargs): print(f通知: Epoch {epoch}, Loss{loss:.4f}) monitor.observer.subscribe(epoch_end, save_model) monitor.observer.subscribe(epoch_end, send_notification) monitor.observer.subscribe(early_stop, lambda **k: print(早停触发)) # 模拟训练 for epoch in range(10): loss 1.0 - epoch * 0.1 accuracy 0.5 epoch * 0.05 monitor.on_epoch_end(epoch, loss, min(accuracy, 0.99))2.5 装饰器模式import time from functools import wraps def log_calls(func): 日志装饰器 wraps(func) def wrapper(*args, **kwargs): print(f调用 {func.__name__}) start time.perf_counter() result func(*args, **kwargs) elapsed time.perf_counter() - start print(f{func.__name__} 完成, 耗时: {elapsed:.4f}s) return result return wrapper def retry(max_attempts3, delay1.0): 重试装饰器 def decorator(func): wraps(func) def wrapper(*args, **kwargs): for attempt in range(max_attempts): try: return func(*args, **kwargs) except Exception as e: if attempt max_attempts - 1: time.sleep(delay) print(f重试 {attempt 1}/{max_attempts}) else: raise return wrapper return decorator def cache_result(func): 缓存装饰器 wraps(func) def wrapper(*args, **kwargs): cache_key str(args) str(kwargs) if not hasattr(wrapper, _cache): wrapper._cache {} if cache_key not in wrapper._cache: wrapper._cache[cache_key] func(*args, **kwargs) return wrapper._cache[cache_key] return wrapper log_calls retry(max_attempts3, delay0.5) cache_result def fetch_data(url): 带日志、重试和缓存的数据获取 print(f从 {url} 获取数据) time.sleep(0.5) return {status: success, data: [1, 2, 3]} # 链式调用 result fetch_data(http://example.com/api) print(result)2.6 上下文管理器class DatabaseConnection: 数据库连接上下文管理器 def __init__(self, dsn): self.dsn dsn self.connection None def __enter__(self): print(f连接数据库: {self.dsn}) self.connection fconn_to_{self.dsn} return self def __exit__(self, exc_type, exc_val, exc_tb): if exc_type: print(f数据库错误: {exc_val}) print(关闭数据库连接) self.connection None return False # 不抑制异常 class Transaction: 事务上下文管理器 def __init__(self, connection): self.connection connection self.committed False def __enter__(self): print(开始事务) return self def __exit__(self, exc_type, exc_val, exc_tb): if exc_type: print(f事务回滚: {exc_val}) self.rollback() return True # 抑制异常 else: self.commit() return False def commit(self): print(提交事务) self.committed True def rollback(self): print(回滚事务) # 使用 with DatabaseConnection(mydb) as db: with Transaction(db.connection): print(执行SQL操作...) # raise Exception(测试异常) # 组合使用 with DatabaseConnection(mydb) as db: with Transaction(db.connection) as txn: print(执行SQL操作...)3. 性能对比3.1 单例模式实现对比import timeit # 模块级 vs 元类 module_singleton class _Instance: def __init__(self): self.value None _instance _Instance() metaclass_singleton class SingletonMeta(type): _instances {} def __call__(cls, *args, **kwargs): if cls not in cls._instances: cls._instances[cls] super().__call__(*args, **kwargs) return cls._instances[cls] class Instance(metaclassSingletonMeta): def __init__(self): self.value None # 测试 module_time timeit.timeit(_instance, setupmodule_singleton, number100000) meta_time timeit.timeit(Instance(), setupmetaclass_singleton, number100000) print(f模块级: {module_time:.4f}s) print(f元类: {meta_time:.4f}s)4. 最佳实践4.1 模式选择指南场景推荐模式替代方案全局唯一对象模块级单例依赖注入对象创建复杂工厂模式Builder算法可切换策略模式函数参数事件处理观察者模式asyncio事件功能扩展装饰器模式mixin4.2 注意事项# ✅ 推荐简单场景用简单实现 # 单例 → 模块级 # 工厂 → 函数字典注册 # ❌ 避免过度使用设计模式 # 简单的if-else能解决的问题不需要工厂模式5. 总结设计模式应用要点适度使用不要为了模式而模式Python风格善用Python特性简化实现组合使用多个模式可以组合解决复杂问题