← 返回首页
🧩

结构型模式

📂 python ⏱ 4 min 649 words

结构型模式

结构型模式关注类和对象的组合方式,通过组合来形成更大的结构。本文将介绍四种重要的结构型模式:适配器、装饰器、代理和外观模式。

适配器模式

适配器模式将一个类的接口转换成客户希望的另一个接口,使原本接口不兼容的类可以一起工作。

对象适配器

from abc import ABC, abstractmethod

class MediaPlayer(ABC):
    """媒体播放器接口"""
    @abstractmethod
    def play(self, filename: str) -> str:
        pass

class VLCLib:
    """VLC库(第三方库)"""
    def vlc_play(self, file: str) -> str:
        return f"VLC正在播放: {file}"

class FFmpegLib:
    """FFmpeg库(第三方库)"""
    def ffmpeg_decode_and_play(self, path: str) -> str:
        return f"FFmpeg解码并播放: {path}"

class VLCAdapter(MediaPlayer):
    """VLC适配器"""
    
    def __init__(self):
        self.vlc = VLCLib()
    
    def play(self, filename: str) -> str:
        return self.vlc.vlc_play(filename)

class FFmpegAdapter(MediaPlayer):
    """FFmpeg适配器"""
    
    def __init__(self):
        self.ffmpeg = FFmpegLib()
    
    def play(self, filename: str) -> str:
        return self.ffmpeg.ffmpeg_decode_and_play(filename)

# 统一使用MediaPlayer接口
def play_media(player: MediaPlayer, file: str):
    print(player.play(file))

# 使用
vlc = VLCAdapter()
ffmpeg = FFmpegAdapter()

play_media(vlc, "song.mp3")     # VLC正在播放: song.mp3
play_media(ffmpeg, "movie.mkv") # FFmpeg解码并播放: movie.mkv

类适配器(使用多重继承)

class OldPrinter:
    """旧式打印机"""
    def old_print(self, text: str) -> str:
        return f"旧式打印: {text}"

class NewPrinterInterface:
    """新打印机接口"""
    @abstractmethod
    def print_document(self, text: str, copies: int = 1) -> str:
        pass

class PrinterAdapter(NewPrinterInterface, OldPrinter):
    """使用多重继承的适配器"""
    
    def print_document(self, text: str, copies: int = 1) -> str:
        results = []
        for _ in range(copies):
            results.append(self.old_print(text))
        return " | ".join(results)

adapter = PrinterAdapter()
print(adapter.print_document("测试文档", 2))
# 旧式打印: 测试文档 | 旧式打印: 测试文档

装饰器模式

装饰器模式动态地给对象添加额外职责,比继承更灵活。Python内置支持装饰器语法。

类装饰器实现

from functools import wraps
from typing import Callable, Any

class DataProcessor:
    """数据处理器"""
    def process(self, data: list) -> list:
        return [x * 2 for x in data]

class ProcessorDecorator:
    """处理器装饰器基类"""
    
    def __init__(self, processor):
        self._processor = processor
    
    def process(self, data: list) -> list:
        return self._processor.process(data)

class FilterDecorator(ProcessorDecorator):
    """过滤装饰器"""
    
    def __init__(self, processor, filter_func: Callable):
        super().__init__(processor)
        self.filter_func = filter_func
    
    def process(self, data: list) -> list:
        filtered = filter(self.filter_func, data)
        return self._processor.process(list(filtered))

class SortDecorator(ProcessorDecorator):
    """排序装饰器"""
    
    def __init__(self, processor, reverse: bool = False):
        super().__init__(processor)
        self.reverse = reverse
    
    def process(self, data: list) -> list:
        sorted_data = sorted(data, reverse=self.reverse)
        return self._processor.process(sorted_data)

class LogDecorator(ProcessorDecorator):
    """日志装饰器"""
    
    def process(self, data: list) -> list:
        print(f"处理前: {data}")
        result = self._processor.process(data)
        print(f"处理后: {result}")
        return result

# 使用装饰器链
processor = DataProcessor()
decorated = LogDecorator(
    SortDecorator(
        FilterDecorator(
            processor,
            lambda x: x > 0
        ),
        reverse=True
    )
)

result = decorated.process([3, -1, 4, -2, 5])
# 处理前: [3, -1, 4, -2, 5]
# 处理后: [10, 8, 6]

使用函数装饰器

import time
from functools import wraps

def cache_result(max_size: int = 128):
    """带大小限制的缓存装饰器"""
    def decorator(func):
        cache = {}
        order = []
        
        @wraps(func)
        def wrapper(*args):
            if args in cache:
                return cache[args]
            
            result = func(*args)
            
            if len(cache) >= max_size:
                oldest = order.pop(0)
                del cache[oldest]
            
            cache[args] = result
            order.append(args)
            return result
        
        wrapper.cache_info = lambda: f"缓存大小: {len(cache)}/{max_size}"
        return wrapper
    return decorator

@cache_result(max_size=3)
def expensive_computation(n: int) -> int:
    time.sleep(0.1)  # 模拟耗时计算
    return n ** 2

print(expensive_computation(5))  # 计算
print(expensive_computation(5))  # 从缓存获取
print(expensive_computation.cache_info())  # 缓存大小: 1/3

代理模式

代理模式为其他对象提供一种代理以控制对这个对象的访问。

延迟加载代理

class HighResImage:
    """高分辨率图像(加载成本高)"""
    
    def __init__(self, filename: str):
        self.filename = filename
        self._load_from_disk()
    
    def _load_from_disk(self):
        print(f"从磁盘加载高分辨率图像: {self.filename}")
        # 模拟耗时加载
        self.data = f"高分辨率图像数据({self.filename})"
    
    def display(self):
        print(f"显示: {self.data}")

class ImageProxy:
    """图像代理,实现延迟加载"""
    
    def __init__(self, filename: str):
        self.filename = filename
        self._image = None
    
    def _load_image(self):
        if self._image is None:
            self._image = HighResImage(self.filename)
    
    def display(self):
        self._load_image()
        self._image.display()

# 使用代理,不会立即加载图像
proxy = ImageProxy("photo.jpg")
print("代理创建完成")  # 立即返回

# 第一次调用时才真正加载
proxy.display()  # 从磁盘加载高分辨率图像: photo.jpg
proxy.display()  # 直接显示,不再加载

访问控制代理

class ProtectedResource:
    """受保护的资源"""
    
    def __init__(self, data: str, owner: str):
        self.data = data
        self.owner = owner
    
    def read(self) -> str:
        return self.data
    
    def write(self, new_data: str):
        self.data = new_data

class AccessProxy:
    """访问控制代理"""
    
    def __init__(self, resource: ProtectedResource, current_user: str):
        self._resource = resource
        self._user = current_user
    
    def read(self) -> str:
        if self._user == self._resource.owner:
            return self._resource.read()
        raise PermissionError(f"用户 {self._user} 无权读取")
    
    def write(self, new_data: str):
        if self._user == self._resource.owner:
            self._resource.write(new_data)
        else:
            raise PermissionError(f"用户 {self._user} 无权写入")

# 使用
resource = ProtectedResource("机密数据", "admin")
proxy = AccessProxy(resource, "admin")
print(proxy.read())  # 机密数据

proxy2 = AccessProxy(resource, "guest")
try:
    proxy2.read()
except PermissionError as e:
    print(e)  # 用户 guest 无权读取

外观模式

外观模式为子系统中的一组接口提供一个统一的高层接口,降低使用复杂度。

class CPU:
    def freeze(self):
        print("CPU: 冻结处理器")
    
    def jump(self, address: int):
        print(f"CPU: 跳转到地址 {address}")
    
    def execute(self):
        print("CPU: 执行指令")

class Memory:
    def load(self, address: int, data: str):
        print(f"内存: 加载数据到地址 {address}")
    
    def free(self, address: int):
        print(f"内存: 释放地址 {address}")

class HardDrive:
    def read(self, sector: int, size: int) -> str:
        print(f"硬盘: 读取扇区 {sector},大小 {size}")
        return "操作系统数据"

class BIOS:
    def run(self):
        print("BIOS: 运行自检")

class ComputerFacade:
    """计算机外观,封装复杂的启动过程"""
    
    def __init__(self):
        self.cpu = CPU()
        self.memory = Memory()
        self.hard_drive = HardDrive()
        self.bios = BIOS()
    
    def start_computer(self):
        """一键启动计算机"""
        print("=== 开始启动计算机 ===")
        self.bios.run()
        self.cpu.freeze()
        boot_sector = self.hard_drive.read(0, 1024)
        self.memory.load(0, boot_sector)
        self.cpu.jump(0)
        self.cpu.execute()
        print("=== 计算机启动完成 ===\n")

# 使用外观模式,简化复杂操作
computer = ComputerFacade()
computer.start_computer()
# 所有子系统的复杂交互被封装在start_computer方法中

模式选择指南

模式 解决的问题 使用场景
适配器 接口不兼容 集成第三方库、遗留系统整合
装饰器 动态添加功能 日志、缓存、权限检查
代理 控制访问 延迟加载、访问控制、远程代理
外观 复杂子系统简化 简化API、分层架构

结构型模式的核心是组合:适配器转换接口,装饰器增强功能,代理控制访问,外观简化复杂性。理解它们的区别和适用场景,能帮助你设计出更灵活、可维护的系统。