外观模式

外观模式（Facade Pattern）为子系统中的一组接口提供一个一致的界面，定义一个高层接口，这个接口使得这一子系统更加容易使用。

问题定义

场景1：复杂系统初始化

# ❌ 问题：客户端需要了解所有细节
class CPU:
def freeze(self):
print("CPU: 冻结")
def jump(self, position):
print(f"CPU: 跳转到 {position}")
def execute(self):
print("CPU: 执行指令")
class Memory:
def load(self, position, data):
print(f"内存: 加载数据到 {position}")
def read(self, position):
print(f"内存: 读取 {position}")
class HardDrive:
def read(self, lba, size):
print(f"硬盘: 读取 LBA={lba}, 大小={size}")
# 问题：启动计算机需要按顺序调用多个子系统的多个方法
def boot_computer():
cpu = CPU()
memory = Memory()
hard_drive = HardDrive()
print("启动中...")
cpu.freeze()
memory.load(0, hard_drive.read(0, 1024))
cpu.jump(0)
cpu.execute()
print("启动完成")
# 客户端需要知道所有细节
boot_computer()

场景2：第三方库集成

# ❌ 问题：复杂API，难以使用
class EmailService:
def connect(self, host, port):
print(f"连接邮件服务器: {host}:{port}")
def authenticate(self, username, password):
print(f"认证: {username}")
def send(self, to, subject, body):
print(f"发送邮件到 {to}")
class SMSService:
def init_gateway(self, api_key):
print(f"初始化短信网关: {api_key}")
def send(self, phone, message):
print(f"发送短信到 {phone}")
class LogService:
def open_log_file(self, filename):
print(f"打开日志文件: {filename}")
def write(self, level, message):
print(f"[{level}] {message}")
def close_log_file(self):
print("关闭日志文件")
# 问题：发送通知需要调用多个服务
def send_notification(to_email, to_phone, message):
# 发送邮件
email = EmailService()
email.connect("smtp.example.com", 587)
email.authenticate("user", "pass")
email.send(to_email, "通知", message)
# 发送短信
sms = SMSService()
sms.init_gateway("api_key_123")
sms.send(to_phone, message)
# 记录日志
log = LogService()
log.open_log_file("app.log")
log.write("INFO", "发送通知")
log.close_log_file()

解决方案

外观模式提供一个简化接口，隐藏子系统的复杂性。

标准实现

子系统

class CPU:
"""CPU子系统"""
def freeze(self):
print("CPU: 冻结")
def jump(self, position):
print(f"CPU: 跳转到 {position}")
def execute(self):
print("CPU: 执行指令")
class Memory:
"""内存子系统"""
def load(self, position, data):
print(f"内存: 加载数据到 {position}")
class HardDrive:
"""硬盘子系统"""
def read(self, lba, size):
print(f"硬盘: 读取 LBA={lba}, 大小={size}")
return "Boot Data"

外观类

class ComputerFacade:
"""计算机外观类 - 简化接口"""
def __init__(self):
self.cpu = CPU()
self.memory = Memory()
self.hard_drive = HardDrive()
def start(self):
"""启动计算机 - 一个方法搞定"""
print("启动计算机...")
self.cpu.freeze()
boot_data = self.hard_drive.read(0, 1024)
self.memory.load(0, boot_data)
self.cpu.jump(0)
self.cpu.execute()
print("计算机启动完成！")
def shutdown(self):
"""关闭计算机"""
print("关闭计算机...")
print("计算机已关闭")

客户端使用

# 使用外观 - 简单！
computer = ComputerFacade()
computer.start()
# 启动计算机...
# CPU: 冻结
# 硬盘: 读取 LBA=0, 大小=1024
# 内存: 加载数据到 0
# CPU: 跳转到 0
# CPU: 执行指令
# 计算机启动完成！
computer.shutdown()
# 关闭计算机...
# 计算机已关闭

实战应用

应用1：通知系统

class EmailService:
"""邮件服务"""
def connect(self):
print("连接邮件服务器")
def authenticate(self):
print("邮件认证")
def send(self, to, subject, body):
print(f"发送邮件到 {to}: {subject}")
def disconnect(self):
print("断开邮件连接")
class SMSService:
"""短信服务"""
def init_gateway(self):
print("初始化短信网关")
def send(self, phone, message):
print(f"发送短信到 {phone}")
def cleanup(self):
print("清理短信网关")
class LogService:
"""日志服务"""
def open_file(self):
print("打开日志文件")
def write(self, message):
print(f"[LOG] {message}")
def close_file(self):
print("关闭日志文件")
class NotificationFacade:
"""通知外观类"""
def __init__(self):
self.email = EmailService()
self.sms = SMSService()
self.log = LogService()
def send_notification(
self,
to_email=None,
to_phone=None,
subject="",
message=""
):
"""发送通知 - 统一接口"""
self.log.write(f"发送通知: {message}")
# 发送邮件
if to_email:
self.email.connect()
self.email.authenticate()
self.email.send(to_email, subject, message)
self.email.disconnect()
# 发送短信
if to_phone:
self.sms.init_gateway()
self.sms.send(to_phone, message)
self.sms.cleanup()
self.log.write("通知发送完成")
# 使用
notifier = NotificationFacade()
notifier.send_notification(
to_email="user@example.com",
to_phone="13800138000",
subject="订单通知",
message="您的订单已发货"
)
# [LOG] 发送通知: 您的订单已发货
# 连接邮件服务器
# 邮件认证
# 发送邮件到 user@example.com: 订单通知
# 断开邮件连接
# 初始化短信网关
# 发送短信到 13800138000
# 清理短信网关
# [LOG] 通知发送完成

应用2：文件转换系统

class FileReader:
"""文件读取器"""
def open_file(self, filename):
print(f"打开文件: {filename}")
def read_content(self):
print("读取文件内容")
return "文件内容"
def close_file(self):
print("关闭文件")
class ImageProcessor:
"""图像处理器"""
def load_image(self, data):
print("加载图像")
def resize(self, width, height):
print(f"调整大小: {width}x{height}")
def compress(self, quality):
print(f"压缩质量: {quality}")
def get_result(self):
return "处理后的图像"
class FileWriter:
"""文件写入器"""
def create_file(self, filename):
print(f"创建文件: {filename}")
def write_content(self, content):
print("写入内容")
def close_file(self):
print("保存并关闭文件")
class ImageConverterFacade:
"""图像转换外观类"""
def __init__(self):
self.reader = FileReader()
self.processor = ImageProcessor()
self.writer = FileWriter()
def convert_image(
self,
input_file,
output_file,
width=None,
height=None,
quality=80
):
"""转换图像 - 一个方法搞定"""
print(f"\n转换图像: {input_file} -> {output_file}")
# 读取
self.reader.open_file(input_file)
data = self.reader.read_content()
self.reader.close_file()
# 处理
self.processor.load_image(data)
if width and height:
self.processor.resize(width, height)
self.processor.compress(quality)
result = self.processor.get_result()
# 写入
self.writer.create_file(output_file)
self.writer.write_content(result)
self.writer.close_file()
print("转换完成！\n")
# 使用
converter = ImageConverterFacade()
# 简单转换
converter.convert_image("input.jpg", "output.jpg")
# 带调整大小和压缩
converter.convert_image(
"input.jpg",
"output.jpg",
width=800,
height=600,
quality=90
)

应用3：电商订单系统

class InventoryService:
"""库存服务"""
def check_stock(self, product_id):
print(f"检查库存: 产品 {product_id}")
return True
def deduct_stock(self, product_id, quantity):
print(f"扣减库存: 产品 {product_id}, 数量 {quantity}")
class PaymentService:
"""支付服务"""
def create_payment(self, order_id, amount):
print(f"创建支付: 订单 {order_id}, 金额 ${amount}")
return "PAY123"
def process_payment(self, payment_id):
print(f"处理支付: {payment_id}")
return True
class ShippingService:
"""物流服务"""
def create_shipment(self, order_id):
print(f"创建运单: 订单 {order_id}")
return "SHIP456"
def schedule_pickup(self, shipment_id):
print(f"安排取件: {shipment_id}")
class NotificationService:
"""通知服务"""
def send_confirmation(self, email):
print(f"发送确认邮件到: {email}")
def send_sms(self, phone, message):
print(f"发送短信到 {phone}: {message}")
class OrderFacade:
"""订单外观类"""
def __init__(self):
self.inventory = InventoryService()
self.payment = PaymentService()
self.shipping = ShippingService()
self.notification = NotificationService()
def place_order(
self,
order_id,
product_id,
quantity,
amount,
customer_email,
customer_phone
):
"""下单 - 一个方法搞定"""
print(f"\n处理订单: {order_id}")
print("=" * 50)
# 1. 检查库存
if not self.inventory.check_stock(product_id):
print("库存不足，订单失败")
return False
# 2. 扣减库存
self.inventory.deduct_stock(product_id, quantity)
# 3. 创建并处理支付
payment_id = self.payment.create_payment(order_id, amount)
if not self.payment.process_payment(payment_id):
print("支付失败，订单失败")
return False
# 4. 创建运单
shipment_id = self.shipping.create_shipment(order_id)
self.shipping.schedule_pickup(shipment_id)
# 5. 发送通知
self.notification.send_confirmation(customer_email)
self.notification.send_sms(
customer_phone,
f"您的订单 {order_id} 已确认"
)
print("=" * 50)
print("订单处理完成！\n")
return True
# 使用
order_system = OrderFacade()
order_system.place_order(
order_id="ORD001",
product_id="PROD123",
quantity=2,
amount=199.99,
customer_email="user@example.com",
customer_phone="13800138000"
)

应用4：数据库操作

class ConnectionPool:
"""连接池"""
def get_connection(self):
print("从连接池获取连接")
return "Connection"
def release_connection(self, conn):
print("释放连接回连接池")
class QueryExecutor:
"""查询执行器"""
def execute_query(self, conn, sql, params):
print(f"执行查询: {sql}")
print(f"参数: {params}")
return [("user1", "Alice"), ("user2", "Bob")]
def execute_update(self, conn, sql, params):
print(f"执行更新: {sql}")
print(f"参数: {params}")
return 1  # 影响的行数
class TransactionManager:
"""事务管理器"""
def begin_transaction(self, conn):
print("开始事务")
def commit(self, conn):
print("提交事务")
def rollback(self, conn):
print("回滚事务")
class DatabaseFacade:
"""数据库外观类"""
def __init__(self):
self.pool = ConnectionPool()
self.executor = QueryExecutor()
self.txn_manager = TransactionManager()
def fetch_all(self, sql, params=None):
"""查询所有 - 自动管理连接"""
conn = self.pool.get_connection()
try:
results = self.executor.execute_query(conn, sql, params or {})
return results
finally:
self.pool.release_connection(conn)
def execute_update(self, sql, params=None):
"""执行更新 - 自动管理连接"""
conn = self.pool.get_connection()
try:
affected = self.executor.execute_update(conn, sql, params or {})
return affected
finally:
self.pool.release_connection(conn)
def execute_transaction(self, operations):
"""执行事务 - 自动管理事务"""
conn = self.pool.get_connection()
self.txn_manager.begin_transaction(conn)
try:
results = []
for op in operations:
result = self.executor.execute_update(conn, op["sql"], op["params"])
results.append(result)
self.txn_manager.commit(conn)
return results
except Exception as e:
self.txn_manager.rollback(conn)
raise e
finally:
self.pool.release_connection(conn)
# 使用
db = DatabaseFacade()
# 简单查询
users = db.fetch_all(
"SELECT * FROM users WHERE age > ?",
{"age": 18}
)
# 简单更新
affected = db.execute_update(
"UPDATE users SET status = ? WHERE id = ?",
{"status": "active", "id": 1}
)
# 事务操作
operations = [
{"sql": "UPDATE account SET balance = balance - ? WHERE id = ?",
"params": {"amount": 100, "id": 1}},
{"sql": "UPDATE account SET balance = balance + ? WHERE id = ?",
"params": {"amount": 100, "id": 2}}
]
results = db.execute_transaction(operations)

外观模式 vs 适配器模式

优缺点

✅ 优点

❌ 缺点

适用场景

本章要点

• ✅ 外观模式提供简化接口
• ✅ 隐藏子系统的复杂性
• ✅ 客户端只依赖外观类
• ✅ 可以在不影响客户端的情况下改变子系统
• ✅ 适用于需要简化复杂系统的场景
• ✅ 与适配器模式目的不同

下一步：组合模式 🚀