实际项目案例

本章通过实际项目案例，展示如何组合使用多个设计模式来解决真实问题。

案例1：电商平台订单系统

项目背景

构建一个电商平台的订单管理系统，需要处理： - 订单创建和状态管理 - 支付处理 - 物流跟踪 - 库存管理 - 通知系统

架构设计

核心代码实现

1. 订单工厂（工厂方法模式）

from abc import ABC, abstractmethod
class Order(ABC):
"""订单抽象类"""
def __init__(self, order_id, customer, items):
self.order_id = order_id
self.customer = customer
self.items = items
self.status = "pending"
@abstractmethod
def calculate_total(self):
pass
def __str__(self):
return f"订单 {self.order_id}: {self.customer} - ${self.calculate_total():.2f}"
class StandardOrder(Order):
"""标准订单"""
def calculate_total(self):
total = sum(item['price'] * item['quantity'] for item in self.items)
return total
class PremiumOrder(Order):
"""高级订单"""
def __init__(self, order_id, customer, items, discount_rate=0.1):
super().__init__(order_id, customer, items)
self.discount_rate = discount_rate
def calculate_total(self):
total = sum(item['price'] * item['quantity'] for item in self.items)
return total * (1 - self.discount_rate)
class OrderFactory:
"""订单工厂"""
@staticmethod
def create_order(order_type, order_id, customer, items, **kwargs):
"""创建订单"""
if order_type == "standard":
return StandardOrder(order_id, customer, items)
elif order_type == "premium":
return PremiumOrder(order_id, customer, items, **kwargs)
else:
raise ValueError(f"未知订单类型: {order_type}")
# 使用
items = [
{"name": "商品A", "price": 100, "quantity": 2},
{"name": "商品B", "price": 50, "quantity": 1}
]
standard_order = OrderFactory.create_order("standard", "ORD-001", "Alice", items)
premium_order = OrderFactory.create_order("premium", "ORD-002", "Bob", items, discount_rate=0.2)
print(standard_order)  # 订单 ORD-001: Alice - $250.00
print(premium_order)   # 订单 ORD-002: Bob - $200.00

2. 订单状态管理（状态模式）

from abc import ABC, abstractmethod
class OrderState(ABC):
"""订单状态接口"""
@abstractmethod
def confirm(self, order):
pass
@abstractmethod
def ship(self, order):
pass
@abstractmethod
def deliver(self, order):
pass
@abstractmethod
def cancel(self, order):
pass
class PendingState(OrderState):
"""待处理状态"""
def confirm(self, order):
print(f"订单 {order.order_id} 已确认")
order.status = "confirmed"
return ConfirmedState()
def ship(self, order):
print(f"订单 {order.order_id} 未确认，无法发货")
def deliver(self, order):
print(f"订单 {order.order_id} 未发货，无法送达")
def cancel(self, order):
print(f"订单 {order.order_id} 已取消")
order.status = "cancelled"
return CancelledState()
class ConfirmedState(OrderState):
"""已确认状态"""
def confirm(self, order):
print(f"订单 {order.order_id} 已确认")
def ship(self, order):
print(f"订单 {order.order_id} 已发货")
order.status = "shipped"
return ShippedState()
def deliver(self, order):
print(f"订单 {order.order_id} 未发货，无法送达")
def cancel(self, order):
print(f"订单 {order.order_id} 已取消")
order.status = "cancelled"
return CancelledState()
class ShippedState(OrderState):
"""已发货状态"""
def confirm(self, order):
print(f"订单 {order.order_id} 已发货")
def ship(self, order):
print(f"订单 {order.order_id} 已发货")
def deliver(self, order):
print(f"订单 {order.order_id} 已送达")
order.status = "delivered"
return DeliveredState()
def cancel(self, order):
print(f"订单 {order.order_id} 已发货，无法取消")
class DeliveredState(OrderState):
"""已送达状态"""
def confirm(self, order):
print(f"订单 {order.order_id} 已送达")
def ship(self, order):
print(f"订单 {order.order_id} 已送达")
def deliver(self, order):
print(f"订单 {order.order_id} 已送达")
def cancel(self, order):
print(f"订单 {order.order_id} 已送达，无法取消")
class CancelledState(OrderState):
"""已取消状态"""
def confirm(self, order):
print(f"订单 {order.order_id} 已取消")
def ship(self, order):
print(f"订单 {order.order_id} 已取消")
def deliver(self, order):
print(f"订单 {order.order_id} 已取消")
def cancel(self, order):
print(f"订单 {order.order_id} 已取消")
# 扩展Order类
class OrderWithState:
"""带状态的订单"""
def __init__(self, order_id, customer, items):
self.order_id = order_id
self.customer = customer
self.items = items
self.status = "pending"
self._state = PendingState()
def set_state(self, state):
"""设置状态"""
self._state = state
def confirm(self):
"""确认订单"""
self._state = self._state.confirm(self)
def ship(self):
"""发货"""
self._state = self._state.ship(self)
def deliver(self):
"""送达"""
self._state = self._state.deliver(self)
def cancel(self):
"""取消"""
self._state = self._state.cancel(self)
# 使用
order = OrderWithState("ORD-001", "Alice", items)
order.confirm()
order.ship()
order.deliver()

3. 支付系统（策略模式 + 单例模式）

from abc import ABC, abstractmethod
class PaymentStrategy(ABC):
"""支付策略接口"""
@abstractmethod
def pay(self, amount):
pass
class CreditCardPayment(PaymentStrategy):
"""信用卡支付"""
def pay(self, amount):
print(f"💳 信用卡支付: ${amount}")
return "CC-" + str(int(amount * 100))
class PayPalPayment(PaymentStrategy):
"""PayPal支付"""
def pay(self, amount):
print(f"🅿️ PayPal支付: ${amount}")
return "PP-" + str(int(amount * 100))
class WeChatPayment(PaymentStrategy):
"""微信支付"""
def pay(self, amount):
print(f"💬 微信支付: ¥{amount * 7}")
return "WX-" + str(int(amount * 100))
class PaymentProcessor:
"""支付处理器 - 单例模式"""
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self):
if not hasattr(self, 'initialized'):
self.strategy = None
self.transaction_history = []
self.initialized = True
def set_payment_strategy(self, strategy: PaymentStrategy):
"""设置支付策略"""
self.strategy = strategy
def process_payment(self, amount):
"""处理支付"""
if not self.strategy:
raise ValueError("未设置支付策略")
transaction_id = self.strategy.pay(amount)
self.transaction_history.append({
"transaction_id": transaction_id,
"amount": amount,
"timestamp": "2024-01-01"
})
return transaction_id
def get_history(self):
"""获取交易历史"""
return self.transaction_history
# 使用
processor = PaymentProcessor()
processor.set_payment_strategy(CreditCardPayment())
transaction_id = processor.process_payment(100)
processor.set_payment_strategy(WeChatPayment())
transaction_id2 = processor.process_payment(200)
print(processor.get_history())

4. 物流跟踪（观察者模式）

from abc import ABC, abstractmethod
class Observer(ABC):
"""观察者接口"""
@abstractmethod
def update(self, order_id, status):
pass
class Subject(ABC):
"""主题接口"""
@abstractmethod
def attach(self, observer: Observer):
pass
@abstractmethod
def detach(self, observer: Observer):
pass
@abstractmethod
def notify(self):
pass
class OrderTrackingSystem(Subject):
"""订单跟踪系统"""
def __init__(self):
self._observers = []
def attach(self, observer: Observer):
self._observers.append(observer)
def detach(self, observer: Observer):
self._observers.remove(observer)
def notify(self, order_id, status):
for observer in self._observers:
observer.update(order_id, status)
class CustomerNotification(Observer):
"""客户通知"""
def update(self, order_id, status):
print(f"📧 发送邮件: 您的订单 {order_id} 状态更新为 {status}")
class LogisticsTeam(Observer):
"""物流团队"""
def update(self, order_id, status):
print(f"📦 物流更新: 订单 {order_id} {status}")
# 使用
tracking = OrderTrackingSystem()
tracking.attach(CustomerNotification())
tracking.attach(LogisticsTeam())
tracking.notify("ORD-001", "已发货")
tracking.notify("ORD-001", "运输中")
tracking.notify("ORD-001", "已送达")

5. 库存管理（单例模式 + 外观模式）

class InventoryManager:
"""库存管理器 - 单例模式"""
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self):
if not hasattr(self, 'initialized'):
self.inventory = {}
self.initialized = True
def add_stock(self, product_id, quantity):
"""添加库存"""
if product_id not in self.inventory:
self.inventory[product_id] = 0
self.inventory[product_id] += quantity
print(f"📦 产品 {product_id} 库存 +{quantity}")
def check_stock(self, product_id, quantity):
"""检查库存"""
available = self.inventory.get(product_id, 0)
return available >= quantity
def reduce_stock(self, product_id, quantity):
"""减少库存"""
if self.check_stock(product_id, quantity):
self.inventory[product_id] -= quantity
print(f"📦 产品 {product_id} 库存 -{quantity}")
return True
print(f"❌ 产品 {product_id} 库存不足")
return False
class WarehouseFacade:
"""仓库外观 - 外观模式"""
def __init__(self):
self.inventory = InventoryManager()
def process_order(self, items):
"""处理订单（简化接口）"""
print("📋 处理订单...")
# 检查库存
for item in items:
if not self.inventory.check_stock(item["product_id"], item["quantity"]):
print(f"❌ 产品 {item['product_id']} 库存不足")
return False
# 减少库存
for item in items:
self.inventory.reduce_stock(item["product_id"], item["quantity"])
print("✅ 订单处理完成")
return True
# 使用
warehouse = WarehouseFacade()
warehouse.inventory.add_stock("PROD-001", 100)
warehouse.inventory.add_stock("PROD-002", 50)
order_items = [
{"product_id": "PROD-001", "quantity": 10},
{"product_id": "PROD-002", "quantity": 5}
]
warehouse.process_order(order_items)

案例2：游戏系统

项目背景

构建一个简单的RPG游戏系统，包含： - 角色创建和状态管理 - 技能系统 - 战斗系统 - 装备系统

核心代码实现

1. 角色创建（工厂方法模式 + 建造者模式）

from abc import ABC, abstractmethod
class Character(ABC):
"""角色抽象类"""
def __init__(self, name, hp, attack, defense):
self.name = name
self.hp = hp
self.attack = attack
self.defense = defense
self.equipment = {}
@abstractmethod
def special_attack(self):
pass
class Warrior(Character):
"""战士"""
def special_attack(self):
print(f"{self.name} 使用技能: 猛烈一击！")
return self.attack * 2
class Mage(Character):
"""法师"""
def special_attack(self):
print(f"{self.name} 使用技能: 火球术！")
return self.attack * 1.5
class CharacterFactory:
"""角色工厂"""
@staticmethod
def create_character(character_type, name):
"""创建角色"""
if character_type == "warrior":
return Warrior(name, hp=100, attack=20, defense=10)
elif character_type == "mage":
return Mage(name, hp=60, attack=30, defense=5)
else:
raise ValueError(f"未知角色类型: {character_type}")
class CharacterBuilder:
"""角色建造者"""
def __init__(self, name):
self.name = name
self.hp = 100
self.attack = 10
self.defense = 10
self.equipment = {}
def set_hp(self, hp):
self.hp = hp
return self
def set_attack(self, attack):
self.attack = attack
return self
def set_defense(self, defense):
self.defense = defense
return self
def add_equipment(self, slot, item):
self.equipment[slot] = item
return self
def build(self):
character = Character(self.name, self.hp, self.attack, self.defense)
character.equipment = self.equipment
return character
# 使用
warrior = CharacterFactory.create_character("warrior", "亚瑟")
mage = CharacterFactory.create_character("mage", "梅林")
custom_char = (CharacterBuilder("兰斯洛特")
.set_hp(120)
.set_attack(25)
.set_defense(15)
.add_equipment("weapon", "圣剑")
.add_equipment("armor", "圣甲")
.build())

2. 战斗系统（策略模式 + 状态模式）

class BattleStrategy(ABC):
"""战斗策略接口"""
@abstractmethod
def attack(self, attacker, defender):
pass
class NormalAttack(BattleStrategy):
"""普通攻击"""
def attack(self, attacker, defender):
damage = max(1, attacker.attack - defender.defense)
defender.hp -= damage
print(f"{attacker.name} 攻击 {defender.name}，造成 {damage} 点伤害")
class CriticalHit(BattleStrategy):
"""暴击"""
def attack(self, attacker, defender):
damage = max(1, int((attacker.attack - defender.defense) * 2))
defender.hp -= damage
print(f"{attacker.name} 暴击 {defender.name}，造成 {damage} 点伤害！")
class CharacterState:
"""角色状态接口"""
def move(self, character):
pass
def attack(self, character):
pass
class NormalState(CharacterState):
"""正常状态"""
def move(self, character):
print(f"{character.name} 正常移动")
def attack(self, character):
print(f"{character.name} 正常攻击")
class PoisonedState(CharacterState):
"""中毒状态"""
def move(self, character):
print(f"{character.name} 中毒，移动缓慢")
character.hp -= 2
def attack(self, character):
print(f"{character.name} 中毒攻击，攻击力下降")
character.hp -= 2
# 使用
warrior = CharacterFactory.create_character("warrior", "亚瑟")
mage = CharacterFactory.create_character("mage", "梅林")
# 设置战斗策略
warrior.strategy = NormalAttack()
warrior.strategy.attack(warrior, mage)
# 设置状态
warrior.state = PoisonedState()
warrior.state.move(warrior)

总结

模式组合原则

1. 不要过度设计 - 简单问题不需要复杂模式
2. 优先组合而非继承 - 组合更灵活
3. 利用Python特性 - 简化模式实现
4. YAGNI原则 - 不要添加不需要的功能

常见模式组合

本章要点

• ✅ 实际项目经常组合使用多个模式
• ✅ 订单系统展示了工厂、状态、策略、观察者的组合
• ✅ 游戏系统展示了工厂、建造者、策略、状态的组合
• ✅ 不要过度设计，根据需求选择模式
• ✅ 优先组合而非继承
• ✅ 利用Python语言特性简化实现

下一步：反模式与重构 🚀