Python 适配器模式

适配器模式是一种结构型设计模式，它允许不兼容的接口之间进行协作。就像现实生活中的电源适配器可以让不同国家的电器插头连接到本地插座一样，在编程中，适配器模式充当两个不兼容接口之间的桥梁。

核心概念

适配器模式主要解决接口不匹配的问题。当我们需要使用一个现有的类，但其接口不符合我们的需求时，适配器模式就能派上用场。它通过创建一个中间层（适配器）来转换接口，使得原本不兼容的类能够协同工作。

模式结构

适配器模式包含三个主要角色：

  • 目标接口(Target)：客户端期望的接口
  • 适配器(Adapter)：实现目标接口，并包装适配者
  • 适配者(Adaptee)：需要被适配的现有类

为什么需要适配器模式

实际问题场景

想象一下，你有一个现有的日志系统，它使用特定的方法记录日志：

实例

class LegacyLogger:
    def write_log(self, message):
        print(f"[LEGACY] {message}")

但现在你希望使用一个新的日志接口：

实例

class NewLogger:
    def log(self, level, message):
        print(f"[{level.upper()}] {message}")

如果没有适配器，你就需要修改所有调用旧日志系统的代码，这既耗时又容易出错。

适配器的优势

  1. 代码复用：可以重用现有的类，无需修改源代码
  2. 解耦合：将接口转换逻辑与业务逻辑分离
  3. 灵活性：可以轻松切换不同的实现
  4. 符合开闭原则：对扩展开放，对修改关闭

适配器模式的实现方式

类适配器（使用继承）

类适配器通过多重继承来实现接口适配：

实例

class LegacyLogger:
    """被适配的旧日志类"""
    def write_log(self, message):
        print(f"[LEGACY] {message}")

class NewLoggerInterface:
    """目标接口"""
    def log(self, level, message):
        pass

class LoggerAdapter(NewLoggerInterface, LegacyLogger):
    """类适配器 - 通过继承实现"""
    def log(self, level, message):
        # 将新接口转换为旧接口
        formatted_message = f"{level}: {message}"
        self.write_log(formatted_message)

# 使用示例
logger = LoggerAdapter()
logger.log("INFO", "系统启动成功")
logger.log("ERROR", "数据库连接失败")

输出结果：

[LEGACY] INFO: 系统启动成功
[LEGACY] ERROR: 数据库连接失败

对象适配器（使用组合）

对象适配器通过组合方式实现，这是更常用的方式：

实例

class LegacyPayment:
    """被适配的旧支付系统"""
    def process_payment(self, amount_in_dollars):
        print(f"处理支付: ${amount_in_dollars}")
        return True

class NewPaymentInterface:
    """新的支付接口"""
    def pay(self, amount, currency="CNY"):
        pass

class PaymentAdapter(NewPaymentInterface):
    """对象适配器 - 通过组合实现"""
   
    def __init__(self, legacy_payment):
        self.legacy_payment = legacy_payment
   
    def pay(self, amount, currency="CNY"):
        # 货币转换逻辑
        if currency == "CNY":
            amount_in_dollars = amount / 7.0  # 简化汇率转换
        else:
            amount_in_dollars = amount
       
        # 调用旧系统的接口
        return self.legacy_payment.process_payment(amount_in_dollars)

# 使用示例
legacy_payment = LegacyPayment()
payment = PaymentAdapter(legacy_payment)

# 使用新接口调用旧系统
payment.pay(700, "CNY")  # 支付700元人民币
payment.pay(100, "USD")  # 支付100美元

输出结果：

处理支付: $100.0
处理支付: $100

实际应用案例

案例1：数据格式适配器

实例

import json

class XMLData:
    """XML数据源"""
    def get_xml_data(self):
        return """
        <users>
            <user>
                <name>张三</name>
                <age>25</age>
            </user>
            <user>
                <name>李四</name>
                <age>30</age>
            </user>
        </users>
        """

class JSONProcessor:
    """JSON处理器（期望JSON格式）"""
    def process_json(self, json_data):
        data = json.loads(json_data)
        for user in data['users']:
            print(f"姓名: {user['name']}, 年龄: {user['age']}")

class XMLToJSONAdapter:
    """XML到JSON的适配器"""
   
    def __init__(self, xml_data):
        self.xml_data = xml_data
   
    def convert_to_json(self):
        # 简化的XML到JSON转换
        xml_content = self.xml_data.get_xml_data()
        # 在实际应用中，这里应该使用xml.etree.ElementTree等库进行解析
        json_data = {
            "users": [
                {"name": "张三", "age": 25},
                {"name": "李四", "age": 30}
            ]
        }
        return json.dumps(json_data, ensure_ascii=False)

# 使用适配器
xml_source = XMLData()
adapter = XMLToJSONAdapter(xml_source)
json_processor = JSONProcessor()

json_data = adapter.convert_to_json()
json_processor.process_json(json_data)

输出结果：

姓名: 张三, 年龄: 25
姓名: 李四, 年龄: 30

案例2：第三方服务适配器

实例

class WeChatPay:
    """微信支付SDK"""
    def wechat_pay(self, amount, user_id):
        print(f"微信支付: 用户{user_id}支付{amount}元")
        return {"status": "success", "transaction_id": "wx_123456"}

class AliPay:
    """支付宝SDK"""
    def alipay(self, money, user_account):
        print(f"支付宝: 账户{user_account}支付{money}元")
        return {"code": 200, "trade_no": "ali_789012"}

class UnifiedPaymentAdapter:
    """统一支付适配器"""
   
    def __init__(self, payment_service):
        self.payment_service = payment_service
   
    def pay(self, amount, user_info):
        # 统一支付接口
        if isinstance(self.payment_service, WeChatPay):
            return self.payment_service.wechat_pay(amount, user_info)
        elif isinstance(self.payment_service, AliPay):
            return self.payment_service.alipay(amount, user_info)
        else:
            raise ValueError("不支持的支付方式")

# 客户端代码可以统一调用
def process_order(payment_adapter, amount, user_info):
    """处理订单支付"""
    result = payment_adapter.pay(amount, user_info)
    print(f"支付结果: {result}")
    return result

# 使用示例
wechat_pay = WeChatPay()
ali_pay = AliPay()

wechat_adapter = UnifiedPaymentAdapter(wechat_pay)
ali_adapter = UnifiedPaymentAdapter(ali_pay)

# 统一的调用方式
process_order(wechat_adapter, 100, "user_001")
process_order(ali_adapter, 200, "user_002")

输出结果：

微信支付: 用户user_001支付100元
支付结果: {'status': 'success', 'transaction_id': 'wx_123456'}
支付宝: 账户user_002支付200元
支付结果: {'code': 200, 'trade_no': 'ali_789012'}

适配器模式的最佳实践

1. 选择合适的实现方式

实现方式 适用场景 优点 缺点
类适配器 适配者类不多，且目标接口简单 实现简单，代码量少 需要多重继承，可能造成类层次复杂
对象适配器 大多数场景，特别是需要适配多个类时 更灵活，符合组合优于继承原则 需要创建适配器对象

2. 保持适配器的单一职责

实例

# 好的实践：职责单一
class DataFormatAdapter:
    """只负责数据格式转换"""
    def convert_format(self, data):
        pass

# 不好的实践：职责过多
class BadAdapter:
    """同时处理格式转换、验证、缓存等"""
    def convert_format(self, data):
        pass
   
    def validate_data(self, data):
        pass
   
    def cache_data(self, data):
        pass

3. 使用接口抽象

实例

from abc import ABC, abstractmethod

class PaymentInterface(ABC):
    """支付接口抽象"""
    @abstractmethod
    def pay(self, amount, user_info):
        pass

class LoggerInterface(ABC):
    """日志接口抽象"""
    @abstractmethod
    def log(self, level, message):
        pass

常见问题与解决方案

问题1：适配器过多导致复杂度增加

解决方案：使用工厂模式创建适配器

实例

class AdapterFactory:
    """适配器工厂"""
   
    @staticmethod
    def create_payment_adapter(service_type):
        if service_type == "wechat":
            return UnifiedPaymentAdapter(WeChatPay())
        elif service_type == "alipay":
            return UnifiedPaymentAdapter(AliPay())
        else:
            raise ValueError(f"不支持的支付类型: {service_type}")

# 使用工厂创建适配器
wechat_adapter = AdapterFactory.create_payment_adapter("wechat")
ali_adapter = AdapterFactory.create_payment_adapter("alipay")

问题2：性能考虑

解决方案：实现懒加载和缓存

实例

class CachingAdapter:
    """带缓存的适配器"""
   
    def __init__(self, adaptee):
        self.adaptee = adaptee
        self._cache = {}
   
    def process_data(self, key):
        if key not in self._cache:
            # 模拟耗时操作
            result = self.adaptee.expensive_operation(key)
            self._cache[key] = result
        return self._cache[key]

实践练习

练习1：创建文件格式适配器

请实现一个文件格式适配器，将 CSV 格式的数据转换为 JSON 格式：

实例

class CSVReader:
    """CSV文件读取器"""
    def read_csv(self, filepath):
        # 模拟读取CSV文件
        return [
            ["name", "age", "city"],
            ["Alice", "25", "Beijing"],
            ["Bob", "30", "Shanghai"]
        ]

class JSONExporter:
    """JSON导出器"""
    def export_json(self, data):
        import json
        print("导出JSON数据:")
        print(json.dumps(data, indent=2, ensure_ascii=False))

# 请在这里实现 CSVToJSONAdapter 类
# 你的代码...

# 测试代码
csv_reader = CSVReader()
json_exporter = JSONExporter()

# 创建适配器并测试
adapter = CSVToJSONAdapter(csv_reader)
json_data = adapter.convert_to_json()
json_exporter.export_json(json_data)

练习2：设计思考

考虑以下场景，你会如何使用适配器模式：

  1. 你的应用程序需要支持多种数据库（MySQL、PostgreSQL、SQLite），但每个数据库的连接方式不同
  2. 你需要集成多个第三方API，但它们的认证方式和返回格式各不相同
  3. 你的系统需要支持多种消息队列（RabbitMQ、Kafka、Redis）

请为其中一个场景设计适配器模式的解决方案。

总结

适配器模式是解决接口不兼容问题的有效工具。通过本文的学习，你应该掌握：

  • 适配器模式的基本概念和适用场景
  • 类适配器和对象适配器的实现方式
  • 在实际项目中的应用技巧
  • 适配器模式的最佳实践和注意事项

记住，适配器模式的核心思想是"转换"而不是"修改"，它帮助我们在不改变现有代码的情况下集成新的功能，是维护大型系统时的重要设计模式。

在实际开发中，合理使用适配器模式可以让你的代码更加灵活、可维护，并且易于扩展。