Mediator Pattern is a behavioral design pattern that reduces direct communication dependencies between objects by introducing a mediator object. Imagine an airport control tower in real life - all aircraft do not communicate directly with each other, but coordinate takeoffs, landings, and routes through the control tower, thus avoiding chaos and conflicts.
\\n\\nIn software development, when multiple objects have complex interaction relationships, the mediator pattern can encapsulate these interactions within the mediator, thereby reducing coupling between objects.
\\n\\nCore Idea
\\n\\nThe core idea of the mediator pattern is "Don't talk directly, coordinate through a mediator". It solves the complexity of many-to-many interactions between objects, making the system easier to maintain and extend.
\\n\\n\\n\\n
Why We Need the Mediator Pattern
\\n\\nProblem Scenario
\\n\\nSuppose we are developing a chat room system where multiple users need to send messages to each other. Without a mediator, the code might look like this:
\\n\\nExamples
\\n\\nclass User:\\n\\n def __init__(self, name):\\n self.name = name\\n self.other_users = [] # Need to know all other Users\\n\\n def send_message(self, message, to_user):\\n print(f"{self.name} Send To {to_user.name}: {message}")\\n to_user.receive_message(message, self)\\n\\n def receive_message(self, message, from_user):\\n print(f"{self.name} Received From {from_user.name} 's message: {message}")\\n\\n# Usage example\\nuser1 = User("Xiaoming")\\nuser2 = User("Xiaohong")\\nuser3 = User("Xiao Gang")\\n\\n# Each User needs to know all other Users\\nuser1.other_users = [user2, user3]\\nuser2.other_users = [user1, user3]\\nuser3.other_users = [user1, user2]\\n\\nuser1.send_message("Hello!", user2)\\n\\nDisadvantages of this design:
\\n\\n- \\n
- High coupling between objects, each user needs to know all other users \\n
- Adding new users requires modifying all existing users \\n
- Difficult to maintain, interaction logic is scattered across various objects \\n
Solution
\\n\\nThe mediator pattern solves this problem by introducing a chat room (mediator):
\\n\\nExamples
\\n\\nclass ChatRoom:\\n\\n def __init__(self):\\n self.users = {}\\n\\n def register_user(self, user):\\n self.users[user.name] = user\\n user.chat_room = self\\n\\n def send_message(self, message, from_user, to_user_name=None):\\n if to_user_name: # Private Chat\\n if to_user_name in self.users:\\n self.users.receive_message(message, from_user)\\n else: # Group chat\\n for user_name, user in self.users.items():\\n if user_name != from_user.name:\\n user.receive_message(message, from_user)\\n\\nclass User:\\n\\n def __init__(self, name):\\n self.name = name\\n self.chat_room = None\\n\\n def send_message(self, message, to_user_name=None):\\n if self.chat_room:\\n self.chat_room.send_message(message, self, to_user_name)\\n\\n def receive_message(self, message, from_user):\\n print(f"{self.name} Received From {from_user.name} 's message: {message}")\\n\\n# Usage example\\nchat_room = ChatRoom()\\n\\nuser1 = User("Xiaoming")\\nuser2 = User("Xiaohong")\\nuser3 = User("Xiao Gang")\\n\\nchat_room.register_user(user1)\\nchat_room.register_user(user2)\\nchat_room.register_user(user3)\\n\\nuser1.send_message("Hello Everyone!") # Broadcast message\\nuser2.send_message("Hello Xiaoming!", "Xiaoming") # Private Chat\\n\\n\\n\\n
Implementation of the Mediator Pattern
\\n\\nBasic Structure
\\n\\nLet's understand the implementation of the mediator pattern through a more complete example:
\\n\\nExamples
\\n\\nfrom abc import ABC, abstractmethod\\nfrom typing import List\\n\\n# Abstract mediator\\nclass Mediator(ABC):\\n\\n @abstractmethod\\n def notify(self, sender: object, event: str, data: dict = None):\\n pass\\n\\n# Concrete mediator - Chat room\\nclass ChatRoomMediator(Mediator):\\n\\n def __init__(self):\\n self.users: List = []\\n\\n def add_user(self, user):\\n self.users.append(user)\\n user.set_mediator(self)\\n\\n def notify(self, sender, event, data=None):\\n if event == "send_message":\\n message = data.get("message")\\n target_user = data.get("target_user")\\n\\n if target_user: # Private Chat\\n for user in self.users:\\n if user.name == target_user:\\n user.receive_message(message, sender.name)\\n else: # Group chat\\n for user in self.users:\\n if user != sender:\\n user.receive_message(message, sender.name)\\n\\n elif event == "user_joined":\\n message = f"System: {sender.name} Joined chat room"\\n for user in self.users:\\n if user != sender:\\n user.receive_message(message, "System")\\n\\n# Base Component Class\\nclass BaseComponent:\\n\\n def __init__(self):\\n self._mediator = None\\n\\n def set_mediator(self, mediator: Mediator):\\n self._mediator = mediator\\n\\n# Concrete component - User\\nclass User(BaseComponent):\\n\\n def __init__(self, name):\\n super().__init__()\\n self.name = name\\n\\n def send_message(self, message, target_user=None):\\n print(f"{self.name} Send message: {message}")\\n self._mediator.notify(self, "send_message", {\\n "message": message,\\n "target_user": target_user\\n })\\n\\n def join_chat(self):\\n self._mediator.notify(self, "user_joined")\\n\\n def receive_message(self, message, from_name):\\n print(f"{self.name} Received From {from_name} 's message: {message}")\\n\\n# Usage example\\ndef main():\\n # Create Mediator\\n chat_room = ChatRoomMediator()\\n\\n # Create User\\n alice = User("Alice")\\n bob = User("Bob")\\n charlie = User("Charlie")\\n\\n # Register User to chat room\\n chat_room.add_user(alice)\\n chat_room.add_user(bob)\\n chat_room.add_user(charlie)\\n\\n print("=== Chat room demo ===")\\n\\n # Alice Join chat room\\n alice.join_chat()\\n\\n # Send message\\n alice.send_message("Hello everyone, I'm Alice!")\\n bob.send_message("Welcome Alice!")\\n charlie.send_message("Alice Hello!", "Alice") # Private Chat\\n\\n # Bob Send Group Message\\n bob.send_message("Is anyone online?")\\n\\nif __name__ == "__main__":\\n main()\\n\\nRunning Results
\\n\\n=== Chat room demo ===\\nSystem: Alice Joined chat room\\nAlice Send message: Hello everyone, I'm Alice!\\nBob Received message from Alice: Hello everyone, I'm Alice!\\nCharlie Received message from Alice: Hello everyone, I'm Alice!\\nBob Send message: Welcome Alice!\\nAlice Received message from Bob: Welcome Alice!\\nCharlie Received message from Bob: Welcome Alice!\\nCharlie Send message: Alice Hello!\\nAlice Received message from Charlie: Alice Hello!\\nBob Send message: Is anyone online?\\nAlice Received message from Bob: Is anyone online?\\nCharlie Received message from Bob: Is anyone online?\\n\\n\\n\\n
UML Structure of the Mediator Pattern
\\n\\nStructure Explanation
\\n\\n- \\n
- Mediator (Mediator Interface): Defines communication interface \\n
- ConcreteMediator (Concrete Mediator): Implements coordination logic, knows all components \\n
- BaseComponent (Base Component): Contains mediator reference \\n
- ConcreteComponent (Concrete Component): Implements specific business logic \\n
\\n\\n
Practical Application Scenarios
\\n\\nScenario 1: GUI Applications
\\n\\nIn graphical user interfaces, various controls (buttons, text boxes, checkboxes, etc.) coordinate interactions through a mediator:
\\n\\nExamples
\\n\\nclass DialogMediator:\\n\\n def __init__(self):\\n self.login_button = None\\n self.username_input = None\\n self.password_input = None\\n self.remember_checkbox = None\\n\\n def notify(self, sender, event):\\n if event == "username_changed" or event == "password_changed":\\n # WhenUserCheck if EnableLoginButton can be enabled when username or password is entered.\\n username = self.username_input.get_text()\\n password = self.password_input.get_text()\\n self.login_button.set_enabled(bool(username and password))\\n\\n elif event == "login_clicked":\\n # Handle Login logic\\n username = self.username_input.get_text()\\n password = self.password_input.get_text()\\n remember = self.remember_checkbox.is_checked()\\n print(f"Login: {username}, Remember me: {remember}")\\n\\n elif event == "remember_changed":\\n print("Remember me option changed")\\n\\nclass UIComponent:\\n\\n def __init__(self, mediator=None):\\n self.mediator = mediator\\n\\n def set_mediator(self, mediator):\\n self.mediator = mediator\\n\\nclass Button(UIComponent):\\n\\n def click(self):\\n if self.mediator:\\n self.mediator.notify(self, "login_clicked")\\n\\n def set_enabled(self, enabled):\\n print(f"Button {'Enable' if enabled else 'Disable'}")\\n\\nclass TextInput(UIComponent):\\n\\n def __init__(self, mediator=None):\\n super().__init__(mediator)\\n self._text = ""\\n\\n def set_text(self, text):\\n self._text = text\\n if self.mediator:\\n self.mediator.notify(self, "username_changed")\\n\\n def get_text(self):\\n return self._text\\n\\n# Usage example\\ndialog = DialogMediator()\\n\\nlogin_btn = Button()\\nusername_input = TextInput()\\npassword_input = TextInput()\\n\\ndialog.login_button = login_btn\\ndialog.username_input = username_input\\ndialog.password_input = password_input\\n\\nlogin_btn.set_mediator(dialog)\\nusername_input.set_mediator(dialog)\\npassword_input.set_mediator(dialog)\\n\\n# Simulate User input\\nusername_input.set_text("user123")\\npassword_input.set_text("pass123")\\nlogin_btn.click()\\n\\nScenario 2: E-commerce Order System
\\n\\nIn e-commerce systems, order processing involves multiple subsystems such as inventory, payment, and logistics:
\\n\\nExamples
\\n\\nclass OrderMediator:\\n\\n def __init__(self):\\n self.inventory_system = None\\n self.payment_system = None\\n self.shipping_system = None\\n self.notification_system = None\\n\\n def place_order(self, order_data):\\n print("=== Start Processing Order ===")\\n\\n # Check inventory\\n if not self.inventory_system.check_stock(order_data):\\n return "Insufficient Stock"\\n\\n # Process Payment\\n payment_result = self.payment_system.process_payment(order_data)\\n if not payment_result:\\n return "Payment Failed"\\n\\n # Update Inventory\\n self.inventory_system.update_stock(order_data)\\n\\n # Arrange shipment\\n shipping_info = self.shipping_system.schedule_delivery(order_data)\\n\\n # Send notification\\n self.notification_system.send_confirmation(order_data, shipping_info)\\n\\n return "Order Processed Successfully"\\n\\nclass InventorySystem:\\n\\n def check_stock(self, order_data):\\n print("Check inventory...")\\n return True\\n\\n def update_stock(self, order_data):\\n print("Update Inventory...")\\n\\nclass PaymentSystem:\\n\\n def process_payment(self, order_data):\\n print("Process Payment...")\\n return True\\n\\nclass ShippingSystem:\\n\\n def schedule_delivery(self, order_data):\\n print("Arrange shipment...")\\n return "Tracking number: SF123456789"\\n\\nclass NotificationSystem:\\n\\n def send_confirmation(self, order_data, shipping_info):\\n print(f"Send confirmation email, shipping information: {shipping_info}")\\n\\n# Usage example\\nmediator = OrderMediator()\\n\\nmediator.inventory_system = InventorySystem()\\nmediator.payment_system = PaymentSystem()\\nmediator.shipping_system = ShippingSystem()\\nmediator.notification_system = NotificationSystem()
YouTip