Python 中介者模式

Python 中的中介者模式（Mediator Pattern）

中介者模式是一种行为型设计模式，其核心目的是：
用一个中介者对象来封装一系列对象之间的交互。中介者使各对象不需要显式地相互引用，从而降低它们之间的耦合度，而且可以独立地改变它们的交互行为。

形象比喻：就像机场的控制塔——每架飞机不需要直接和其他飞机通信，只需和控制塔（中介者）联系，由控制塔协调起飞、降落、航线，避免飞机之间直接“聊天”导致混乱。

中介者模式的优点

• 降低耦合：对象之间不再直接持有引用（松耦合）
• 集中控制交互逻辑：复杂通信规则集中在中介者中，便于维护
• 简化对象协议：每个对象只关注自身行为和与中介者的通信
• 易于扩展：新增同事对象只需注册到中介者
• 支持广播和一对一通信

典型应用场景

• 聊天室（所有用户通过服务器发送消息）
• GUI 组件交互（按钮、文本框、下拉框之间的联动）
• 工作流系统（各部门通过流程引擎协调）
• 游戏中单位间通信（玩家、NPC、怪物通过游戏管理器交互）
• 微服务事件总线

Python 实现示例：聊天室系统

from abc import ABC, abstractmethod
from typing import List

# 中介者接口（Mediator）
class ChatRoom(ABC):
    @abstractmethod
    def send_message(self, sender: 'User', message: str):
        pass

    @abstractmethod
    def register(self, user: 'User'):
        pass

# 具体中介者：聊天室
class ConcreteChatRoom(ChatRoom):
    def __init__(self):
        self.users: List['User'] = []

    def register(self, user: 'User'):
        if user not in self.users:
            self.users.append(user)
            user.chat_room = self
            print(f"{user.name} 加入聊天室")

    def send_message(self, sender: 'User', message: str):
        # 中介者负责广播（排除发送者自己）
        print(f"[{sender.name}]: {message}")
        for user in self.users:
            if user != sender:  # 不发给自己
                user.receive(message)

# 同事类（Colleague）
class User:
    def __init__(self, name: str):
        self.name = name
        self.chat_room: ChatRoom = None  # 持有中介者引用

    def send(self, message: str):
        if self.chat_room is None:
            print(f"{self.name} 尚未加入任何聊天室")
            return
        print(f"{self.name} 发送消息...")
        self.chat_room.send_message(self, message)

    def receive(self, message: str):
        print(f"  → {self.name} 收到: {message}")

# 客户端使用
if __name__ == "__main__":
    # 创建中介者
    chat_room = ConcreteChatRoom()

    # 创建用户（同事）
    alice = User("Alice")
    bob = User("Bob")
    charlie = User("Charlie")

    # 用户注册到中介者
    chat_room.register(alice)
    chat_room.register(bob)
    chat_room.register(charlie)

    print("\n--- 开始聊天 ---")
    alice.send("大家好！")
    bob.send("Hi Alice!")
    charlie.send("今天天气真好～")
    alice.send("确实！要去散步吗？")

输出：

Alice 加入聊天室
Bob 加入聊天室
Charlie 加入聊天室

--- 开始聊天 ---
Alice 发送消息...
[Alice]: 大家好！
  → Bob 收到: 大家好！
  → Charlie 收到: 大家好！
Bob 发送消息...
[Bob]: Hi Alice!
  → Alice 收到: Hi Alice!
  → Charlie 收到: Hi Alice!
Charlie 发送消息...
[Charlie]: 今天天气真好～
  → Alice 收到: 今天天气真好～
  → Bob 收到: 今天天气真好～
Alice 发送消息...
[Alice]: 确实！要去散步吗？
  → Bob 收到: 确实！要去散步吗？
  → Charlie 收到: 确实！要去散步吗？

用户之间没有直接引用，所有通信都通过 ChatRoom 中介者完成。

更实用的例子：GUI 组件中介者

假设有一个表单：字体选择框、粗体复选框、斜体复选框、预览文本框。当用户改变任意一个选项，其他组件应同步更新。

class DialogMediator:
    def __init__(self):
        self.font_selector = None
        self.bold_checkbox = None
        self.italic_checkbox = None
        self.preview_label = None

    def register_components(self, font_selector, bold_checkbox, italic_checkbox, preview_label):
        self.font_selector = font_selector
        self.bold_checkbox = bold_checkbox
        self.italic_checkbox = italic_checkbox
        self.preview_label = preview_label

    def notify(self, sender, event):
        if sender == self.font_selector:
            self.update_preview()
        elif sender == self.bold_checkbox:
            self.update_preview()
        elif sender == self.italic_checkbox:
            self.update_preview()

    def update_preview(self):
        font = self.font_selector.get_value()
        bold = "bold" if self.bold_checkbox.is_checked() else "normal"
        italic = "italic" if self.italic_checkbox.is_checked() else "normal"
        style = f"{bold} {italic} {font}"
        self.preview_label.set_text(f"预览文字 - 样式: {style}")

# 模拟组件类
class FontSelector:
    def __init__(self, mediator: DialogMediator):
        self.mediator = mediator
        self.value = "Arial"

    def get_value(self):
        return self.value

    def change(self, new_font):
        self.value = new_font
        self.mediator.notify(self, "change")

class CheckBox:
    def __init__(self, mediator: DialogMediator, label: str):
        self.mediator = mediator
        self.label = label
        self.checked = False

    def is_checked(self):
        return self.checked

    def toggle(self):
        self.checked = not self.checked
        self.mediator.notify(self, "toggle")

class PreviewLabel:
    def set_text(self, text):
        print(f"[预览更新]: {text}")

# 使用
mediator = DialogMediator()

font_sel = FontSelector(mediator)
bold_cb = CheckBox(mediator, "Bold")
italic_cb = CheckBox(mediator, "Italic")
preview = PreviewLabel()

mediator.register_components(font_sel, bold_cb, italic_cb, preview)

# 模拟用户操作
font_sel.change("Times New Roman")
bold_cb.toggle()
italic_cb.toggle()
bold_cb.toggle()  # 再次切换

输出：

[预览更新]: 预览文字 - 样式: normal normal Times New Roman
[预览更新]: 预览文字 - 样式: bold normal Times New Roman
[预览更新]: 预览文字 - 样式: bold italic Times New Roman
[预览更新]: 预览文字 - 样式: normal italic Times New Roman

组件之间无需相互引用，所有联动逻辑集中在中介者中。

中介者模式结构总结

角色	说明
Mediator	抽象中介者（ChatRoom 接口）
ConcreteMediator	具体中介者（ConcreteChatRoom），协调同事
Colleague	同事类（User），持有中介者引用
ConcreteColleague	具体同事（Alice、Bob 等）

中介者模式 vs 观察者模式对比

模式	通信方式	耦合度	典型场景
中介者	所有通信通过中心中介者	同事松耦合	聊天室、GUI 组件联动
观察者	一对多广播	主题与观察者松耦合	事件通知、发布订阅

中介者更适合多对多复杂交互，观察者更适合一对多通知。

Python 中的实用建议

• 小型系统：直接用一个类作为中介者（最简单）
• 大型系统：结合事件总线（如 blinker 库）实现轻量中介者
• GUI 框架：PyQt/PySide 的信号槽机制本质就是中介者模式的一种实现
• 微服务：消息队列（如 RabbitMQ、Kafka）充当分布式中介者

注意事项

• 中介者不要变成“上帝类”（God Class），承担过多责任
• 如果交互逻辑简单，优先考虑直接引用或观察者模式
• 支持动态注册/注销同事（用户上线/下线）

中介者模式是处理对象间复杂交互的强大工具，能显著降低系统耦合，尤其在 GUI、聊天系统、工作流中非常实用。

如果你想看更高级的例子（如带私聊功能的聊天室、使用事件总线的分布式中介者、GUI 表单完整实现），欢迎继续问！