24.1 设计模式基础

24.1.1 设计模式分类

24.1.2 SOLID原则

24.2 创建型模式

24.2.1 工厂模式

from abc import ABC, abstractmethod

class PaymentMethod(ABC):
    """支付方式抽象类"""
    @abstractmethod
    def pay(self, amount):
        pass

class CreditCard(PaymentMethod):
    def pay(self, amount):
        print(f"信用卡支付 {amount} 元")

class Alipay(PaymentMethod):
    def pay(self, amount):
        print(f"支付宝支付 {amount} 元")

class PaymentFactory:
    """支付工厂"""
    @staticmethod
    def create(method: str) -> PaymentMethod:
        if method == 'credit':
            return CreditCard()
        elif method == 'alipay':
            return Alipay()
        raise ValueError(f"不支持的支付方式: {method}")

# 使用示例
payment = PaymentFactory.create('alipay')
payment.pay(100)

24.2.2 单例模式

class DatabaseConnection:
    _instance = None
    
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            # 初始化数据库连接
            cls._instance.conn = "模拟数据库连接"
        return cls._instance
    
    def query(self, sql):
        print(f"执行查询: {sql}")
        return "结果数据"

# 测试单例
db1 = DatabaseConnection()
db2 = DatabaseConnection()
print(db1 is db2)  # True

24.3 结构型模式

24.3.1 适配器模式

class OldSystem:
    """旧系统接口"""
    def legacy_request(self, data: dict):
        print(f"旧系统处理: {data}")

class NewSystem:
    """新系统期望的接口"""
    def json_request(self, json_str: str):
        print(f"新系统处理: {json_str}")

class Adapter(NewSystem):
    """适配器转换旧系统接口"""
    def __init__(self, old_system: OldSystem):
        self.old_system = old_system
    
    def json_request(self, json_str: str):
        import json
        data = json.loads(json_str)
        self.old_system.legacy_request(data)

# 客户端代码
def client_code(system: NewSystem):
    system.json_request('{"key": "value"}')

old = OldSystem()
adapter = Adapter(old)
client_code(adapter)  # 通过适配器调用旧系统

24.3.2 装饰器模式

from functools import wraps

def log_execution(func):
    """记录函数执行的装饰器"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        print(f"开始执行 {func.__name__}")
        result = func(*args, **kwargs)
        print(f"完成执行 {func.__name__}")
        return result
    return wrapper

def validate_input(func):
    """验证参数的装饰器"""
    @wraps(func)
    def wrapper(text: str):
        if not isinstance(text, str):
            raise ValueError("输入必须是字符串")
        return func(text)
    return wrapper

@log_execution
@validate_input
def process_text(text):
    """处理文本"""
    return text.upper()

# 使用装饰后的函数
print(process_text("hello"))  # 正常执行
# process_text(123)          # 触发验证错误

24.4 行为型模式

24.4.1 观察者模式

class Subject:
    """被观察者"""
    def __init__(self):
        self._observers = []
    
    def attach(self, observer):
        self._observers.append(observer)
    
    def detach(self, observer):
        self._observers.remove(observer)
    
    def notify(self, data):
        for observer in self._observers:
            observer.update(data)

class Observer(ABC):
    """观察者抽象类"""
    @abstractmethod
    def update(self, data):
        pass

class Logger(Observer):
    """日志观察者"""
    def update(self, data):
        print(f"[LOG] 状态变化: {data}")

class AlertSystem(Observer):
    """告警观察者"""
    def update(self, data):
        if data.get('warning'):
            print(f"[ALERT] 警告: {data['warning']}")

# 使用示例
subject = Subject()
subject.attach(Logger())
subject.attach(AlertSystem())

subject.notify({"status": "OK"})
subject.notify({"warning": "CPU过载"})

24.4.2 策略模式

from abc import ABC, abstractmethod

class CompressionStrategy(ABC):
    """压缩策略接口"""
    @abstractmethod
    def compress(self, data: bytes) -> bytes:
        pass

class ZipCompression(CompressionStrategy):
    def compress(self, data):
        print("使用ZIP算法压缩")
        return b"compressed_with_zip"

class RarCompression(CompressionStrategy):
    def compress(self, data):
        print("使用RAR算法压缩")
        return b"compressed_with_rar"

class FileCompressor:
    """文件压缩器"""
    def __init__(self, strategy: CompressionStrategy):
        self._strategy = strategy
    
    def set_strategy(self, strategy: CompressionStrategy):
        self._strategy = strategy
    
    def compress_file(self, file_path: str) -> bytes:
        with open(file_path, 'rb') as f:
            data = f.read()
        return self._strategy.compress(data)

# 使用示例
compressor = FileCompressor(ZipCompression())
compressor.compress_file("test.txt")

# 运行时切换策略
compressor.set_strategy(RarCompression())
compressor.compress_file("test.txt")

24.5 系统架构

24.5.1 分层架构

myapp/
├── presentation/   # 表现层
│   ├── web/
│   └── cli/
├── service/        # 业务逻辑层
│   ├── order.py
│   └── payment.py
├── repository/     # 数据访问层
│   ├── models.py
│   └── db.py
└── domain/         # 领域模型
    ├── entities.py
    └── value_objects.py

24.5.2 微服务通信

# 基于消息队列的通信
import pika

def publish_event(event):
    connection = pika.BlockingConnection(
        pika.ConnectionParameters('localhost')
    )
    channel = connection.channel()
    channel.queue_declare(queue='order_events')
    channel.basic_publish(
        exchange='',
        routing_key='order_events',
        body=json.dumps(event)
    )
    connection.close()

# 基于REST的通信
import requests

def get_user_info(user_id):
    response = requests.get(
        f"http://user-service/api/users/{user_id}",
        timeout=3
    )
    response.raise_for_status()
    return response.json()

24.6 应用举例

案例1：电商订单

from dataclasses import dataclass
from typing import List

@dataclass
class OrderItem:
    product_id: str
    quantity: int
    price: float

class Order:
    """订单领域模型"""
    def __init__(self, order_id: str, user_id: str):
        self.order_id = order_id
        self.user_id = user_id
        self._items: List[OrderItem] = []
        self._status = "created"
    
    def add_item(self, item: OrderItem):
        self._items.append(item)
    
    @property
    def total_amount(self):
        return sum(item.price * item.quantity for item in self._items)
    
    def checkout(self, payment_strategy):
        """策略模式应用"""
        payment_strategy.process(self.total_amount)
        self._status = "paid"
    
    def notify_observers(self, event):
        """观察者模式应用"""
        for observer in self._observers:
            observer.update(event)

class OrderService:
    """订单服务层"""
    def __init__(self, repository, notifier):
        self._repo = repository
        self._notifier = notifier
    
    def create_order(self, user_id, items):
        order = Order(str(uuid.uuid4()), user_id)
        for item in items:
            order.add_item(OrderItem(**item))
        
        self._repo.save(order)
        self._notifier.send(
            "order_created",
            {"order_id": order.order_id, "user_id": user_id}
        )
        return order

案例2：数据管道架构

from abc import ABC, abstractmethod
from concurrent.futures import ThreadPoolExecutor

class DataSource(ABC):
    """数据源抽象类"""
    @abstractmethod
    def read(self):
        pass

class Transformer(ABC):
    """数据转换抽象类"""
    @abstractmethod
    def transform(self, data):
        pass

class DataPipeline:
    """数据处理管道"""
    def __init__(self, source: DataSource, transformers: List[Transformer]):
        self.source = source
        self.transformers = transformers
        self.executor = ThreadPoolExecutor(max_workers=4)
    
    def process(self):
        """责任链模式处理数据"""
        raw_data = self.source.read()
        
        result = raw_data
        for transformer in self.transformers:
            result = self.executor.submit(transformer.transform, result).result()
        
        return result

# 具体实现
class CSVSource(DataSource):
    def read(self):
        print("从CSV读取数据")
        return [{"name": "Alice", "age": "25"}, {"name": "Bob", "age": "30"}]

class AgeToInt(Transformer):
    def transform(self, data):
        print("转换年龄为整数")
        for item in data:
            item['age'] = int(item['age'])
        return data

class AddTimestamp(Transformer):
    def transform(self, data):
        print("添加时间戳")
        for item in data:
            item['timestamp'] = datetime.now().isoformat()
        return data

# 构建管道
pipeline = DataPipeline(
    source=CSVSource(),
    transformers=[AgeToInt(), AddTimestamp()]
)
processed_data = pipeline.process()
print(processed_data)

24.7 知识图谱

24.8 学习总结

核心要点：

• 理解SOLID设计原则
• 掌握常用设计模式
• 区分架构层次职责
• 合理应用模式组合

持续更新Python编程学习日志与技巧，敬请关注！

#编程# #学习# #python# #在头条记录我的2025#