【引言】

凌晨3点，你盯着屏幕上那堆继承链深不见底的代码，陷入了深深的自我怀疑。class UserAuthHandler extends BaseAuthHandler extends AbstractHandler... ——这种继承地狱是不是你每天都要面对的现实？

Hardcode 是最爽的，if-else 写起来也是最快的。但当你需要给第三方API加一层缓存、给数据库操作加上重试逻辑、或者把一个Dict变成对象时，那些所谓的"快速实现"就会变成维护者的噩梦。

结构型模式（Structural Patterns）就是来解这个围的。它们不改变类和对象的结构，但能让你灵活地组合它们。简单说：让你的代码开箱即用，而不是每次都从零造轮子。

今天聊聊Python里最实用的5种结构型模式：Adapter、Decorator、Facade、Proxy、Bridge。

【核心原理】

1. Adapter Pattern（适配器模式）

类比：就像你去国外出差，发现插座不兼容。买个转换器（Adapter），你的充电器就能用了。

本质：把一个类的接口转换成客户端期望的接口，让原本接口不兼容的类可以协同工作。

适用场景：

• 集成第三方库，接口不匹配
• 重构老代码，逐步替换实现
• 统一多个相似但接口不同的类

2. Decorator Pattern（装饰器模式）

类比：给咖啡加奶、加糖、加巧克力。咖啡还是咖啡，只是功能增强了。

本质：动态地给对象添加额外职责，比继承更灵活。

适用场景：

• 日志记录
• 性能监控
• 缓存
• 权限校验

3. Facade Pattern（外观模式）

类比：公司前台，你不用知道每个部门在哪，找前台就行了。

本质：为复杂的子系统提供简单接口，隐藏内部复杂性。

适用场景：

• 简化复杂API的调用
• 解耦客户端和子系统
• 构建分层架构

4. Proxy Pattern（代理模式）

类比：VPN代理，你访问的是代理，代理再转发请求。

本质：为其他对象提供代理，以控制对这个对象的访问。

适用场景：

• 延迟初始化
• 访问控制
• 远程代理
• 智能引用（引用计数）

5. Bridge Pattern（桥接模式）

类比：遥控器和电视。遥控器不需要知道电视型号，电视也不需要知道遥控器。

本质：将抽象与实现分离，让它们可以独立变化。

适用场景：

• 多维度的类继承问题
• 需要运行时切换实现
• 避免类爆炸

【核心代码】

1. Adapter Pattern（适配器模式）

from abc import ABC, abstractmethodfrom typing importDict, Any# 第三方支付接口（无法修改）class ThirdPartyPaymentSDK:def process_payment(self, amount: float, currency: str, card_info: Dict[str, Any]) -> Dict[str, Any]:"""第三方支付接口，参数格式不匹配我们的系统"""return {'success': True,'transaction_id': 'TP_123456','amount': amount,'currency': currency        }# 我们系统的支付接口标准class PaymentProcessor(ABC):    @abstractmethoddef pay(self, amount: float, payment_method: str) -> Dict[str, Any]:pass# 适配器：把第三方SDK接口转换成我们的接口class PaymentAdapter(PaymentProcessor):def __init__(self, third_party_sdk: ThirdPartyPaymentSDK):self._sdk = third_party_sdkdef pay(self, amount: float, payment_method: str) -> Dict[str, Any]:"""转换接口调用"""# 解析payment_method获取卡号信息（简化示例）        card_info = self._parse_payment_method(payment_method)# 调用第三方SDK        result = self._sdk.process_payment(            amount=amount,            currency='USD',            card_info=card_info        )# 转换返回格式return {'status': 'success'if result['success'] else'failed','transaction_id': result['transaction_id'],'message': 'Payment processed successfully'        }def _parse_payment_method(self, payment_method: str) -> Dict[str, Any]:"""解析支付方式为卡号信息（实际项目中会更复杂）"""# 示例：payment_method格式为 "card:4242..."if payment_method.startswith('card:'):            card_number = payment_method[5:]return {'card_number': card_number,'expiry': '12/25','cvv': '123'            }raise ValueError(f"Unsupported payment method: {payment_method}")# 使用示例def main():    third_party_sdk = ThirdPartyPaymentSDK()    adapter = PaymentAdapter(third_party_sdk)    result = adapter.pay(amount=100.0, payment_method='card:4242424242424242')print(result)  # {'status': 'success', 'transaction_id': 'TP_123456', 'message': 'Payment processed successfully'}if __name__ == '__main__':    main()

2. Decorator Pattern（装饰器模式）

import timefrom functools import wrapsfrom typing importCallable, Any# 基础装饰器：记录执行时间def timing_decorator(func: Callable) -> Callable:    @wraps(func)def wrapper(*args, **kwargs) -> Any:        start_time = time.time()        result = func(*args, **kwargs)        end_time = time.time()        execution_time = end_time - start_timeprint(f"[TIMING] {func.__name__} executed in {execution_time:.4f} seconds")return resultreturn wrapper# 带参数的装饰器：重试逻辑def retry_decorator(max_attempts: int = 3, delay: float = 1.0) -> Callable:def decorator(func: Callable) -> Callable:        @wraps(func)def wrapper(*args, **kwargs) -> Any:            last_exception = Nonefor attempt inrange(1, max_attempts + 1):try:return func(*args, **kwargs)except Exception as e:                    last_exception = eprint(f"[RETRY] {func.__name__} failed (attempt {attempt}/{max_attempts}): {e}")if attempt < max_attempts:                        time.sleep(delay)raise last_exceptionreturn wrapperreturn decorator# 带缓存的装饰器def memoize_decorator(func: Callable) -> Callable:    cache = {}    @wraps(func)def wrapper(*args, **kwargs) -> Any:# 创建缓存键（简化版，实际项目中要处理不可哈希参数）        cache_key = (args, tuple(sorted(kwargs.items())))if cache_key notin cache:print(f"[CACHE] Computing result for {func.__name__}")            cache[cache_key] = func(*args, **kwargs)else:print(f"[CACHE] Returning cached result for {func.__name__}")return cache[cache_key]return wrapper# 组合使用装饰器@timing_decorator@retry_decorator(max_attempts=3, delay=0.5)@memoize_decoratordef expensive_computation(n: int) -> int:"""模拟耗时计算"""print(f"Computing fibonacci({n})...")    time.sleep(0.5)  # 模拟耗时if n <= 1:return nreturn expensive_computation(n - 1) + expensive_computation(n - 2)# 使用示例def main():# 第一次调用：计算并缓存print("First call:")    result1 = expensive_computation(5)print(f"Result: {result1}\n")# 第二次调用：从缓存返回print("Second call:")    result2 = expensive_computation(5)print(f"Result: {result2}\n")# 第三次调用：重试演示（模拟失败）print("Third call (simulated failure):")try:        @retry_decorator(max_attempts=2, delay=0.1)def failing_function():raise ValueError("Intentional failure")        failing_function()except ValueError as e:print(f"Failed after retries: {e}")if __name__ == '__main__':    main()

3. Facade Pattern（外观模式）

from typing importList, Dict, Any, Optionalimport json# 子系统1：文件操作class FileManager:def read_file(self, file_path: str) -> Optional[str]:"""读取文件"""try:with open(file_path, 'r', encoding='utf-8') as f:return f.read()except FileNotFoundError:print(f"[FileManager] File not found: {file_path}")return Noneexcept Exception as e:print(f"[FileManager] Error reading file: {e}")return Nonedef write_file(self, file_path: str, content: str) -> bool:"""写入文件"""try:with open(file_path, 'w', encoding='utf-8') as f:                f.write(content)return Trueexcept Exception as e:print(f"[FileManager] Error writing file: {e}")return False# 子系统2：数据解析class DataParser:def parse_json(self, json_str: str) -> Optional[Dict[str, Any]]:"""解析JSON字符串"""try:return json.loads(json_str)except json.JSONDecodeError as e:print(f"[DataParser] Invalid JSON: {e}")return Nonedef validate_data(self, data: Dict[str, Any], required_fields: List[str]) -> bool:"""验证数据完整性"""for field in required_fields:if field notin data:print(f"[DataParser] Missing required field: {field}")return Falsereturn True# 子系统3：数据转换class DataTransformer:def transform_to_uppercase(self, data: Dict[str, Any]) -> Dict[str, Any]:"""将字符串字段转换为大写"""        result = {}for key, value in data.items():if isinstance(value, str):                result[key] = value.upper()else:                result[key] = valuereturn resultdef filter_data(self, data: Dict[str, Any], excluded_keys: List[str]) -> Dict[str, Any]:"""过滤掉指定的字段"""return {k: v for k, v in data.items() if k notin excluded_keys}# 外观类：简化复杂操作class DataProcessingFacade:"""数据处理外观类，隐藏内部复杂性"""def __init__(self):self._file_manager = FileManager()self._data_parser = DataParser()self._data_transformer = DataTransformer()def process_config_file(        self,        file_path: str,        required_fields: List[str],        to_uppercase: bool = False,        exclude_fields: Optional[List[str]] = None) -> Optional[Dict[str, Any]]:"""        一键处理配置文件：读取、解析、验证、转换        参数:            file_path: 文件路径            required_fields: 必需字段列表            to_uppercase: 是否转换为大写            exclude_fields: 要排除的字段列表        返回:            处理后的数据字典，失败返回None        """# Step 1: 读取文件print("[Facade] Step 1: Reading file...")        file_content = self._file_manager.read_file(file_path)if file_content is None:return None# Step 2: 解析JSONprint("[Facade] Step 2: Parsing JSON...")        parsed_data = self._data_parser.parse_json(file_content)if parsed_data is None:return None# Step 3: 验证数据print("[Facade] Step 3: Validating data...")if not self._data_parser.validate_data(parsed_data, required_fields):return None# Step 4: 转换数据（可选）if to_uppercase:print("[Facade] Step 4: Transforming to uppercase...")            parsed_data = self._data_transformer.transform_to_uppercase(parsed_data)if exclude_fields:print(f"[Facade] Step 5: Excluding fields: {exclude_fields}")            parsed_data = self._data_transformer.filter_data(parsed_data, exclude_fields)print("[Facade] Processing completed successfully!")return parsed_data# 使用示例def main():# 创建测试数据    test_data = {"name": "John Doe","email": "john@example.com","age": 30,"role": "developer"    }# 写入测试文件    test_file = "config.json"with open(test_file, 'w', encoding='utf-8') as f:        json.dump(test_data, f)# 使用外观类处理文件    facade = DataProcessingFacade()print("=" * 50)print("Scenario 1: Basic processing")print("=" * 50)    result = facade.process_config_file(        file_path=test_file,        required_fields=["name", "email"]    )print(f"Result: {result}\n")print("=" * 50)print("Scenario 2: Processing with transformation")print("=" * 50)    result = facade.process_config_file(        file_path=test_file,        required_fields=["name", "email"],        to_uppercase=True,        exclude_fields=["age"]    )print(f"Result: {result}\n")if __name__ == '__main__':    main()

4. Proxy Pattern（代理模式）

from typing importOptional, Dict, Anyimport time# 真实主题：数据库连接class DatabaseConnection:def __init__(self, connection_string: str):self._connection_string = connection_stringself._is_connected = Falseprint(f"[Database] Initializing connection to {connection_string}")def connect(self) -> bool:"""建立数据库连接"""print(f"[Database] Connecting to database...")        time.sleep(0.5)  # 模拟连接耗时self._is_connected = Trueprint("[Database] Connected successfully!")return Truedef query(self, sql: str) -> Optional[Dict[str, Any]]:"""执行查询"""if not self._is_connected:raise RuntimeError("Not connected to database!")print(f"[Database] Executing query: {sql}")# 模拟查询结果return {'id': 1, 'name': 'Sample Data', 'result': 'success'}def disconnect(self) -> None:"""断开连接"""print("[Database] Disconnecting...")self._is_connected = False# 代理：延迟初始化 + 缓存 + 访问控制class DatabaseProxy:def __init__(self, connection_string: str, max_cache_size: int = 10):self._connection_string = connection_stringself._real_db: Optional[DatabaseConnection] = None# 延迟初始化self._query_cache: Dict[str, Dict[str, Any]] = {}  # 查询缓存self._max_cache_size = max_cache_sizeself._access_count = 0# 访问计数def connect(self) -> bool:"""延迟初始化：只在需要时才创建真实连接"""if self._real_db is None:print("[Proxy] Initializing lazy database connection...")self._real_db = DatabaseConnection(self._connection_string)return self._real_db.connect()def query(self, sql: str, use_cache: bool = True) -> Optional[Dict[str, Any]]:"""        智能代理：缓存 + 访问控制 + 延迟初始化        参数:            sql: SQL查询语句            use_cache: 是否使用缓存        返回:            查询结果        """self._access_count += 1print(f"[Proxy] Query #{self._access_count}")# 检查缓存if use_cache and sql in self._query_cache:print(f"[Proxy] Cache hit for query: {sql}")return self._query_cache[sql]print("[Proxy] Cache miss, executing query...")# 延迟初始化if self._real_db is None:print("[Proxy] Auto-connecting (lazy initialization)")self.connect()# 执行查询        result = self._real_db.query(sql)# 缓存结果if use_cache and result is not None:if len(self._query_cache) >= self._max_cache_size:# 缓存满了，删除最早的（简单LRU实现）                oldest_key = next(iter(self._query_cache))print(f"[Proxy] Cache full, evicting: {oldest_key}")del self._query_cache[oldest_key]self._query_cache[sql] = resultprint(f"[Proxy] Cached result for query: {sql}")return resultdef disconnect(self) -> None:"""断开连接并清理缓存"""if self._real_db is not None:self._real_db.disconnect()self._query_cache.clear()print("[Proxy] Cache cleared")def get_cache_stats(self) -> Dict[str, Any]:"""获取缓存统计信息"""return {'cache_size': len(self._query_cache),'max_cache_size': self._max_cache_size,'access_count': self._access_count,'cached_queries': list(self._query_cache.keys())        }# 使用示例def main():# 创建代理    proxy = DatabaseProxy("postgresql://localhost/mydb", max_cache_size=3)print("=" * 60)print("Test 1: First query (cache miss)")print("=" * 60)    result1 = proxy.query("SELECT * FROM users WHERE id = 1")print(f"Result: {result1}\n")print("=" * 60)print("Test 2: Second same query (cache hit)")print("=" * 60)    result2 = proxy.query("SELECT * FROM users WHERE id = 1")print(f"Result: {result2}\n")print("=" * 60)print("Test 3: Different query (cache miss)")print("=" * 60)    result3 = proxy.query("SELECT * FROM users WHERE id = 2")print(f"Result: {result3}\n")print("=" * 60)print("Test 4: Cache statistics")print("=" * 60)    stats = proxy.get_cache_stats()print(f"Cache stats: {stats}\n")# 测试缓存满的情况print("=" * 60)print("Test 5: Testing cache eviction")print("=" * 60)    proxy.query("SELECT * FROM products WHERE id = 1")  # 第3个查询    proxy.query("SELECT * FROM orders WHERE id = 1")     # 第4个查询（应该触发缓存清理）    stats = proxy.get_cache_stats()print(f"Cache stats after eviction: {stats}\n")# 清理    proxy.disconnect()if __name__ == '__main__':    main()

5. Bridge Pattern（桥接模式）

from abc import ABC, abstractmethodfrom typing importList, Optional# 实现层抽象：消息发送方式class MessageSender(ABC):"""消息发送者抽象接口"""    @abstractmethoddef send(self, message: str, recipient: str) -> bool:"""发送消息"""pass# 具体实现：短信发送class SMSMessageSender(MessageSender):def send(self, message: str, recipient: str) -> bool:print(f"[SMS] Sending SMS to {recipient}")print(f"[SMS] Message: {message[:50]}...")# 实际项目中会调用短信APIreturn True# 具体实现：邮件发送class EmailMessageSender(MessageSender):def send(self, message: str, recipient: str) -> bool:print(f"[Email] Sending Email to {recipient}")print(f"[Email] Subject: Notification")print(f"[Email] Body: {message[:50]}...")# 实际项目中会调用邮件APIreturn True# 具体实现：推送通知发送class PushNotificationSender(MessageSender):def send(self, message: str, recipient: str) -> bool:print(f"[Push] Sending Push Notification to {recipient}")print(f"[Push] Message: {message[:50]}...")# 实际项目中会调用推送服务APIreturn True# 抽象层：消息通知class Notification(ABC):"""通知抽象类"""def __init__(self, sender: MessageSender):self._sender = sender  # 桥接：持有发送者引用def send(self, recipient: str) -> bool:"""发送通知"""pass# 具体抽象：普通通知class NormalNotification(Notification):def __init__(self, sender: MessageSender, message: str):super().__init__(sender)self._message = messagedef send(self, recipient: str) -> bool:print(f"[Normal] Sending normal notification...")return self._sender.send(self._message, recipient)# 具体抽象：紧急通知class UrgentNotification(Notification):def __init__(self, sender: MessageSender, message: str):super().__init__(sender)self._message = f"[URGENT] {message}"def send(self, recipient: str) -> bool:print(f"[Urgent] Sending urgent notification (high priority)...")return self._sender.send(self._message, recipient)# 具体抽象：带附件的通知class AttachmentNotification(Notification):def __init__(self, sender: MessageSender, message: str, attachment: str):super().__init__(sender)self._message = messageself._attachment = attachmentdef send(self, recipient: str) -> bool:print(f"[Attachment] Sending notification with attachment: {self._attachment}")        full_message = f"{self._message}\n\n[Attachment: {self._attachment}]"return self._sender.send(full_message, recipient)# 扩展：通知管理器（组合多个通知方式）class NotificationManager:"""通知管理器：可以组合多种通知方式"""def __init__(self):self._notifications: List[Notification] = []def add_notification(self, notification: Notification) -> None:"""添加通知"""self._notifications.append(notification)def send_all(self, recipient: str) -> Dict[str, bool]:"""发送所有通知"""        results = {}for idx, notification in enumerate(self._notifications):print(f"\n{'='*60}")print(f"Sending notification {idx + 1}/{len(self._notifications)}")print('='*60)            results[f"notification_{idx+1}"] = notification.send(recipient)return results# 使用示例def main():# 创建不同的发送者    sms_sender = SMSMessageSender()    email_sender = EmailMessageSender()    push_sender = PushNotificationSender()print("="*60)print("Scenario 1: Bridge Pattern - Normal Notifications")print("="*60)# 使用桥接模式：不同的通知类型可以搭配不同的发送方式# 组合1：普通通知 + 短信    normal_sms = NormalNotification(sms_sender, "Your order has been shipped")    normal_sms.send("+8613800138000")print("\n" + "="*60)# 组合2：普通通知 + 邮件    normal_email = NormalNotification(email_sender, "Your order has been shipped")    normal_email.send("user@example.com")print("\n" + "="*60)print("Scenario 2: Bridge Pattern - Urgent Notifications")print("="*60)# 组合3：紧急通知 + 推送    urgent_push = UrgentNotification(push_sender, "Server alert: CPU usage 95%")    urgent_push.send("admin_user")print("\n" + "="*60)# 组合4：紧急通知 + 邮件 + 短信（多渠道）    urgent_email = UrgentNotification(email_sender, "Server alert: CPU usage 95%")    urgent_email.send("admin@example.com")    urgent_sms = UrgentNotification(sms_sender, "Server alert: CPU usage 95%")    urgent_sms.send("+8613900139000")print("\n" + "="*60)print("Scenario 3: Bridge Pattern - Attachment Notification")print("="*60)# 组合5：带附件通知 + 邮件    attachment_email = AttachmentNotification(        email_sender,"Monthly report ready","report_january_2026.pdf"    )    attachment_email.send("manager@example.com")print("\n" + "="*60)print("Scenario 4: Notification Manager - Multi-channel")print("="*60)# 使用通知管理器发送多渠道通知    manager = NotificationManager()    manager.add_notification(NormalNotification(sms_sender, "Welcome to our service"))    manager.add_notification(NormalNotification(email_sender, "Welcome to our service"))    manager.add_notification(UrgentNotification(push_sender, "Welcome to our service"))    results = manager.send_all("new_user")print(f"\nAll results: {results}")if __name__ == '__main__':    main()

【避坑指南】

1. Adapter Pattern：注意双向适配

坑点：只做单向适配，导致系统不灵活。

# ❌ 错误示例：只能把第三方适配成我们的接口class BadAdapter(PaymentProcessor):# 只能把ThirdPartyPaymentSDK适配成PaymentProcessorpass# ✅ 正确示例：支持双向适配class TwoWayAdapter:def __init__(self, obj: Any):self._obj = objdef adapt_to_our_interface(self) -> PaymentProcessor:"""转换成我们的接口"""# 实现转换逻辑passdef adapt_to_third_party_interface(self) -> ThirdPartyPaymentSDK:"""转换成第三方接口"""# 实现转换逻辑pass

性能问题：过度使用Adapter会导致调用链过长，影响性能。建议在系统边界使用，内部尽量保持接口一致。

2. Decorator Pattern：装饰器顺序陷阱

坑点：装饰器顺序错误导致逻辑混乱。

# ❌ 错误示例：装饰器顺序问题@retry_decorator(max_attempts=3)@memoize_decorator  # 缓存会在重试前生效，导致错误结果被缓存defunstable_api_call():# 可能失败的API调用pass# ✅ 正确示例：装饰器顺序正确@memoize_decorator  # 应该在最外层，只缓存成功的调用@retry_decorator(max_attempts=3)  # 重试失败后才缓存defstable_api_call():# 可能失败的API调用pass

性能问题：多层装饰器叠加会导致函数调用链过长，建议不超过3层装饰器。

3. Proxy Pattern：缓存一致性

坑点：代理缓存与真实数据不一致。

# ❌ 错误示例：缓存没有失效机制class BadProxy:def __init__(self):self._cache = {}def query(self, sql: str):if sql in self._cache:return self._cache[sql]  # 永远不更新缓存        result = self._real_db.query(sql)self._cache[sql] = resultreturn result# ✅ 正确示例：带TTL的缓存class GoodProxy:def __init__(self, ttl: int = 60):self._cache = {}self._cache_timestamps = {}self._ttl = ttldef query(self, sql: str):        now = time.time()# 检查缓存是否过期if sql in self._cache and (now - self._cache_timestamps[sql]) < self._ttl:return self._cache[sql]# 查询并更新缓存        result = self._real_db.query(sql)self._cache[sql] = resultself._cache_timestamps[sql] = nowreturn result

性能问题：Proxy如果实现不当（比如每个方法都创建新代理），会导致对象创建开销过大。

4. Facade Pattern：过度封装

坑点：把所有操作都塞进Facade，导致Facade变成上帝类（God Class）。

# ❌ 错误示例：Facade职责过多class GodFacade:def __init__(self):self._db = Database()self._cache = Cache()self._logger = Logger()self._email = Email()self._sms = SMS()self._push = Push()# ...更多依赖def operation1(self): passdef operation2(self): pass# ...50+个方法# ✅ 正确示例：职责分离的Facadeclass DatabaseFacade:"""只负责数据库相关操作"""def __init__(self, db, cache, logger):self._db = dbself._cache = cacheself._logger = loggerdef query_with_cache(self, sql): passclass NotificationFacade:"""只负责通知相关操作"""def __init__(self, email, sms, push, logger):self._email = emailself._sms = smsself._push = pushself._logger = loggerdef send_notification(self, message, channels): pass

性能问题：Facade如果只是简单转发调用，没有实际封装价值，反而增加了间接层。应该在Facade层做有意义的组合操作。

5. Bridge Pattern：过度抽象

坑点：为了用Bridge而用Bridge，导致代码复杂度爆炸。

# ❌ 错误示例：过度使用Bridgeclass AbstractAbstractAbstraction(ABC):"""三层抽象，完全没必要"""passclass ConcreteAbstraction(AbstractAbstraction):pass# ✅ 正确示例：只在真正需要时使用Bridgeclass PaymentMethod(ABC):"""支付方式抽象"""passclass CreditCardPayment(PaymentMethod):"""信用卡支付"""pass

性能问题：Bridge模式的间接调用会轻微影响性能，但在大多数场景下可以忽略。如果对性能极度敏感，需要评估是否值得。

【总结】

代码写得烂，不是因为你菜，而是因为你懒得思考。 结构型模式不是银弹，但它们能让你的代码从"能跑"进化到"优雅"。别等到接盘侠来骂你时，才想起重构。

Stay curious, keep coding.