在行为型模式中，中介者模式是解决 “多对象间网状耦合” 问题的核心模式。它就像 “机场调度中心”—— 多个航班（对象）无需直接沟通起飞、降落时间，只需通过调度中心（中介者）协调，避免航班间的冲突与混乱。在软件开发中，当系统包含多个相互依赖、频繁交互的对象（如 UI 组件、服务模块），且对象间直接耦合导致代码难以维护时，中介者模式可通过 “中介者对象” 统一管理对象间的交互逻辑，将 “网状耦合” 转化为 “星型结构”，降低系统复杂度。本章将从中介者模式的核心概念出发，详细讲解其实现原理、Python 代码实现、实际应用场景及与其他模式的差异。

13.1 中介者模式的定义与核心问题

13.1.1 定义

中介者模式（Mediator Pattern）的官方定义为：用一个中介对象来封装一系列的对象交互。中介者使各对象不需要显式地相互引用，从而使其耦合松散，而且可以独立地改变它们之间的交互。

简单来说，中介者模式就像 “房屋中介”—— 房东（对象 A）和租客（对象 B）无需直接沟通房屋租赁细节（价格、看房时间、合同签订），只需通过中介（中介者）传递信息，中介负责协调双方需求，避免直接耦合。例如，电商购物车系统中，“商品选择”“库存检查”“价格计算”“优惠券抵扣” 四个模块需频繁交互（选择商品后需检查库存、更新价格、计算优惠），通过中介者模式，各模块只需向中介者发送事件，由中介者统一协调其他模块响应，无需模块间直接调用。

13.1.2 核心问题：多对象交互的网状耦合困境

在未使用中介者模式时，多个对象直接交互会形成 “网状耦合” 结构，面临以下问题：

1. 耦合度极高：每个对象需持有其他多个对象的引用（如商品模块需引用库存、价格、优惠券模块），若某一对象修改（如库存模块接口变更），所有依赖它的对象都需同步修改；

2. 代码维护困难：对象间的交互逻辑分散在各个对象中（如商品选择后，商品模块需调用库存检查、价格计算），问题排查时需追踪多个对象的调用链路，定位难度大；

3. 扩展性差：新增对象时（如新增 “运费计算” 模块），需修改所有相关对象的代码，使其引用新模块（如商品模块需调用运费计算），违背开闭原则；

4. 复用性低：对象因强依赖其他对象，难以单独复用（如将库存模块复用至其他系统，需同时迁移商品、价格等依赖模块）；

5. 交互逻辑混乱：多个对象同时交互时（如商品选择、优惠券使用、库存不足同时发生），容易出现逻辑冲突（如优惠券已抵扣但库存不足，需回滚优惠），难以协调。

中介者模式通过 “中介者对象” 解决上述问题：所有对象将交互逻辑委托给中介者，对象间仅通过中介者传递信息（如发送事件、请求数据），无需直接引用，将 “网状耦合” 转化为 “对象 - 中介者 - 对象” 的星型结构，实现 “解耦、集中管理、灵活扩展” 的交互逻辑。

13.1.3 生活中的中介者模式案例

• 机场调度中心：航班（对象）→ 调度中心（中介者）→ 其他航班 / 塔台（对象），调度中心协调航班起飞、降落时间，避免冲突；

• 房屋中介：房东（对象 A）/ 租客（对象 B）→ 中介（中介者），中介传递需求、协调看房、签订合同；

• 聊天室：用户（对象）→ 聊天室服务器（中介者）→ 其他用户（对象），用户发送消息至服务器，服务器转发给其他用户，无需用户间直接连接；

• 交通指挥中心：车辆（对象）→ 指挥中心（中介者）→ 红绿灯 / 其他车辆（对象），指挥中心统一调度交通，避免拥堵。

13.2 中介者模式的核心角色

中介者模式包含 4 个核心角色，通过 “中介者统一协调” 实现对象间的解耦，各角色职责明确且协作紧密：

角色间的协作流程

1. 客户端创建 “具体中介者” 对象；

2. 客户端创建多个 “具体同事类” 对象，调用set_mediator()方法关联中介者；

3. 客户端将同事对象注册到中介者（中介者维护同事对象集合，便于后续调用）；

4. 某一同事对象状态变化（如用户选择商品），调用send_event()向中介者发送事件（如 “商品选择事件”，携带商品 ID、数量）；

5. 中介者的handle_event()方法接收事件，根据事件类型和参数，调用其他同事对象的业务方法（如调用库存模块检查库存、价格模块计算总价）；

6. 其他同事对象执行业务逻辑后，若需进一步交互，可通过send_event()向中介者发送新事件（如库存不足，发送 “库存不足事件”）；

7. 中介者继续协调其他对象响应新事件（如库存不足时，调用 UI 模块显示提示信息），直至交互结束。

13.3 中介者模式的 Python 实现（基础版）

以 “电商购物车” 为例，实现包含 “商品选择”“库存检查”“价格计算”“优惠券抵扣” 4 个同事模块的中介者，各模块通过中介者协调交互，避免直接耦合。

13.3.1 步骤 1：定义抽象中介者（Mediator）

创建抽象中介者类，定义 “注册同事对象” 和 “处理事件” 的抽象方法，规范中介者与同事对象的交互接口。

from abc import ABC, abstractmethod

from typing import Dict, Any, Optional

# 抽象中介者：购物车中介者（CartMediator）

class CartMediator(ABC):

def __init__(self):

self.colleagues: Dict[str, "CartColleague"] = {} # 存储同事对象（key：模块名，value：同事对象）

@abstractmethod

def register_colleague(self, colleague: "CartColleague") -> None:

"""抽象方法：注册同事对象到中介者"""

pass

@abstractmethod

def handle_event(self, colleague_name: str, event_type: str, data: Optional[Dict] = None) -> Any:

"""

抽象方法：处理同事对象发送的事件

:param colleague_name: 发送事件的同事对象名称

:param event_type: 事件类型（如"select_product"“check_stock”）

:param data: 事件数据（如商品ID、数量）

:return: 事件处理结果（可选）

"""

pass

13.3.2 步骤 2：定义抽象同事类（Colleague）

创建抽象同事类，定义 “关联中介者” 和 “发送事件” 的抽象方法，规范同事对象与中介者的交互方式。

# 抽象同事类：购物车同事模块（CartColleague）

class CartColleague(ABC):

def __init__(self, name: str):

self.name = name # 同事对象名称（如"product"“stock”“price”“coupon”）

self.mediator: Optional[CartMediator] = None # 关联的中介者

@abstractmethod

def set_mediator(self, mediator: CartMediator) -> None:

"""抽象方法：关联中介者"""

pass

@abstractmethod

def send_event(self, event_type: str, data: Optional[Dict] = None) -> Any:

"""抽象方法：向中介者发送事件"""

pass

@abstractmethod

def handle_event(self, event_type: str, data: Optional[Dict] = None) -> Any:

"""抽象方法：处理中介者转发的事件（如其他模块触发的事件）"""

pass

13.3.3 步骤 3：实现具体同事类（Concrete Colleague）

分别实现 “商品选择”“库存检查”“价格计算”“优惠券抵扣” 4 个同事模块，每个模块实现自身的业务逻辑，状态变化时通过中介者发送事件。

同事类 1：商品选择模块（ProductColleague）

# 具体同事类1：商品选择模块（ProductColleague）

class ProductColleague(CartColleague):

def __init__(self):

super().__init__(name="product")

self.selected_products: Dict[str, int] = {} # 已选商品（key：商品ID，value：数量）

def set_mediator(self, mediator: CartMediator) -> None:

"""关联中介者"""

self.mediator = mediator

def send_event(self, event_type: str, data: Optional[Dict] = None) -> Any:

"""向中介者发送事件（如“选择商品”“取消选择”）"""

if self.mediator:

return self.mediator.handle_event(self.name, event_type, data)

else:

raise ValueError("未关联中介者，无法发送事件")

def handle_event(self, event_type: str, data: Optional[Dict] = None) -> Any:

"""处理中介者转发的事件（如其他模块请求获取已选商品）"""

if event_type == "get_selected_products":

# 价格模块或优惠券模块请求已选商品

print(f"[{self.name}] 处理事件：{event_type}，返回已选商品：{self.selected_products}")

return self.selected_products

else:

print(f"[{self.name}] 不支持的事件类型：{event_type}")

return None

def select_product(self, product_id: str, quantity: int) -> None:

"""选择商品（自身业务逻辑），发送“商品选择”事件给中介者"""

if quantity <= 0:

print(f"[{self.name}] 商品{product_id}选择数量无效：{quantity}")

return

# 更新已选商品

self.selected_products[product_id] = self.selected_products.get(product_id, 0) + quantity

print(f"[{self.name}] 已选择商品：{product_id}，数量：{self.selected_products[product_id]}")

# 向中介者发送“商品选择”事件，触发其他模块响应

self.send_event(

event_type="select_product",

data={"product_id": product_id, "quantity": quantity}

)

def unselect_product(self, product_id: str) -> None:

"""取消选择商品，发送“取消选择”事件给中介者"""

if product_id in self.selected_products:

del self.selected_products[product_id]

print(f"[{self.name}] 已取消选择商品：{product_id}")

# 向中介者发送“取消选择”事件

self.send_event(

event_type="unselect_product",

data={"product_id": product_id}

)

else:

print(f"[{self.name}] 未选择商品：{product_id}，无法取消")

同事类 2：库存检查模块（StockColleague）

# 具体同事类2：库存检查模块（StockColleague）

class StockColleague(CartColleague):

def __init__(self):

super().__init__(name="stock")

# 模拟库存数据（key：商品ID，value：可用库存）

self.stock_data: Dict[str, int] = {

"prod_001": 100, # 商品1库存100

"prod_002": 5, # 商品2库存5

"prod_003": 0 # 商品3库存0

}

def set_mediator(self, mediator: CartMediator) -> None:

self.mediator = mediator

def send_event(self, event_type: str, data: Optional[Dict] = None) -> Any:

if self.mediator:

return self.mediator.handle_event(self.name, event_type, data)

else:

raise ValueError("未关联中介者，无法发送事件")

def handle_event(self, event_type: str, data: Optional[Dict] = None) -> Any:

"""处理事件：检查库存、扣减库存"""

if event_type == "check_stock" and data:

product_id = data.get("product_id")

quantity = data.get("quantity", 0)

if not product_id:

return {"success": False, "message": "商品ID不能为空"}

# 检查库存

available_stock = self.stock_data.get(product_id, 0)

if available_stock >= quantity:

result = {"success": True, "available_stock": available_stock, "message": "库存充足"}

else:

result = {"success": False, "available_stock": available_stock, "message": "库存不足"}

print(f"[{self.name}] 处理事件：{event_type}，商品{product_id}，需求{quantity}，结果：{result['message']}")

return result

elif event_type == "deduct_stock" and data:

# 扣减库存（仅下单时触发）

product_id = data.get("product_id")

quantity = data.get("quantity", 0)

if product_id in self.stock_data and self.stock_data[product_id] >= quantity:

self.stock_data[product_id] -= quantity

print(f"[{self.name}] 处理事件：{event_type}，商品{product_id}，扣减{quantity}，剩余库存：{self.stock_data[product_id]}")

return {"success": True, "message": "库存扣减成功"}

else:

print(f"[{self.name}] 处理事件：{event_type}，商品{product_id}库存不足，无法扣减")

return {"success": False, "message": "库存不足"}

else:

print(f"[{self.name}] 不支持的事件类型：{event_type}")

return None

同事类 3：价格计算模块（PriceColleague）

# 具体同事类3：价格计算模块（PriceColleague）

class PriceColleague(CartColleague):

def __init__(self):

super().__init__(name="price")

# 商品单价（key：商品ID，value：单价（元））

self.product_prices: Dict[str, float] = {

"prod_001": 99.9,

"prod_002": 199.8,

"prod_003": 299.7

}

self.total_price: float = 0.0 # 总价（未抵扣优惠券）

def set_mediator(self, mediator: CartMediator) -> None:

self.mediator = mediator

def send_event(self, event_type: str, data: Optional[Dict] = None) -> Any:

if self.mediator:

return self.mediator.handle_event(self.name, event_type, data)

else:

raise ValueError("未关联中介者，无法发送事件")

</doubaocanvas>