第7章桥接模式

学习目标

完成本章学习后，读者将能够：

理解桥接模式的核心概念、形式化定义与理论基础
掌握抽象与实现分离的设计原则与实现机制
设计处理多维度变化的类型层次结构
运用Protocol实现类型安全的桥接模式
识别桥接模式与适配器模式、策略模式的边界
评估桥接模式在复杂系统设计中的权衡取舍

7.1 模式定义

7.1.1 形式化定义

桥接模式（Bridge Pattern） 将抽象部分与实现部分分离，使它们都可以独立地变化。

从形式化角度，桥接模式可定义为笛卡尔积的解耦：

$$ \text{Bridge}: \mathcal{A} \times \mathcal{I} \rightarrow \mathcal{A} \oplus \mathcal{I} $$

其中：

$\mathcal{A}$ 为抽象层（Abstraction Layer）
$\mathcal{I}$ 为实现层（Implementation Layer）
$\oplus$ 表示正交组合

类爆炸问题的形式化分析：

假设系统有两个独立变化维度：

维度1（抽象）：$n$ 种变化
维度2（实现）：$m$ 种变化

使用继承方案： $$ \text{Classes}_{\text{inheritance}} = n \times m $$

使用桥接模式： $$ \text{Classes}_{\text{bridge}} = n + m $$

复杂度降低比例： $$ \text{Reduction} = 1 - \frac{n + m}{n \times m} = 1 - \frac{1}{n} - \frac{1}{m} $$

当 $n, m \geq 3$ 时，复杂度降低超过 33%。

桥接模式的组合约束：

$$ \forall a \in \mathcal{A}, \forall i \in \mathcal{I}: \text{valid}(a, i) \Rightarrow a.\text{set_implementation}(i) $$

即任何抽象可以与任何实现组合，前提是实现满足抽象的接口契约。

7.1.2 历史背景与学术渊源

桥接模式的概念源于软件架构的演进：

1. "优先组合 over 继承"原则

桥接模式是"组合优于继承"（Composition over Inheritance）原则的经典应用。该原则可追溯至1980年代的面向对象设计理论，强调通过组合实现灵活性。

2. GoF分类（1994）

Gamma等人在《Design Patterns》中将桥接模式归类为结构型模式，强调其"将抽象与实现解耦"的核心价值。GoF指出：

"Decouple an abstraction from its implementation so that the two can vary independently."

3. Pimpl惯用法的影响

C++中的Pimpl（Pointer to Implementation）惯用法与桥接模式有相似的设计意图：隐藏实现细节，降低编译依赖。桥接模式可视为Pimpl的泛化形式。

4. 桥接模式与策略模式的关系

桥接模式与策略模式结构相似，但意图不同：

桥接模式：分离抽象与实现（结构型）
策略模式：封装可互换的算法（行为型）

7.1.3 模式动机

桥接模式解决以下核心问题：

问题1：多维度变化的类爆炸

python

class Shape:
    pass

class Circle(Shape):
    pass

class Square(Shape):
    pass

class VectorCircle(Circle):
    pass

class RasterCircle(Circle):
    pass

class VectorSquare(Square):
    pass

class RasterSquare(Square):
    pass

当增加新的形状或渲染方式时，类数量呈乘法增长。

问题2：抽象与实现的紧耦合

python

class BadShape:
    def __init__(self, render_type: str):
        if render_type == "vector":
            self._render_vector()
        elif render_type == "raster":
            self._render_raster()

抽象与实现耦合在同一类中，难以独立扩展。

问题3：运行时实现切换困难

继承方案在编译时确定实现，无法在运行时动态切换。

7.1.4 UML结构模型

┌─────────────────────────────────────────────────────────────┐
│                      桥接模式结构                            │
└─────────────────────────────────────────────────────────────┘

┌─────────────────────────┐       ┌─────────────────────────┐
│     Abstraction         │       │    Implementation       │
│     (抽象层)            │       │    (实现接口)           │
├─────────────────────────┤       ├─────────────────────────┤
│ # impl: Implementation  │──────▶│ + operation_impl()      │
├─────────────────────────┤ 组合  │ + another_operation()   │
│ + operation()           │       └────────────△────────────┘
│ + set_impl(impl)        │                    │
└────────────△────────────┘           ┌────────┴────────┐
             │                        │                 │
             │ 继承          ┌────────┴────┐    ┌──────┴──────┐
             │              │ ConcreteA   │    │ ConcreteB   │
┌────────────┴────────────┐ │ Impl        │    │ Impl        │
│   RefinedAbstraction    │ ├─────────────┤    ├─────────────┤
│   (扩展抽象)            │ │+operation_  │    │+operation_  │
├─────────────────────────┤ │ impl()      │    │ impl()      │
│ + operation()           │ └─────────────┘    └─────────────┘
│ + new_operation()       │
└─────────────────────────┘


┌─────────────────────────────────────────────────────────────┐
│                      多维度变化示意                          │
└─────────────────────────────────────────────────────────────┘

维度1：抽象层变化
┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│   Abstraction   │────▶│ RefinedAbs A    │────▶│ RefinedAbs B    │
└─────────────────┘     └─────────────────┘     └─────────────────┘

维度2：实现层变化
┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│ Implementation  │────▶│ ConcreteImpl A  │     │ ConcreteImpl B  │
└─────────────────┘     └─────────────────┘     └─────────────────┘

组合结果：
┌─────────────────────────────────────────────────────────────┐
│  RefinedAbs A + ConcreteImpl A                              │
│  RefinedAbs A + ConcreteImpl B                              │
│  RefinedAbs B + ConcreteImpl A                              │
│  RefinedAbs B + ConcreteImpl B                              │
└─────────────────────────────────────────────────────────────┘

参与者职责：

参与者	职责
Abstraction	定义抽象接口，维护对Implementation的引用
RefinedAbstraction	扩展Abstraction接口
Implementation	定义实现接口，不与Abstraction对应
ConcreteImplementation	实现Implementation接口的具体类

7.2 Python实现机制

7.2.1 标准实现

python

from abc import ABC, abstractmethod
from typing import TypeVar, Generic, Optional

T = TypeVar('T')

class Implementation(ABC):
    """
    实现接口：定义所有具体实现必须提供的操作
    
    这是"实现"维度的抽象层
    不与Abstraction的接口直接对应
    """
    
    @abstractmethod
    def operation_impl(self) -> str:
        """基础操作"""
        pass
    
    @abstractmethod
    def another_operation(self, data: str) -> str:
        """另一个操作"""
        pass


class ConcreteImplementationA(Implementation):
    """具体实现A：提供一种实现方式"""
    
    def operation_impl(self) -> str:
        return "ImplementationA: 执行基础操作"
    
    def another_operation(self, data: str) -> str:
        return f"ImplementationA: 处理 '{data}'"


class ConcreteImplementationB(Implementation):
    """具体实现B：提供另一种实现方式"""
    
    def operation_impl(self) -> str:
        return "ImplementationB: 执行基础操作（优化版）"
    
    def another_operation(self, data: str) -> str:
        return f"ImplementationB: 高效处理 '{data.upper()}'"


class Abstraction:
    """
    抽象类：定义抽象接口
    
    这是"抽象"维度的抽象层
    持有Implementation引用，将操作委托给实现
    """
    
    def __init__(self, implementation: Implementation):
        self._implementation = implementation
    
    def set_implementation(self, implementation: Implementation) -> None:
        """
        运行时切换实现
        
        这是桥接模式的关键优势之一
        """
        self._implementation = implementation
    
    def operation(self) -> str:
        """抽象操作：委托给实现"""
        return f"Abstraction: Base -> {self._implementation.operation_impl()}"
    
    def combined_operation(self, data: str) -> str:
        """组合多个实现操作"""
        base = self._implementation.operation_impl()
        processed = self._implementation.another_operation(data)
        return f"Abstraction: Combined\n  {base}\n  {processed}"


class RefinedAbstraction(Abstraction):
    """
    扩展抽象：添加新的抽象操作
    
    可以复用任何Implementation
    """
    
    def enhanced_operation(self) -> str:
        """增强操作：组合实现操作"""
        return f"RefinedAbstraction: Enhanced -> {self._implementation.operation_impl()}"
    
    def operation(self) -> str:
        """重写操作：添加额外行为"""
        return f"RefinedAbstraction: {self._implementation.operation_impl()}"


def client_code(abstraction: Abstraction) -> None:
    """客户端代码：仅依赖抽象接口"""
    print(abstraction.operation())


def demo_basic_bridge():
    impl_a = ConcreteImplementationA()
    impl_b = ConcreteImplementationB()
    
    abstraction = Abstraction(impl_a)
    print("=== 使用实现A ===")
    print(abstraction.operation())
    print(abstraction.combined_operation("test data"))
    
    print("\n=== 运行时切换到实现B ===")
    abstraction.set_implementation(impl_b)
    print(abstraction.operation())
    print(abstraction.combined_operation("test data"))
    
    print("\n=== 使用扩展抽象 ===")
    refined = RefinedAbstraction(impl_a)
    print(refined.operation())
    print(refined.enhanced_operation())

demo_basic_bridge()

7.2.2 基于Protocol的现代实现

python

from typing import Protocol, runtime_checkable, TypeVar, Any
from dataclasses import dataclass

@runtime_checkable
class Renderer(Protocol):
    """
    渲染器协议：定义渲染操作的接口
    
    使用Protocol实现结构化类型
    任何实现这些方法的类都自动满足协议
    """
    
    def draw_circle(self, x: float, y: float, radius: float, color: str) -> str:
        ...
    
    def draw_rectangle(self, x: float, y: float, width: float, height: float, color: str) -> str:
        ...
    
    def draw_line(self, x1: float, y1: float, x2: float, y2: float, color: str) -> str:
        ...


class VectorRenderer:
    """矢量渲染器：隐式实现Renderer协议"""
    
    def draw_circle(self, x: float, y: float, radius: float, color: str) -> str:
        return f"矢量圆: 中心({x}, {y}), 半径{radius}, 颜色{color}"
    
    def draw_rectangle(self, x: float, y: float, width: float, height: float, color: str) -> str:
        return f"矢量矩形: ({x}, {y}), {width}×{height}, 颜色{color}"
    
    def draw_line(self, x1: float, y1: float, x2: float, y2: float, color: str) -> str:
        return f"矢量线: ({x1}, {y1}) → ({x2}, {y2}), 颜色{color}"


class RasterRenderer:
    """光栅渲染器：隐式实现Renderer协议"""
    
    def draw_circle(self, x: float, y: float, radius: float, color: str) -> str:
        return f"像素圆: 中心({int(x)}, {int(y)}), 半径{int(radius)}px, 颜色{color}"
    
    def draw_rectangle(self, x: float, y: float, width: float, height: float, color: str) -> str:
        return f"像素矩形: ({int(x)}, {int(y)}), {int(width)}×{int(height)}px, 颜色{color}"
    
    def draw_line(self, x1: float, y1: float, x2: float, y2: float, color: str) -> str:
        return f"像素线: ({int(x1)}, {int(y1)}) → ({int(x2)}, {int(y2)}), 颜色{color}"


class SVGRenderer:
    """SVG渲染器：生成SVG代码"""
    
    def draw_circle(self, x: float, y: float, radius: float, color: str) -> str:
        return f'<circle cx="{x}" cy="{y}" r="{radius}" fill="{color}"/>'
    
    def draw_rectangle(self, x: float, y: float, width: float, height: float, color: str) -> str:
        return f'<rect x="{x}" y="{y}" width="{width}" height="{height}" fill="{color}"/>'
    
    def draw_line(self, x1: float, y1: float, x2: float, y2: float, color: str) -> str:
        return f'<line x1="{x1}" y1="{y1}" x2="{x2}" y2="{y2}" stroke="{color}"/>'


@dataclass
class Point:
    x: float
    y: float


class Shape:
    """
    图形抽象：定义图形接口
    
    持有渲染器引用，将绘制操作委托给渲染器
    """
    
    def __init__(self, renderer: Renderer, color: str = "black"):
        self._renderer = renderer
        self._color = color
    
    def set_renderer(self, renderer: Renderer) -> None:
        self._renderer = renderer
    
    def set_color(self, color: str) -> None:
        self._color = color
    
    def draw(self) -> str:
        raise NotImplementedError


class Circle(Shape):
    """圆形：扩展抽象"""
    
    def __init__(self, renderer: Renderer, center: Point, radius: float, color: str = "black"):
        super().__init__(renderer, color)
        self._center = center
        self._radius = radius
    
    @property
    def center(self) -> Point:
        return self._center
    
    @center.setter
    def center(self, value: Point) -> None:
        self._center = value
    
    @property
    def radius(self) -> float:
        return self._radius
    
    @radius.setter
    def radius(self, value: float) -> None:
        self._radius = value
    
    def draw(self) -> str:
        return self._renderer.draw_circle(
            self._center.x, self._center.y, self._radius, self._color
        )


class Rectangle(Shape):
    """矩形：扩展抽象"""
    
    def __init__(self, renderer: Renderer, origin: Point, width: float, height: float, color: str = "black"):
        super().__init__(renderer, color)
        self._origin = origin
        self._width = width
        self._height = height
    
    def draw(self) -> str:
        return self._renderer.draw_rectangle(
            self._origin.x, self._origin.y, self._width, self._height, self._color
        )


class Line(Shape):
    """线段：扩展抽象"""
    
    def __init__(self, renderer: Renderer, start: Point, end: Point, color: str = "black"):
        super().__init__(renderer, color)
        self._start = start
        self._end = end
    
    def draw(self) -> str:
        return self._renderer.draw_line(
            self._start.x, self._start.y, self._end.x, self._end.y, self._color
        )


def demo_protocol_bridge():
    vector = VectorRenderer()
    raster = RasterRenderer()
    svg = SVGRenderer()
    
    print("=== 矢量渲染 ===")
    circle = Circle(vector, Point(100, 100), 50, "red")
    print(circle.draw())
    
    print("\n=== 切换到光栅渲染 ===")
    circle.set_renderer(raster)
    print(circle.draw())
    
    print("\n=== 切换到SVG渲染 ===")
    circle.set_renderer(svg)
    print(circle.draw())
    
    print("\n=== 验证协议实现 ===")
    print(f"VectorRenderer是否实现Renderer: {isinstance(vector, Renderer)}")
    print(f"RasterRenderer是否实现Renderer: {isinstance(raster, Renderer)}")
    print(f"SVGRenderer是否实现Renderer: {isinstance(svg, Renderer)}")

demo_protocol_bridge()

7.2.3 泛型桥接模式

python

from abc import ABC, abstractmethod
from typing import TypeVar, Generic, Callable, Any, Optional
from dataclasses import dataclass

T = TypeVar('T')
R = TypeVar('R')

class Implementation(ABC, Generic[T, R]):
    """
    泛型实现接口
    
    T: 输入类型
    R: 输出类型
    """
    
    @abstractmethod
    def execute(self, input_data: T) -> R:
        pass


class Abstraction(ABC, Generic[T, R]):
    """
    泛型抽象接口
    
    T: 输入类型
    R: 输出类型
    """
    
    def __init__(self, implementation: Implementation[T, R]):
        self._implementation = implementation
    
    @abstractmethod
    def process(self, input_data: T) -> R:
        pass
    
    def set_implementation(self, implementation: Implementation[T, R]) -> None:
        self._implementation = implementation


@dataclass
class Query:
    """查询数据"""
    table: str
    conditions: dict
    fields: list


@dataclass
class QueryResult:
    """查询结果"""
    rows: list
    row_count: int
    execution_time: float


class DatabaseImplementation(Implementation[Query, QueryResult], ABC):
    """数据库实现基类"""
    
    @abstractmethod
    def connect(self) -> bool:
        pass
    
    @abstractmethod
    def disconnect(self) -> None:
        pass


class MySQLImplementation(DatabaseImplementation):
    """MySQL实现"""
    
    def connect(self) -> bool:
        print("MySQL: 连接成功")
        return True
    
    def disconnect(self) -> None:
        print("MySQL: 断开连接")
    
    def execute(self, input_data: Query) -> QueryResult:
        print(f"MySQL: 执行查询 {input_data.table}")
        return QueryResult(
            rows=[{"id": 1, "name": "MySQL用户"}],
            row_count=1,
            execution_time=0.01
        )


class PostgreSQLImplementation(DatabaseImplementation):
    """PostgreSQL实现"""
    
    def connect(self) -> bool:
        print("PostgreSQL: 连接成功")
        return True
    
    def disconnect(self) -> None:
        print("PostgreSQL: 断开连接")
    
    def execute(self, input_data: Query) -> QueryResult:
        print(f"PostgreSQL: 执行查询 {input_data.table}")
        return QueryResult(
            rows=[{"id": 1, "name": "PostgreSQL用户"}],
            row_count=1,
            execution_time=0.008
        )


class Repository(Abstraction[Query, QueryResult]):
    """仓库抽象"""
    
    def __init__(self, db: DatabaseImplementation):
        super().__init__(db)
        self._db = db
    
    def process(self, input_data: Query) -> QueryResult:
        return self._implementation.execute(input_data)
    
    def find_by_id(self, table: str, id: int) -> QueryResult:
        query = Query(
            table=table,
            conditions={"id": id},
            fields=["*"]
        )
        return self.process(query)
    
    def find_all(self, table: str) -> QueryResult:
        query = Query(
            table=table,
            conditions={},
            fields=["*"]
        )
        return self.process(query)


class CachedRepository(Repository):
    """带缓存的仓库：扩展抽象"""
    
    def __init__(self, db: DatabaseImplementation):
        super().__init__(db)
        self._cache: dict = {}
    
    def process(self, input_data: Query) -> QueryResult:
        cache_key = f"{input_data.table}:{input_data.conditions}"
        
        if cache_key in self._cache:
            print("  [缓存命中]")
            return self._cache[cache_key]
        
        print("  [缓存未命中]")
        result = super().process(input_data)
        self._cache[cache_key] = result
        return result


def demo_generic_bridge():
    mysql = MySQLImplementation()
    pg = PostgreSQLImplementation()
    
    mysql.connect()
    
    repo = Repository(mysql)
    print("\n=== MySQL查询 ===")
    result = repo.find_by_id("users", 1)
    print(f"结果: {result.rows}")
    
    print("\n=== 切换到PostgreSQL ===")
    repo.set_implementation(pg)
    pg.connect()
    result = repo.find_all("users")
    print(f"结果: {result.rows}")
    
    print("\n=== 使用缓存仓库 ===")
    cached_repo = CachedRepository(mysql)
    cached_repo.find_by_id("users", 1)
    cached_repo.find_by_id("users", 1)
    
    mysql.disconnect()
    pg.disconnect()

demo_generic_bridge()

7.3 高级应用技术

7.3.1 多维度桥接

python

from abc import ABC, abstractmethod
from typing import Protocol, runtime_checkable
from dataclasses import dataclass
from enum import Enum

class LogLevel(Enum):
    DEBUG = "debug"
    INFO = "info"
    WARNING = "warning"
    ERROR = "error"


@runtime_checkable
class LogFormatter(Protocol):
    """日志格式化器：维度1"""
    
    def format(self, level: LogLevel, message: str, context: dict) -> str:
        ...


@runtime_checkable
class LogWriter(Protocol):
    """日志写入器：维度2"""
    
    def write(self, formatted_message: str) -> None:
        ...


class TextFormatter:
    """文本格式化器"""
    
    def format(self, level: LogLevel, message: str, context: dict) -> str:
        import datetime
        timestamp = datetime.datetime.now().isoformat()
        context_str = " | ".join(f"{k}={v}" for k, v in context.items())
        return f"[{timestamp}] [{level.value.upper()}] {message} | {context_str}"


class JSONFormatter:
    """JSON格式化器"""
    
    def format(self, level: LogLevel, message: str, context: dict) -> str:
        import json
        import datetime
        entry = {
            "timestamp": datetime.datetime.now().isoformat(),
            "level": level.value,
            "message": message,
            "context": context
        }
        return json.dumps(entry, ensure_ascii=False)


class ConsoleWriter:
    """控制台写入器"""
    
    def write(self, formatted_message: str) -> None:
        print(formatted_message)


class FileWriter:
    """文件写入器"""
    
    def __init__(self, file_path: str):
        self._file_path = file_path
    
    def write(self, formatted_message: str) -> None:
        with open(self._file_path, 'a', encoding='utf-8') as f:
            f.write(formatted_message + '\n')


class SyslogWriter:
    """系统日志写入器（模拟）"""
    
    def __init__(self, host: str, port: int):
        self._host = host
        self._port = port
    
    def write(self, formatted_message: str) -> None:
        print(f"[SYSLOG {self._host}:{self._port}] {formatted_message}")


class Logger:
    """
    日志器：桥接两个维度
    
    维度1：格式化方式（Text, JSON, XML...）
    维度2：输出目标（Console, File, Syslog...）
    """
    
    def __init__(self, formatter: LogFormatter, writer: LogWriter):
        self._formatter = formatter
        self._writer = writer
    
    def set_formatter(self, formatter: LogFormatter) -> None:
        self._formatter = formatter
    
    def set_writer(self, writer: LogWriter) -> None:
        self._writer = writer
    
    def log(self, level: LogLevel, message: str, **context) -> None:
        formatted = self._formatter.format(level, message, context)
        self._writer.write(formatted)
    
    def debug(self, message: str, **context) -> None:
        self.log(LogLevel.DEBUG, message, **context)
    
    def info(self, message: str, **context) -> None:
        self.log(LogLevel.INFO, message, **context)
    
    def warning(self, message: str, **context) -> None:
        self.log(LogLevel.WARNING, message, **context)
    
    def error(self, message: str, **context) -> None:
        self.log(LogLevel.ERROR, message, **context)


def demo_multi_dimension_bridge():
    text_formatter = TextFormatter()
    json_formatter = JSONFormatter()
    console_writer = ConsoleWriter()
    file_writer = FileWriter("app.log")
    syslog_writer = SyslogWriter("logs.example.com", 514)
    
    print("=== 文本格式 + 控制台输出 ===")
    logger = Logger(text_formatter, console_writer)
    logger.info("用户登录", user_id=123, ip="192.168.1.1")
    
    print("\n=== JSON格式 + 控制台输出 ===")
    logger.set_formatter(json_formatter)
    logger.info("用户登录", user_id=123, ip="192.168.1.1")
    
    print("\n=== JSON格式 + Syslog输出 ===")
    logger.set_writer(syslog_writer)
    logger.error("数据库连接失败", error_code=500, retry_count=3)

demo_multi_dimension_bridge()

7.3.2 动态桥接

python

from abc import ABC, abstractmethod
from typing import Dict, Type, Any, Callable, Optional
from dataclasses import dataclass
from enum import Enum

class ProviderType(Enum):
    AWS = "aws"
    AZURE = "azure"
    GCP = "gcp"
    ALIYUN = "aliyun"


@dataclass
class CloudConfig:
    """云配置"""
    region: str
    credentials: dict
    settings: dict


class CloudProvider(ABC):
    """云服务提供商接口"""
    
    @abstractmethod
    def create_instance(self, name: str, config: CloudConfig) -> str:
        pass
    
    @abstractmethod
    def delete_instance(self, instance_id: str) -> bool:
        pass
    
    @abstractmethod
    def list_instances(self) -> list:
        pass


class AWSProvider(CloudProvider):
    def create_instance(self, name: str, config: CloudConfig) -> str:
        return f"aws-{name}-{config.region}"
    
    def delete_instance(self, instance_id: str) -> bool:
        return True
    
    def list_instances(self) -> list:
        return [{"id": "aws-1", "type": "ec2"}]


class AzureProvider(CloudProvider):
    def create_instance(self, name: str, config: CloudConfig) -> str:
        return f"azure-{name}-{config.region}"
    
    def delete_instance(self, instance_id: str) -> bool:
        return True
    
    def list_instances(self) -> list:
        return [{"id": "azure-1", "type": "vm"}]


class GCPProvider(CloudProvider):
    def create_instance(self, name: str, config: CloudConfig) -> str:
        return f"gcp-{name}-{config.region}"
    
    def delete_instance(self, instance_id: str) -> bool:
        return True
    
    def list_instances(self) -> list:
        return [{"id": "gcp-1", "type": "compute"}]


class ProviderFactory:
    """提供商工厂"""
    
    _providers: Dict[ProviderType, Type[CloudProvider]] = {
        ProviderType.AWS: AWSProvider,
        ProviderType.AZURE: AzureProvider,
        ProviderType.GCP: GCPProvider,
    }
    
    @classmethod
    def create(cls, provider_type: ProviderType) -> CloudProvider:
        provider_class = cls._providers.get(provider_type)
        if not provider_class:
            raise ValueError(f"不支持的提供商: {provider_type}")
        return provider_class()
    
    @classmethod
    def register(cls, provider_type: ProviderType, provider_class: Type[CloudProvider]) -> None:
        cls._providers[provider_type] = provider_class


class CloudService:
    """
    云服务抽象：动态桥接不同提供商
    
    支持运行时切换提供商
    """
    
    def __init__(self, provider: CloudProvider, config: CloudConfig):
        self._provider = provider
        self._config = config
    
    def switch_provider(self, provider_type: ProviderType, config: CloudConfig = None) -> None:
        """运行时切换提供商"""
        self._provider = ProviderFactory.create(provider_type)
        if config:
            self._config = config
    
    def create_instance(self, name: str) -> str:
        return self._provider.create_instance(name, self._config)
    
    def delete_instance(self, instance_id: str) -> bool:
        return self._provider.delete_instance(instance_id)
    
    def list_instances(self) -> list:
        return self._provider.list_instances()


class MultiCloudManager:
    """
    多云管理器：管理多个云服务实例
    
    展示桥接模式在多云场景的应用
    """
    
    def __init__(self):
        self._services: Dict[str, CloudService] = {}
    
    def add_service(
        self, 
        name: str, 
        provider_type: ProviderType, 
        config: CloudConfig
    ) -> None:
        provider = ProviderFactory.create(provider_type)
        self._services[name] = CloudService(provider, config)
    
    def get_service(self, name: str) -> Optional[CloudService]:
        return self._services.get(name)
    
    def create_instance(self, service_name: str, instance_name: str) -> str:
        service = self.get_service(service_name)
        if service:
            return service.create_instance(instance_name)
        raise ValueError(f"服务 '{service_name}' 不存在")
    
    def list_all_instances(self) -> Dict[str, list]:
        return {
            name: service.list_instances()
            for name, service in self._services.items()
        }


def demo_dynamic_bridge():
    manager = MultiCloudManager()
    
    manager.add_service(
        "aws-prod",
        ProviderType.AWS,
        CloudConfig(region="us-east-1", credentials={}, settings={})
    )
    
    manager.add_service(
        "azure-dev",
        ProviderType.AZURE,
        CloudConfig(region="eastus", credentials={}, settings={})
    )
    
    print("=== 创建实例 ===")
    print(f"AWS: {manager.create_instance('aws-prod', 'web-server')}")
    print(f"Azure: {manager.create_instance('azure-dev', 'api-server')}")
    
    print("\n=== 所有实例 ===")
    for name, instances in manager.list_all_instances().items():
        print(f"{name}: {instances}")
    
    print("\n=== 运行时切换提供商 ===")
    service = manager.get_service('aws-prod')
    service.switch_provider(
        ProviderType.GCP, 
        CloudConfig(region="us-central1", credentials={}, settings={})
    )
    print(f"切换后创建: {service.create_instance('new-instance')}")


demo_dynamic_bridge()

7.4 企业级应用示例

7.4.1 报表生成系统

python

from abc import ABC, abstractmethod
from typing import Protocol, runtime_checkable, Dict, Any, List
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime

class ReportType(Enum):
    SALES = "sales"
    INVENTORY = "inventory"
    FINANCIAL = "financial"
    USER_ACTIVITY = "user_activity"


@dataclass
class ReportData:
    """报表数据"""
    title: str
    generated_at: datetime
    data: List[Dict[str, Any]]
    summary: Dict[str, Any] = field(default_factory=dict)
    metadata: Dict[str, str] = field(default_factory=dict)


@runtime_checkable
class DataExtractor(Protocol):
    """数据提取器：维度1"""
    
    def extract(self, report_type: ReportType, params: dict) -> ReportData:
        ...


@runtime_checkable
class ReportFormatter(Protocol):
    """报表格式化器：维度2"""
    
    def format(self, data: ReportData) -> str:
        ...


@runtime_checkable
class ReportExporter(Protocol):
    """报表导出器：维度3"""
    
    def export(self, content: str, filename: str) -> str:
        ...


class DatabaseExtractor:
    """数据库数据提取器"""
    
    def __init__(self, connection_string: str):
        self._connection_string = connection_string
    
    def extract(self, report_type: ReportType, params: dict) -> ReportData:
        print(f"从数据库提取 {report_type.value} 数据")
        
        mock_data = [
            {"id": 1, "name": "项目A", "value": 1000},
            {"id": 2, "name": "项目B", "value": 2000},
        ]
        
        return ReportData(
            title=f"{report_type.value.title()} Report",
            generated_at=datetime.now(),
            data=mock_data,
            summary={"total": 3000, "count": 2},
            metadata={"source": "database"}
        )


class APIExtractor:
    """API数据提取器"""
    
    def __init__(self, base_url: str):
        self._base_url = base_url
    
    def extract(self, report_type: ReportType, params: dict) -> ReportData:
        print(f"从API提取 {report_type.value} 数据")
        
        mock_data = [
            {"id": 1, "name": "API项目A", "value": 1500},
            {"id": 2, "name": "API项目B", "value": 2500},
        ]
        
        return ReportData(
            title=f"{report_type.value.title()} Report (API)",
            generated_at=datetime.now(),
            data=mock_data,
            summary={"total": 4000, "count": 2},
            metadata={"source": "api", "endpoint": f"/reports/{report_type.value}"}
        )


class HTMLFormatter:
    """HTML格式化器"""
    
    def format(self, data: ReportData) -> str:
        rows = "\n".join(
            f"<tr><td>{row['id']}</td><td>{row['name']}</td><td>{row['value']}</td></tr>"
            for row in data.data
        )
        
        return f"""
<!DOCTYPE html>
<html>
<head><title>{data.title}</title></head>
<body>
<h1>{data.title}</h1>
<p>生成时间: {data.generated_at}</p>
<table border="1">
<tr><th>ID</th><th>名称</th><th>值</th></tr>
{rows}
</table>
<h2>汇总</h2>
<p>总计: {data.summary.get('total', 0)}</p>
</body>
</html>
        """.strip()


class CSVFormatter:
    """CSV格式化器"""
    
    def format(self, data: ReportData) -> str:
        lines = [f"# {data.title}", f"# 生成时间: {data.generated_at}"]
        lines.append("id,name,value")
        
        for row in data.data:
            lines.append(f"{row['id']},{row['name']},{row['value']}")
        
        lines.append(f"# 总计: {data.summary.get('total', 0)}")
        
        return "\n".join(lines)


class JSONFormatter:
    """JSON格式化器"""
    
    def format(self, data: ReportData) -> str:
        import json
        
        report_dict = {
            "title": data.title,
            "generated_at": data.generated_at.isoformat(),
            "data": data.data,
            "summary": data.summary,
            "metadata": data.metadata
        }
        
        return json.dumps(report_dict, indent=2, ensure_ascii=False)


class FileExporter:
    """文件导出器"""
    
    def __init__(self, output_dir: str = "./reports"):
        self._output_dir = output_dir
    
    def export(self, content: str, filename: str) -> str:
        import os
        os.makedirs(self._output_dir, exist_ok=True)
        
        filepath = os.path.join(self._output_dir, filename)
        with open(filepath, 'w', encoding='utf-8') as f:
            f.write(content)
        
        return filepath


class EmailExporter:
    """邮件导出器"""
    
    def __init__(self, smtp_config: dict):
        self._smtp_config = smtp_config
    
    def export(self, content: str, filename: str) -> str:
        print(f"发送邮件: {filename}")
        print(f"  SMTP: {self._smtp_config.get('host', 'localhost')}")
        return f"email:{filename}"


class ReportGenerator:
    """
    报表生成器：桥接三个维度
    
    维度1：数据源（Database, API, File...）
    维度2：格式（HTML, CSV, JSON, PDF...）
    维度3：导出方式（File, Email, S3...）
    """
    
    def __init__(
        self,
        extractor: DataExtractor,
        formatter: ReportFormatter,
        exporter: ReportExporter
    ):
        self._extractor = extractor
        self._formatter = formatter
        self._exporter = exporter
    
    def set_extractor(self, extractor: DataExtractor) -> None:
        self._extractor = extractor
    
    def set_formatter(self, formatter: ReportFormatter) -> None:
        self._formatter = formatter
    
    def set_exporter(self, exporter: ReportExporter) -> None:
        self._exporter = exporter
    
    def generate(
        self, 
        report_type: ReportType, 
        params: dict = None,
        filename: str = None
    ) -> str:
        """生成报表"""
        params = params or {}
        
        data = self._extractor.extract(report_type, params)
        
        formatted = self._formatter.format(data)
        
        if not filename:
            ext = self._get_extension()
            filename = f"{report_type.value}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.{ext}"
        
        return self._exporter.export(formatted, filename)
    
    def _get_extension(self) -> str:
        if isinstance(self._formatter, HTMLFormatter):
            return "html"
        elif isinstance(self._formatter, CSVFormatter):
            return "csv"
        elif isinstance(self._formatter, JSONFormatter):
            return "json"
        return "txt"


def demo_report_system():
    db_extractor = DatabaseExtractor("postgresql://localhost/app")
    api_extractor = APIExtractor("https://api.example.com")
    
    html_formatter = HTMLFormatter()
    csv_formatter = CSVFormatter()
    json_formatter = JSONFormatter()
    
    file_exporter = FileExporter("./reports")
    email_exporter = EmailExporter({"host": "smtp.example.com"})
    
    generator = ReportGenerator(db_extractor, html_formatter, file_exporter)
    
    print("=== HTML报表 ===")
    result = generator.generate(ReportType.SALES)
    print(f"输出: {result}")
    
    print("\n=== 切换到CSV格式 ===")
    generator.set_formatter(csv_formatter)
    result = generator.generate(ReportType.INVENTORY)
    print(f"输出: {result}")
    
    print("\n=== 切换到API数据源 ===")
    generator.set_extractor(api_extractor)
    generator.set_formatter(json_formatter)
    result = generator.generate(ReportType.FINANCIAL)
    print(f"输出: {result}")

demo_report_system()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267

7.4.2 支付处理系统

python

from abc import ABC, abstractmethod
from typing import Protocol, runtime_checkable, Dict, Any, Optional
from dataclasses import dataclass
from decimal import Decimal
from enum import Enum

class PaymentMethod(Enum):
    CREDIT_CARD = "credit_card"
    DEBIT_CARD = "debit_card"
    BANK_TRANSFER = "bank_transfer"
    DIGITAL_WALLET = "digital_wallet"


class PaymentStatus(Enum):
    PENDING = "pending"
    PROCESSING = "processing"
    COMPLETED = "completed"
    FAILED = "failed"
    REFUNDED = "refunded"


@dataclass
class PaymentInfo:
    """支付信息"""
    amount: Decimal
    currency: str
    method: PaymentMethod
    customer_id: str
    description: str = ""
    metadata: Dict[str, Any] = None


@dataclass
class PaymentResult:
    """支付结果"""
    success: bool
    status: PaymentStatus
    transaction_id: str
    message: str
    amount: Decimal
    currency: str
    fee: Decimal = Decimal("0")
    metadata: Dict[str, Any] = None


@runtime_checkable
class PaymentGateway(Protocol):
    """支付网关：维度1"""
    
    def process(self, payment: PaymentInfo) -> PaymentResult:
        ...
    
    def refund(self, transaction_id: str, amount: Decimal) -> PaymentResult:
        ...
    
    def get_status(self, transaction_id: str) -> PaymentStatus:
        ...


@runtime_checkable
class FraudDetector(Protocol):
    """欺诈检测器：维度2"""
    
    def analyze(self, payment: PaymentInfo) -> tuple[bool, str]:
        """返回 (是否安全, 原因)"""
        ...


@runtime_checkable
class PaymentNotifier(Protocol):
    """支付通知器：维度3"""
    
    def notify(self, result: PaymentResult, customer_id: str) -> None:
        ...


class StripeGateway:
    """Stripe支付网关"""
    
    def __init__(self, api_key: str):
        self._api_key = api_key
    
    def process(self, payment: PaymentInfo) -> PaymentResult:
        print(f"Stripe: 处理 {payment.amount} {payment.currency}")
        return PaymentResult(
            success=True,
            status=PaymentStatus.COMPLETED,
            transaction_id=f"ch_stripe_{hash(payment.customer_id) % 10000}",
            message="支付成功",
            amount=payment.amount,
            currency=payment.currency,
            fee=payment.amount * Decimal("0.029")
        )
    
    def refund(self, transaction_id: str, amount: Decimal) -> PaymentResult:
        print(f"Stripe: 退款 {amount}")
        return PaymentResult(
            success=True,
            status=PaymentStatus.REFUNDED,
            transaction_id=f"re_stripe_{hash(transaction_id) % 10000}",
            message="退款成功",
            amount=amount,
            currency="USD"
        )
    
    def get_status(self, transaction_id: str) -> PaymentStatus:
        return PaymentStatus.COMPLETED


class PayPalGateway:
    """PayPal支付网关"""
    
    def __init__(self, client_id: str, secret: str):
        self._client_id = client_id
        self._secret = secret
    
    def process(self, payment: PaymentInfo) -> PaymentResult:
        print(f"PayPal: 处理 {payment.amount} {payment.currency}")
        return PaymentResult(
            success=True,
            status=PaymentStatus.COMPLETED,
            transaction_id=f"PAY_{hash(payment.customer_id) % 10000}",
            message="支付成功",
            amount=payment.amount,
            currency=payment.currency,
            fee=payment.amount * Decimal("0.034")
        )
    
    def refund(self, transaction_id: str, amount: Decimal) -> PaymentResult:
        print(f"PayPal: 退款 {amount}")
        return PaymentResult(
            success=True,
            status=PaymentStatus.REFUNDED,
            transaction_id=f"REFUND_{hash(transaction_id) % 10000}",
            message="退款成功",
            amount=amount,
            currency="USD"
        )
    
    def get_status(self, transaction_id: str) -> PaymentStatus:
        return PaymentStatus.COMPLETED


class RuleBasedFraudDetector:
    """基于规则的欺诈检测器"""
    
    def __init__(self, max_amount: Decimal = Decimal("10000")):
        self._max_amount = max_amount
        self._suspicious_customers: set = set()
    
    def analyze(self, payment: PaymentInfo) -> tuple[bool, str]:
        if payment.amount > self._max_amount:
            return False, f"金额超过限制: {payment.amount} > {self._max_amount}"
        
        if payment.customer_id in self._suspicious_customers:
            return False, "可疑客户"
        
        if payment.method == PaymentMethod.BANK_TRANSFER and payment.amount > Decimal("5000"):
            return False, "大额转账需要人工审核"
        
        return True, "通过"


class MLFraudDetector:
    """机器学习欺诈检测器"""
    
    def __init__(self, model_endpoint: str):
        self._model_endpoint = model_endpoint
    
    def analyze(self, payment: PaymentInfo) -> tuple[bool, str]:
        score = hash(f"{payment.customer_id}{payment.amount}") % 100 / 100
        
        if score > 0.8:
            return False, f"高风险交易 (分数: {score:.2f})"
        
        return True, f"低风险交易 (分数: {score:.2f})"


class EmailNotifier:
    """邮件通知器"""
    
    def notify(self, result: PaymentResult, customer_id: str) -> None:
        print(f"发送邮件通知给 {customer_id}: {result.message}")


class SMSNotifier:
    """短信通知器"""
    
    def notify(self, result: PaymentResult, customer_id: str) -> None:
        print(f"发送短信通知给 {customer_id}: {result.status.value}")


class PaymentProcessor:
    """
    支付处理器：桥接三个维度
    
    维度1：支付网关（Stripe, PayPal, Alipay...）
    维度2：欺诈检测（规则引擎, ML模型...）
    维度3：通知方式（邮件, 短信, Webhook...）
    """
    
    def __init__(
        self,
        gateway: PaymentGateway,
        fraud_detector: FraudDetector,
        notifier: PaymentNotifier
    ):
        self._gateway = gateway
        self._fraud_detector = fraud_detector
        self._notifier = notifier
    
    def set_gateway(self, gateway: PaymentGateway) -> None:
        self._gateway = gateway
    
    def set_fraud_detector(self, detector: FraudDetector) -> None:
        self._fraud_detector = detector
    
    def set_notifier(self, notifier: PaymentNotifier) -> None:
        self._notifier = notifier
    
    def process_payment(self, payment: PaymentInfo) -> PaymentResult:
        """处理支付"""
        is_safe, reason = self._fraud_detector.analyze(payment)
        
        if not is_safe:
            result = PaymentResult(
                success=False,
                status=PaymentStatus.FAILED,
                transaction_id="",
                message=f"欺诈检测失败: {reason}",
                amount=payment.amount,
                currency=payment.currency
            )
            self._notifier.notify(result, payment.customer_id)
            return result
        
        result = self._gateway.process(payment)
        self._notifier.notify(result, payment.customer_id)
        
        return result
    
    def refund(self, transaction_id: str, amount: Decimal) -> PaymentResult:
        """退款"""
        result = self._gateway.refund(transaction_id, amount)
        return result


def demo_payment_system():
    stripe = StripeGateway("sk_test_123")
    paypal = PayPalGateway("client_id", "secret")
    
    rule_detector = RuleBasedFraudDetector(max_amount=Decimal("5000"))
    ml_detector = MLFraudDetector("https://ml.example.com/fraud")
    
    email_notifier = EmailNotifier()
    sms_notifier = SMSNotifier()
    
    processor = PaymentProcessor(stripe, rule_detector, email_notifier)
    
    print("=== 正常支付 ===")
    payment = PaymentInfo(
        amount=Decimal("100.00"),
        currency="USD",
        method=PaymentMethod.CREDIT_CARD,
        customer_id="customer_001"
    )
    result = processor.process_payment(payment)
    print(f"结果: {result.status.value}")
    
    print("\n=== 大额支付（触发规则） ===")
    large_payment = PaymentInfo(
        amount=Decimal("6000.00"),
        currency="USD",
        method=PaymentMethod.CREDIT_CARD,
        customer_id="customer_002"
    )
    result = processor.process_payment(large_payment)
    print(f"结果: {result.message}")
    
    print("\n=== 切换到ML检测 + 短信通知 ===")
    processor.set_fraud_detector(ml_detector)
    processor.set_notifier(sms_notifier)
    result = processor.process_payment(payment)
    print(f"结果: {result.status.value}")

demo_payment_system()

7.5 与其他模式比较

7.5.1 桥接 vs 适配器 vs 策略

python

from abc import ABC, abstractmethod
from typing import Protocol

class Implementation(Protocol):
    def execute(self) -> str: ...

class ConcreteImplA:
    def execute(self) -> str:
        return "实现A"

class ConcreteImplB:
    def execute(self) -> str:
        return "实现B"

class BridgeAbstraction:
    """
    桥接模式示例
    
    目的：分离抽象与实现，使两者独立变化
    特点：抽象持有实现引用，可运行时切换
    """
    
    def __init__(self, impl: Implementation):
        self._impl = impl
    
    def operation(self) -> str:
        return f"桥接: {self._impl.execute()}"


class LegacySystem:
    def old_execute(self) -> str:
        return "遗留系统"

class AdapterTarget:
    """
    适配器模式示例
    
    目的：使不兼容接口能够一起工作
    特点：转换接口，不改变抽象
    """
    
    def __init__(self, legacy: LegacySystem):
        self._legacy = legacy
    
    def execute(self) -> str:
        return f"适配: {self._legacy.old_execute()}"


class StrategyContext:
    """
    策略模式示例
    
    目的：封装可互换的算法
    特点：客户端选择策略，行为可替换
    """
    
    def __init__(self, strategy: Implementation):
        self._strategy = strategy
    
    def execute_strategy(self) -> str:
        return f"策略: {self._strategy.execute()}"


def demo_pattern_comparison():
    impl_a = ConcreteImplA()
    impl_b = ConcreteImplB()
    
    print("=== 桥接模式 ===")
    bridge = BridgeAbstraction(impl_a)
    print(bridge.operation())
    
    print("\n=== 适配器模式 ===")
    legacy = LegacySystem()
    adapter = AdapterTarget(legacy)
    print(adapter.execute())
    
    print("\n=== 策略模式 ===")
    context = StrategyContext(impl_b)
    print(context.execute_strategy())

demo_pattern_comparison()

7.5.2 模式选择指南

场景	推荐模式	原因
多维度独立变化	桥接模式	避免类爆炸，支持运行时切换
接口不兼容	适配器模式	转换接口，不改变结构
算法可互换	策略模式	封装行为，客户端选择
需要隐藏实现细节	桥接模式	抽象与实现解耦
需要简化复杂接口	外观模式	提供统一入口

7.6 反模式与最佳实践

7.6.1 常见反模式

python

from typing import Any

class BridgeAntiPatterns:
    """桥接模式反模式示例"""
    
    @staticmethod
    def anti_pattern_1_no_abstraction():
        """
        反模式：没有真正的抽象
        
        问题：抽象层只是简单委托，没有添加价值
        """
        class Implementation:
            def do_something(self) -> str:
                return "done"
        
        class UselessAbstraction:
            def __init__(self, impl: Implementation):
                self._impl = impl
            
            def do_something(self) -> str:
                return self._impl.do_something()
        
        print("反模式1：无意义的抽象层")
        print("  问题：抽象层没有添加任何价值")
    
    @staticmethod
    def anti_pattern_2_tight_coupling():
        """
        反模式：抽象与实现紧耦合
        
        问题：抽象直接依赖具体实现类
        """
        class ConcreteImpl:
            def execute(self) -> str:
                return "concrete"
        
        class BadAbstraction:
            def __init__(self):
                self._impl = ConcreteImpl()
            
            def operation(self) -> str:
                return self._impl.execute()
        
        print("\n反模式2：紧耦合")
        print("  问题：抽象直接实例化具体实现")
    
    @staticmethod
    def anti_pattern_3_over_engineering():
        """
        反模式：过度工程
        
        问题：简单场景使用复杂模式
        """
        print("\n反模式3：过度工程")
        print("  问题：单维度变化不需要桥接模式")


def demo_anti_patterns():
    BridgeAntiPatterns.anti_pattern_1_no_abstraction()
    BridgeAntiPatterns.anti_pattern_2_tight_coupling()
    BridgeAntiPatterns.anti_pattern_3_over_engineering()

demo_anti_patterns()

7.6.2 最佳实践清单

python

BEST_PRACTICE_CHECKLIST = """
桥接模式最佳实践清单
====================

□ 抽象设计
  ├─ 抽象层应定义高层操作
  ├─ 抽象可以组合多个实现操作
  └─ 提供运行时切换实现的能力

□ 实现设计
  ├─ 实现接口不与抽象接口对应
  ├─ 实现提供原子操作
  └─ 实现可以有多种变体

□ 耦合管理
  ├─ 抽象仅依赖实现接口
  ├─ 实现不知道抽象的存在
  └─ 使用依赖注入设置实现

□ 扩展策略
  ├─ 新增抽象不影响现有实现
  ├─ 新增实现不影响现有抽象
  └─ 两者可以独立扩展

□ 性能考量
  ├─ 避免过度委托
  ├─ 考虑实现的缓存
  └─ 注意对象创建开销
"""

7.7 决策指南

7.7.1 使用决策树

┌─────────────────────────────────────────────────────────────┐
│                  桥接模式使用决策树                          │
└─────────────────────────────────────────────────────────────┘

                    是否存在多个变化维度？
                           │
                           ▼
              ┌────────────────────────┐
              │ 维度是否独立变化？      │
              └────────────┬───────────┘
                     │           │
                    是           否
                     │           │
                     ▼           ▼
          ┌─────────────┐  ┌─────────────────┐
          │ 是否需要    │  │ 考虑其他模式    │
          │ 运行时切换？│  │ (策略/适配器)   │
          └─────────────┘  └─────────────────┘
                     │
              ┌──────┴──────┐
              是            否
              │             │
              ▼             ▼
      ┌─────────────┐ ┌─────────────┐
      │ 使用桥接模式 │ │ 考虑继承    │
      └─────────────┘ └─────────────┘
              │
              ▼
      ┌────────────────────────┐
      │ 设计抽象层和实现层      │
      └────────────┬───────────┘
                     │
                     ▼
      ┌────────────────────────┐
      │ 使用组合连接两者        │
      └─────────────────────────┘

7.7.2 快速参考卡

python

BRIDGE_QUICK_REFERENCE = """
桥接模式快速参考
================

┌─────────────────────────────────────────────────────────────┐
│ 核心概念                                                    │
├─────────────────────────────────────────────────────────────┤
│ 将抽象与实现分离，使两者可以独立变化                        │
└─────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────┐
│ 实现结构                                                    │
├─────────────────────────────────────────────────────────────┤
│ class Implementation(ABC):                                 │
│     @abstractmethod                                        │
│     def operation_impl(self): ...                          │
│                                                             │
│ class Abstraction:                                         │
│     def __init__(self, impl: Implementation):              │
│         self._impl = impl                                  │
│     def operation(self):                                   │
│         return self._impl.operation_impl()                 │
└─────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────┐
│ 适用场景                                                    │
├─────────────────────────────────────────────────────────────┤
│ ✓ 多个独立变化的维度                                        │
│ ✓ 需要运行时切换实现                                        │
│ ✓ 避免类数量爆炸                                            │
│ ✓ 跨平台应用                                                │
│ ✓ 需要隐藏实现细节                                          │
└─────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────┐
│ 注意事项                                                    │
├─────────────────────────────────────────────────────────────┤
│ ⚠ 确保抽象和实现真正独立                                    │
│ ⚠ 抽象层应提供有意义的操作                                  │
│ ⚠ 使用依赖注入设置实现                                      │
│ ⚠ 简单场景不要过度设计                                      │
└─────────────────────────────────────────────────────────────┘
"""

7.8 小结

桥接模式通过组合将抽象与实现分离，是处理多维度变化的核心方案。本章从形式化定义、历史渊源、实现机制、高级技术、企业级应用等多个维度进行了深入分析。

核心要点：

形式化定义：桥接模式将笛卡尔积 $\mathcal{A} \times \mathcal{I}$ 解耦为正交组合 $\mathcal{A} \oplus \mathcal{I}$，显著降低类数量。
设计原则：体现"组合优于继承"原则，通过组合实现抽象与实现的解耦。
多维度桥接：可以桥接两个或更多维度，如日志系统的格式化器和写入器。
Protocol支持：Python 3.8+ 的Protocol提供结构化类型支持，简化桥接模式实现。
企业级应用：报表生成系统、支付处理系统等展示了模式在复杂场景中的价值。
最佳实践：确保抽象和实现真正独立、抽象层提供有意义的操作、使用依赖注入。

桥接模式特别适用于存在多个独立变化维度的场景，是避免类爆炸、实现灵活架构的重要工具。

第7章 桥接模式 ​

学习目标 ​

7.1 模式定义 ​

7.1.1 形式化定义 ​

7.1.2 历史背景与学术渊源 ​

7.1.3 模式动机 ​

7.1.4 UML结构模型 ​

7.2 Python实现机制 ​

7.2.1 标准实现 ​

7.2.2 基于Protocol的现代实现 ​

7.2.3 泛型桥接模式 ​

7.3 高级应用技术 ​

7.3.1 多维度桥接 ​

7.3.2 动态桥接 ​

7.4 企业级应用示例 ​

7.4.1 报表生成系统 ​

7.4.2 支付处理系统 ​

7.5 与其他模式比较 ​

7.5.1 桥接 vs 适配器 vs 策略 ​

7.5.2 模式选择指南 ​

7.6 反模式与最佳实践 ​

7.6.1 常见反模式 ​

7.6.2 最佳实践清单 ​

7.7 决策指南 ​

7.7.1 使用决策树 ​

7.7.2 快速参考卡 ​

7.8 小结 ​

第7章桥接模式

学习目标

7.1 模式定义

7.1.1 形式化定义

7.1.2 历史背景与学术渊源

7.1.3 模式动机

7.1.4 UML结构模型

7.2 Python实现机制

7.2.1 标准实现

7.2.2 基于Protocol的现代实现

7.2.3 泛型桥接模式

7.3 高级应用技术

7.3.1 多维度桥接

7.3.2 动态桥接

7.4 企业级应用示例

7.4.1 报表生成系统

7.4.2 支付处理系统

7.5 与其他模式比较

7.5.1 桥接 vs 适配器 vs 策略

7.5.2 模式选择指南

7.6 反模式与最佳实践

7.6.1 常见反模式

7.6.2 最佳实践清单

7.7 决策指南

7.7.1 使用决策树

7.7.2 快速参考卡

7.8 小结