Python 中介者模式

中介者模式是一种行为设计模式，它通过引入一个中介者对象来减少对象之间的直接通信依赖。想象一下现实生活中的机场控制塔 - 所有飞机不直接相互通信，而是通过控制塔来协调起降和航线，这样就避免了混乱和冲突。

在软件开发中，当多个对象之间存在复杂的交互关系时，中介者模式能够将这些交互封装在中介者中，从而降低对象之间的耦合度。

核心思想

中介者模式的核心思想是 "不要直接对话，通过中介来协调"。它解决了对象间多对多交互的复杂性，让系统更易于维护和扩展。

为什么需要中介者模式

问题场景

假设我们正在开发一个聊天室系统，有多个用户需要相互发送消息。如果没有中介者，代码可能会是这样：

实例

class User:
    def __init__(self, name):
        self.name = name
        self.other_users = []  # 需要知道所有其他用户
   
    def send_message(self, message, to_user):
        print(f"{self.name} 发送给 {to_user.name}: {message}")
        to_user.receive_message(message, self)
   
    def receive_message(self, message, from_user):
        print(f"{self.name} 收到来自 {from_user.name} 的消息: {message}")

# 使用示例
user1 = User("小明")
user2 = User("小红")
user3 = User("小刚")

# 每个用户都需要知道其他所有用户
user1.other_users = [user2, user3]
user2.other_users = [user1, user3]
user3.other_users = [user1, user2]

user1.send_message("你好！", user2)

这种设计的缺点：

  • 对象间耦合度高，每个用户都需要知道其他所有用户
  • 添加新用户时需要修改所有现有用户
  • 难以维护，交互逻辑分散在各个对象中

解决方案

中介者模式通过引入聊天室（中介者）来解决这个问题：

实例

class ChatRoom:
    def __init__(self):
        self.users = {}
   
    def register_user(self, user):
        self.users[user.name] = user
        user.chat_room = self
   
    def send_message(self, message, from_user, to_user_name=None):
        if to_user_name:  # 私聊
            if to_user_name in self.users:
                self.users[to_user_name].receive_message(message, from_user)
        else:  # 群聊
            for user_name, user in self.users.items():
                if user_name != from_user.name:
                    user.receive_message(message, from_user)

class User:
    def __init__(self, name):
        self.name = name
        self.chat_room = None
   
    def send_message(self, message, to_user_name=None):
        if self.chat_room:
            self.chat_room.send_message(message, self, to_user_name)
   
    def receive_message(self, message, from_user):
        print(f"{self.name} 收到来自 {from_user.name} 的消息: {message}")

# 使用示例
chat_room = ChatRoom()

user1 = User("小明")
user2 = User("小红")
user3 = User("小刚")

chat_room.register_user(user1)
chat_room.register_user(user2)
chat_room.register_user(user3)

user1.send_message("大家好！")  # 群发消息
user2.send_message("你好小明！", "小明")  # 私聊

中介者模式的实现

基本结构

让我们通过一个更完整的例子来理解中介者模式的实现：

实例

from abc import ABC, abstractmethod
from typing import List

# 抽象中介者
class Mediator(ABC):
    @abstractmethod
    def notify(self, sender: object, event: str, data: dict = None):
        pass

# 具体中介者 - 聊天室
class ChatRoomMediator(Mediator):
    def __init__(self):
        self.users: List[User] = []
   
    def add_user(self, user):
        self.users.append(user)
        user.set_mediator(self)
   
    def notify(self, sender, event, data=None):
        if event == "send_message":
            message = data.get("message")
            target_user = data.get("target_user")
           
            if target_user:  # 私聊
                for user in self.users:
                    if user.name == target_user:
                        user.receive_message(message, sender.name)
            else:  # 群聊
                for user in self.users:
                    if user != sender:
                        user.receive_message(message, sender.name)
       
        elif event == "user_joined":
            message = f"系统: {sender.name} 加入了聊天室"
            for user in self.users:
                if user != sender:
                    user.receive_message(message, "系统")

# 基础组件类
class BaseComponent:
    def __init__(self):
        self._mediator = None
   
    def set_mediator(self, mediator: Mediator):
        self._mediator = mediator

# 具体组件 - 用户
class User(BaseComponent):
    def __init__(self, name):
        super().__init__()
        self.name = name
   
    def send_message(self, message, target_user=None):
        print(f"{self.name} 发送消息: {message}")
        self._mediator.notify(self, "send_message", {
            "message": message,
            "target_user": target_user
        })
   
    def join_chat(self):
        self._mediator.notify(self, "user_joined")
   
    def receive_message(self, message, from_name):
        print(f"{self.name} 收到来自 {from_name} 的消息: {message}")

# 使用示例
def main():
    # 创建中介者
    chat_room = ChatRoomMediator()
   
    # 创建用户
    alice = User("Alice")
    bob = User("Bob")
    charlie = User("Charlie")
   
    # 注册用户到聊天室
    chat_room.add_user(alice)
    chat_room.add_user(bob)
    chat_room.add_user(charlie)
   
    print("=== 聊天室演示 ===")
   
    # Alice 加入聊天室
    alice.join_chat()
   
    # 发送消息
    alice.send_message("大家好，我是 Alice！")
    bob.send_message("欢迎 Alice！")
    charlie.send_message("Alice 你好！", "Alice")  # 私聊
   
    # Bob 发送群消息
    bob.send_message("有人在线吗？")

if __name__ == "__main__":
    main()

运行结果

=== 聊天室演示 ===
系统: Alice 加入了聊天室
Alice 发送消息: 大家好，我是 Alice！
Bob 收到来自 Alice 的消息: 大家好，我是 Alice！
Charlie 收到来自 Alice 的消息: 大家好，我是 Alice！
Bob 发送消息: 欢迎 Alice！
Alice 收到来自 Bob 的消息: 欢迎 Alice！
Charlie 收到来自 Bob 的消息: 欢迎 Alice！
Charlie 发送消息: Alice 你好！
Alice 收到来自 Charlie 的消息: Alice 你好！
Bob 发送消息: 有人在线吗？
Alice 收到来自 Bob 的消息: 有人在线吗？
Charlie 收到来自 Bob 的消息: 有人在线吗？

中介者模式的 UML 结构

结构说明

  • Mediator（中介者接口）：定义通信接口
  • ConcreteMediator（具体中介者）：实现协调逻辑，了解所有组件
  • BaseComponent（基础组件）：包含中介者引用
  • ConcreteComponent（具体组件）：实现具体业务逻辑

实际应用场景

场景 1：GUI 应用程序

在图形用户界面中，各种控件（按钮、文本框、复选框等）通过中介者来协调交互：

实例

class DialogMediator:
    def __init__(self):
        self.login_button = None
        self.username_input = None
        self.password_input = None
        self.remember_checkbox = None
   
    def notify(self, sender, event):
        if event == "username_changed" or event == "password_changed":
            # 当用户名或密码输入时，检查是否可以启用登录按钮
            username = self.username_input.get_text()
            password = self.password_input.get_text()
            self.login_button.set_enabled(bool(username and password))
       
        elif event == "login_clicked":
            # 处理登录逻辑
            username = self.username_input.get_text()
            password = self.password_input.get_text()
            remember = self.remember_checkbox.is_checked()
            print(f"登录: {username}, 记住我: {remember}")
       
        elif event == "remember_changed":
            print("记住我选项已更改")

class UIComponent:
    def __init__(self, mediator=None):
        self.mediator = mediator
   
    def set_mediator(self, mediator):
        self.mediator = mediator

class Button(UIComponent):
    def click(self):
        if self.mediator:
            self.mediator.notify(self, "login_clicked")
   
    def set_enabled(self, enabled):
        print(f"按钮 {'启用' if enabled else '禁用'}")

class TextInput(UIComponent):
    def __init__(self, mediator=None):
        super().__init__(mediator)
        self._text = ""
   
    def set_text(self, text):
        self._text = text
        if self.mediator:
            self.mediator.notify(self, "username_changed")
   
    def get_text(self):
        return self._text

# 使用示例
dialog = DialogMediator()

login_btn = Button()
username_input = TextInput()
password_input = TextInput()

dialog.login_button = login_btn
dialog.username_input = username_input
dialog.password_input = password_input

login_btn.set_mediator(dialog)
username_input.set_mediator(dialog)
password_input.set_mediator(dialog)

# 模拟用户输入
username_input.set_text("user123")
password_input.set_text("pass123")
login_btn.click()

场景 2：电商订单系统

在电商系统中，订单处理涉及库存、支付、物流等多个子系统：

实例

class OrderMediator:
    def __init__(self):
        self.inventory_system = None
        self.payment_system = None
        self.shipping_system = None
        self.notification_system = None
   
    def place_order(self, order_data):
        print("=== 开始处理订单 ===")
       
        # 检查库存
        if not self.inventory_system.check_stock(order_data):
            return "库存不足"
       
        # 处理支付
        payment_result = self.payment_system.process_payment(order_data)
        if not payment_result:
            return "支付失败"
       
        # 更新库存
        self.inventory_system.update_stock(order_data)
       
        # 安排发货
        shipping_info = self.shipping_system.schedule_delivery(order_data)
       
        # 发送通知
        self.notification_system.send_confirmation(order_data, shipping_info)
       
        return "订单处理成功"

class InventorySystem:
    def check_stock(self, order_data):
        print("检查库存...")
        return True
   
    def update_stock(self, order_data):
        print("更新库存...")

class PaymentSystem:
    def process_payment(self, order_data):
        print("处理支付...")
        return True

class ShippingSystem:
    def schedule_delivery(self, order_data):
        print("安排发货...")
        return "快递单号: SF123456789"

class NotificationSystem:
    def send_confirmation(self, order_data, shipping_info):
        print(f"发送确认邮件，发货信息: {shipping_info}")

# 使用示例
mediator = OrderMediator()

mediator.inventory_system = InventorySystem()
mediator.payment_system = PaymentSystem()
mediator.shipping_system = ShippingSystem()
mediator.notification_system = NotificationSystem()

order_data = {"product_id": "123", "quantity": 2, "user_id": "user001"}
result = mediator.place_order(order_data)
print(f"订单结果: {result}")

中介者模式的优缺点

优点

优点 说明
单一职责原则 将组件间的交互逻辑集中到中介者中
开闭原则 可以引入新的中介者而无需修改现有组件
降低耦合度 减少组件间的直接依赖
易于复用 单个组件可以更容易地在不同上下文中复用

缺点

缺点 说明
中介者可能变得复杂 随着交互逻辑增多，中介者可能变成上帝对象
性能开销 所有通信都经过中介者，可能引入性能瓶颈
调试困难 交互逻辑集中在中介者中，调试时可能需要跟踪多个组件

最佳实践和注意事项

1. 合理划分中介者职责

不要让一个中介者处理所有事情，可以根据业务领域划分多个中介者：

实例

class UserRegistrationMediator:
    def notify(self, sender, event, data):
        if event == "user_registered":
            # 处理用户注册后的相关逻辑
            self.send_welcome_email(data)
            self.create_user_profile(data)
            self.add_to_mailing_list(data)

class OrderProcessingMediator:
    def notify(self, sender, event, data):
        if event == "order_created":
            # 处理订单创建后的相关逻辑
            self.validate_order(data)
            self.process_payment(data)
            self.update_inventory(data)

2. 避免中介者过度复杂

如果中介者变得过于复杂，考虑使用其他模式或重构：

实例

# 不好的做法 - 中介者包含太多逻辑
class ComplexMediator:
    def handle_event(self, event_type, data):
        if event_type == "user_action":
            # 大量复杂的业务逻辑...
            pass
        elif event_type == "system_event":
            # 更多复杂的逻辑...
            pass

# 好的做法 - 使用策略模式分解复杂逻辑
class EventHandler:
    def handle(self, data):
        pass

class UserActionHandler(EventHandler):
    def handle(self, data):
        # 专门处理用户动作
        pass

class SimpleMediator:
    def __init__(self):
        self.handlers = {}
   
    def register_handler(self, event_type, handler):
        self.handlers[event_type] = handler
   
    def notify(self, event_type, data):
        if event_type in self.handlers:
            self.handlers[event_type].handle(data)

3. 考虑使用观察者模式结合

中介者模式常与观察者模式结合使用：

实例

from abc import ABC, abstractmethod
from typing import List

class Observer(ABC):
    @abstractmethod
    def update(self, event, data):
        pass

class Observable:
    def __init__(self):
        self._observers: List[Observer] = []
   
    def add_observer(self, observer):
        self._observers.append(observer)
   
    def notify_observers(self, event, data=None):
        for observer in self._observers:
            observer.update(event, data)

class ChatMediator(Observable):
    def send_message(self, message, sender, receiver=None):
        # 中介者逻辑
        event_data = {
            "message": message,
            "sender": sender,
            "receiver": receiver
        }
        self.notify_observers("message_sent", event_data)

class LoggingObserver(Observer):
    def update(self, event, data):
        if event == "message_sent":
            print(f"日志: {data['sender']} 发送了消息")

练习题目

练习 1：改进聊天室系统

扩展之前的聊天室系统，添加以下功能：

  1. 支持创建多个聊天室
  2. 支持用户在不同聊天室间切换
  3. 添加管理员功能（踢人、禁言等）

练习 2：设计机票预订系统

使用中介者模式设计一个机票预订系统，包含以下组件：

  • 航班查询
  • 座位选择
  • 支付处理
  • 票务生成
  • 通知发送

要求各个组件通过中介者协调工作。

练习 3：智能家居控制系统

设计一个智能家居控制系统，包含：

  • 灯光控制
  • 温度调节
  • 安防监控
  • 场景模式

通过中介者实现设备间的智能联动。

总结

中介者模式是处理复杂对象交互的强大工具。它通过引入协调中心来简化系统架构，特别适用于以下场景：

  • 对象间存在复杂的网状交互关系
  • 需要集中控制多个对象间的交互逻辑
  • 希望降低对象间的耦合度

记住，中介者模式不是万能的，当交互逻辑很简单时，直接的对象间通信可能更合适。但在处理复杂系统时，合理使用中介者模式能够显著提高代码的可维护性和扩展性。

关键要点：

  • 中介者封装了对象间的交互逻辑
  • 组件只与中介者通信，不直接相互依赖
  • 中介者可能变得复杂，需要合理设计
  • 结合其他模式（如观察者）可以获得更好的效果