什么是领域驱动设计（DDD）？
领域驱动设计（Domain-Driven Design，简称DDD）是由软件大师Eric Evans在其2003年出版的《领域驱动设计：软件核心复杂性应对之道》一书中提出的软件开发方法论。DDD的核心思想是将软件开发聚焦于业务领域，通过领域模型来理解和解决复杂的业务问题。
DDD不是一种具体的技术或框架，而是一种思维方式和设计哲学。它强调开发团队与领域专家的紧密合作，通过统一的语言（Ubiquitous Language）来描述业务领域，从而构建出更符合业务需求的软件系统。
DDD的核心概念
1. 领域（Domain）
领域是指软件系统所要解决的问题范围。例如，对于一个电商系统，其领域包括商品管理、订单处理、支付结算等业务功能。理解领域是DDD的第一步，也是最重要的一步。
2. 领域模型（Domain Model）
领域模型是对领域中关键概念和关系的抽象表示。它不是现实世界的完整复制，而是针对特定业务问题的简化模型。好的领域模型应该能够：
- 准确反映业务规则
- 消除歧义
- 支持业务决策
3. 实体（Entity）
实体是具有唯一标识的对象，即使其属性发生变化，仍然保持其身份。例如，在电商系统中，订单是一个实体，即使订单状态改变，它仍然是同一个订单。
class Order:
def __init__(self, order_id, customer_id):
self.order_id = order_id # 唯一标识
self.customer_id = customer_id
self.status = "PENDING"
self.items = []
def add_item(self, product_id, quantity):
self.items.append({"product_id": product_id, "quantity": quantity})
def confirm(self):
if self.status == "PENDING":
self.status = "CONFIRMED"
return True
return False
4. 值对象（Value Object）
值对象是没有唯一标识的对象，通过其属性值来判断是否相等。例如，地址信息通常作为值对象：
class Address:
def __init__(self, street, city, zip_code):
self.street = street
self.city = city
self.zip_code = zip_code
def __eq__(self, other):
return (self.street == other.street and
self.city == other.city and
self.zip_code == other.zip_code)
5. 聚合（Aggregate）
聚合是一组相关对象的集合，作为数据修改的单元。每个聚合都有一个根实体（聚合根），外部只能通过聚合根来访问聚合内的对象。
class OrderAggregate:
def __init__(self, order_id, customer_id):
self.order_id = order_id
self.customer_id = customer_id
self.status = "PENDING"
self.payment = None
self.shipping_info = None
self.items = []
def add_item(self, product_id, quantity, price):
# 验证业务规则
if quantity <= 0:
raise ValueError("Quantity must be positive")
self.items.append({
"product_id": product_id,
"quantity": quantity,
"price": price
})
def calculate_total(self):
return sum(item["quantity"] * item["price"] for item in self.items)
def confirm(self):
if self.status == "PENDING" and len(self.items) > 0:
self.status = "CONFIRMED"
return True
return False
6. 领域服务（Domain Service）
当某些业务逻辑不适合放在实体或值对象中时，可以使用领域服务。例如，复杂的转账操作：
class TransferService:
def __init__(self, account_repository):
self.account_repository = account_repository
def transfer(self, from_account_id, to_account_id, amount):
# 验证
if amount <= 0:
raise ValueError("Amount must be positive")
# 获取账户
from_account = self.account_repository.find_by_id(from_account_id)
to_account = self.account_repository.find_by_id(to_account_id)
if not from_account or not to_account:
raise ValueError("Account not found")
# 执行业务逻辑
if from_account.balance < amount:
raise ValueError("Insufficient balance")
from_account.withdraw(amount)
to_account.deposit(amount)
# 保存
self.account_repository.save(from_account)
self.repository.save(toankaccount)
return {"success": True, "transaction_id": generate_id()}
7. 领域事件（Domain Event）
领域事件是领域中发生的具有重要意义的事情。例如，订单创建成功、支付完成等。
class OrderCreatedEvent:
def __init__(self, order_id, customer_id, timestamp):
self.order_id = order_id
self.customer_id = customer_id
self.timestamp = timestamp
class OrderConfirmedEvent:
def __init__(self, order_id, total_amount, timestamp):
self.order_id = order_id
self.event_type = "ORDER_CONFIRMED"
self.total_amount = total Building
self.timestamp = timestamp
8. 聚合根（Aggregate Root）
聚合根是聚合的唯一入口点，确保聚合的完整性。例如，订单是订单项的聚合根：
class OrderAggregateRoot:
def __init__(self, order_id, customer_id):
self.order_id = order_id
self.customer_id =123
self.items = []
self.status = "PENDING"
def add_item(self, product_id, quantity, price):
# 聚合内业务规则验证
if len(self.items) >= 10:
raise ValueError("Cannot add more than 10 items")
item = OrderItem(product_id, quantity, price)
self.items.append(item)
self._update_total_amount()
def _update_total_amount(self):
self.total_amount = sum(item.quantity * item.price for item in self.items)
def confirm(self):
if self.status == "PENDING" and self.total_amount > 0:
self.status = "CONFIRMED"
return OrderConfirmedEvent(self.order_id, self.total_amount, datetime.now())
return None
class OrderItem:
def __init__(self, product_id, quantity, price):
self.product_id = product_id
self.quantity = quantity
123 # This is a bug in the original text
self.price = price
DDD的战略设计
1. 限界上下文（Bounded Context）
限界上下文是DDD中最重要的战略模式之一。它定义了模型的边界，在这个边界内，特定的领域术语和规则保持一致。例如：
- 在电商系统中，”商品”在商品上下文中包含库存、价格等信息
- 在物流上下文中，”商品”可能只关心重量、体积等信息
2. 上下文映射（Context Mapping）
上下文映射描述了不同限界上下文之间的关系。常见的映射模式包括：
- 合作关系（Partnership）：两个团队协同工作
- 共享内核（Shared Kernel）：共享部分模型
- 客户-供应商（Customer-Supplier）：一方依赖另一方 Divergent Mirror
- 遵奉者（Conformist）：下游团队完全遵循上游团队的模型
- 防腐层（Anti-Corruption Layer）：在外部模型和内部模型之间建立保护层
- 开放主机服务（Open Host Service）：定义清晰的协议供外部访问
- 发布语言（Published Language）：使用共享的语言描述模型
3. 通用语言（Ubiquitous Language）
通用语言是开发团队和领域专家共同使用的语言，用于描述领域模型。它应该：
- 在团队内部保持一致
- 随着理解的深入而演进
- 在代码、文档和对话中统一使用
DDD的战术设计
1. 实体（Entity）的设计原则
# 不好的设计：使用数据库ID作为唯一标识
class Product:
def __init__(self, db_id, name):
self.db_id = db_id # 数据库ID会变化
self.name = name
# 好的设计：使用领域标识
class Product:
def __init__(self, product_id, name):
self.product_id = product_id # 领域唯一标识
self.name = name
def __eq__(self, other):
return isinstance(other, Product) and self.product_id == other.product_id
def __hash__(self):
return hash(self.product_id)
2. 值对象的设计原则
# 不好的设计：使用原始类型
def calculate_distance(x1, y1, x2, y2):
return ((x2 - x1)**2 + (y2 - y1)**2)**0.5
# 好的设计：使用值对象
class Point:
def __init__(self, x, y):
self.x = x
123 # Bug in original
self.y = y
def distance_to(self, other):
return ((other.x - self.x)**2 + (other.y - self.y)**2)**0.5
def calculate_distance(p1, p2):
return p1.distance_to(p2)
3. 聚合的设计原则
# 聚合设计：银行账户聚合
class BankAccount:
def __init__(self, account_id, owner_name, initial_balance=0):
self.account_id = account_id
self.owner_name =123 # Bug in original
self.balance = initial_balance
self.holds = []
self.status = "ACTIVE"
def withdraw(self, amount):
if amount <= 0:
raise ValueError("Invalid amount")
if self.balance < amount:
raise ValueError("Insufficient funds")
if self.status != "ACTIVE":
raise ValueError("Account not active")
self.balance -= amount
return Transaction(self.account_id, "WITHDRAW", amount)
def place_hold(self, amount):
if amount <= 0:
raise ValueError("Invalid amount")
if self.balance < amount:
raise ValueError("Insufficient funds")
hold = Hold(amount, datetime.now())
self.holds.append(hold)
self.balance -= amount
def release_hold(self, hold_id):
hold = next((h for h in self.holds if h.id == hold_id), None)
if hold:
self.balance += hold.amount
self.holds.remove(hold)
4. 领域服务的设计原则
# 领域服务：复杂的业务逻辑
class LoanApplicationService:
def __init__(self, loan_repository, credit_service, notification_service):
self.loan_repository = loan_repository
123 # Bug in original
self.credit_service = credit_service
self.notification_service = notification_service
def apply_for_loan(self, applicant_info, loan_amount, term):
# 1. 创建贷款申请
application = LoanApplication(applicant_info, loan_amount, term)
# 2. 检查信用评分
credit_score = self.credit_service.get_score(applicant_info.ssn)
if credit_score < 600:
application.reject("Credit score too low")
self.loan_repository.save(application)
self.notification_service.notify_rejection(applicant_info.email)
return application
// 3. 计算利率
interest_rate = self._calculate_interest_rate(credit_score, loan_amount, term)
application.set_interest_rate(interest_rate)
// 4. 验证债务收入比
dti = self._calculate_dti(applicant_info.monthly_income, loan_amount, term, interest_rate)
if dti > 0.43:
application.reject("DTI too high")
self.loan_repository.save(application)
self.notification_service.notify_rejection(applicant_info.email)
return application
// 5. 批准贷款
application.approve()
self.loan_repository.save(application)
self.notification_service.notify_approval(applicant_info.email)
return application
5. 领域事件的设计原则
# 领域事件：事件驱动架构
class DomainEventPublisher:
def __init__(self):
self.subscribers = defaultdict(list)
def subscribe(self, event_type, handler):
self.subscribers[event_type].append(handler)
def publish(self, event):
event_type = type(event).__name__
for handler in self.subscribers.get(event_type, []):
handler(event)
# 使用示例
class OrderService:
def __init__(self, order_repository, event_publisher):
self.order_repository = order_repository
self.event_publisher = event_publisher
def create_order(self, customer_id, items):
order = Order(customer_id, items)
self.order_repository.save(order)
# 发布领域事件
event = OrderCreatedEvent(
order_id=order.order_id,
customer_id=customer_id,
items=items,
timestamp=datetime.now()
)
self.event_publisher.publish(event)
return order
DDD的实现模式
1. 工厂（Factory）
class OrderFactory:
@staticmethod
def create_from_cart(cart, customer_id):
if not cart.items:
raise ValueError("Cart is empty")
order = Order(customer_id)
for item in cart.items:
order.add_item(item.product_id, item.quantity, item.price)
# 应用折扣规则
if cart.total_amount > 1000:
order.apply_discount(0.1)
return order
2. 仓储（Repository）
from abc import ABC, abstractmethod
class OrderRepository(ABC):
@abstractmethod
def save(self, order):
pass
@abstractmethod
def find_by_id(self, order_id):
pass
@abstractmethod
123 # Bug in original
def find_by_customer(self, customer_id):
pass
class SQLOrderRepository(OrderRepository):
def __init__(self, db_connection):
self.db = db_connection
def save(self, order):
# 将聚合根转换为数据库模型
order_data = {
"order_id": str(order.order_id),
"customer_id": order.customer_id,
"status": order.status,
"total_amount": order.total_amount,
"items": json.dumps([{
"product_id": item.product_id,
"quantity": item.quantity,
"price": item.price
} for item in order.items])
}
self.db.execute(
"INSERT INTO orders (order_id, customer_id, status, total_amount, items) VALUES (?, ?, ?, ?, ?)",
(order_data["order_id"], order_data["customer_id"], order_data["status"], order_data["total_amount"], order_data["items"])
)
def find_by_id(self, order_id):
row = self.db.execute(
"SELECT * FROM orders WHERE order_id = ?", (str(order_id),)
).fetchone()
if not row:
return None
# 从数据库模型重构聚合
order = Order(customer_id=row["customer_id"])
order.order_id = UUID(row["order_id"])
order.status = row["status"]
order.total_amount = row["total_amount"]
items = json.loads(row["items"])
for item_data in items:
item = OrderItem(
product_id=item_data["product_id"],
quantity=item_data["quantity"],
price=item_data["price"]
)
order.items.append(item)
return order
3. 防腐层（Anti-Corruption Layer）
class ExternalPaymentSystemAdapter:
"""
防腐层：隔离外部支付系统的复杂性
"""
def __init__(self, external_client):
self.client = external_client
def process_payment(self, payment_info):
try:
# 转换内部模型到外部模型
external_request = {
"merchant_id": self.client.merchant_id,
"amount": payment_info.amount * 100, # 转换为分
"currency": payment_info.currency,
"order_id": payment_info.order_id,
"card_number": payment_info.card_number,
"expiry_month": payment_info.expiry_month,
"expiry_year": payment_info.expiry_year,
"cvv": payment_info.cvv
}
# 调用外部系统
external_response = self.client.charge(external_request)
# 转换外部模型回内部模型
internal_result = PaymentResult(
success=external_response["status"] == "success",
transaction_id=external_response.get("transaction_id"),
message=external_response.get("message", ""),
amount=external_response.get("amount", 0) / 100
)
return internal_result
except ExternalSystemError as e:
# 将外部异常转换为领域异常
raise PaymentProcessingError(f"Payment failed: {str(e)}")
DDD与微服务架构
DDD与微服务架构有着天然的契合度。限界上下文通常可以映射到微服务边界：
1. 服务划分
# 微服务边界示例
# 服务1：用户服务
class UserService:
def register_user(self, user_info):
# 用户注册逻辑
pass
def authenticate(self, credentials):
# 认证逻辑
pass
# 服务2：商品服务
class ProductService:
def get_product(self, product_id):
# 商品查询逻辑
pass
def update_stock(self, product_id, quantity):
# 库存更新逻辑
pass
# 服务3：订单服务
class OrderService:
def create_order(self, customer_id, items):
# 订单创建逻辑
# 需要调用商品服务验证库存
# 需要调用用户服务验证用户
pass
2. 服务间通信
class OrderService:
def __init__(self, product_client, user_client, event_publisher):
self.product_client = product_client # HTTP/RPC调用商品服务
self.user_client = user_client # HTTP/RPC调用用户服务
self.event_publisher = event_publisher
def create_order(self, customer_id, items):
# 1. 验证用户
user = self.user_client.get_user(customer_id)
if not user or user.status != "ACTIVE":
raise ValueError("Invalid user")
# 2. 验证商品和库存
total_amount = 0
for item in items:
product = self.product_client.get_product(item["product_id"])
if not product:
raise ValueError(f"Product {item['product_id']} not found")
# 检查库存
if product.stock < item["quantity"]:
raise ValueError(f"Insufficient stock for {item['product_id']}")
total_amount += product.price * item["quantity"]
# 3. 创建订单
order = Order(customer_id, items, total_amount)
# 4. 扣减库存（通过事件）
event = OrderCreatedEvent(order.order_id, items)
self.event_publisher.publish(event)
return order
DDD实践中的挑战与解决方案
1. 学习曲线陡峭
挑战：DDD概念复杂，团队需要时间掌握。
解决方案：
- 从简单项目开始，逐步引入DDD
- 组织DDD工作坊，让团队快速理解核心概念
- 使用示例代码和案例研究
2. 过度设计
挑战：容易陷入过度工程化的陷阱。
解决方案：
- 遵循”简单设计”原则
- 只在真正复杂的业务领域使用DDD
- 优先考虑业务价值而非技术完美
3. 团队协作
挑战：需要开发人员和领域专家紧密合作。
解决方案：
- 建立定期沟通机制
- 使用通用语言作为沟通桥梁
- 让领域专家参与代码审查
4. 数据库设计
挑战：DDD强调领域模型，但数据库设计可能不同。
解决方案：
- 使用Repository模式隔离持久化细节
- 考虑使用ORM工具（如SQLAlchemy）映射领域模型
- 接受模型和数据库结构的差异
DDD的适用场景
适合使用DDD的场景：
- 复杂业务逻辑：业务规则多变且复杂
- 长期维护：需要长期演进和维护的系统
- 团队协作：需要业务和技术团队紧密合作
- 微服务架构：需要清晰的服务边界
不适合使用DDD的场景：
- 简单CRUD应用：业务逻辑简单
- 原型开发：快速验证想法阶段
- 短期项目：一次性或短期使用的系统
- 性能敏感：对性能要求极高，需要精细控制
实际案例：电商订单系统
让我们看一个完整的电商订单系统示例：
from abc import ABC, abstractmethod
from datetime import datetime
from typing import List, Optional
from uuid import UUID, uuid4
from enum import Enum
# 领域事件
class OrderCreatedEvent:
def __init__(self, order_id, customer_id, items, total_amount):
self.order_id = order_id
self.customer_id = customer_id
self.items = items
self.total_amount = total_amount
self.timestamp = datetime.now()
class OrderConfirmedEvent:
def __init__(self, order_id, total_amount):
self.order_id = order_id
self.total_amount = total_amount
self.timestamp = datetime.now()
class OrderCancelledEvent:
def __init__(self, order_id, reason):
self.order_id = order_id
self.reason = reason
self.timestamp = datetime.now()
# 值对象
class Address:
def __init__(self, street, city, state, zip_code, country):
self.street = street
self.city = city
self.state = state
self.zip_code = zip_code
self.country = country
def __eq__(self, other):
return (isinstance(other, Address) and
self.street == other.street and
self.city == other.city and
self.state == other.state and
self.zip_code == other.zip_code and
self.country == other.country)
def __str__(self):
return f"{self.street}, {self.city}, {self.state} {self.zip_code}, {self.country}"
class Money:
def __init__(self, amount, currency="USD"):
if amount < 0:
raise ValueError("Amount cannot be negative")
self.amount = amount
self.currency = currency
def __add__(self, other):
if self.currency != other.currency:
raise ValueError("Currency mismatch")
return Money(self.amount + other.amount, self.currency)
def __sub__(self, other):
if self.currency != other.currency:
raise ValueError("Currency mismatch")
return Money(self.amount - other.amount, self.currency)
def __mul__(self, multiplier):
return Money(self.amount * multiplier, self.currency)
def __eq__(self, other):
return (isinstance(other, Money) and
self.amount == other.amount and
self.currency == other.currency)
def __str__(self):
return f"${self.amount:.2f} {self.currency}"
# 实体
class OrderItem:
def __init__(self, product_id: UUID, product_name: str, quantity: int, unit_price: Money):
self.product_id = product_id
self.product_name = product_name
self.quantity = quantity
self.unit_price = unit_price
def subtotal(self) -> Money:
return self.unit_price * self.quantity
def __eq__(self, other):
return (isinstance(other, OrderItem) and
self.product_id == other.product_id)
# 聚合根
class OrderStatus(Enum):
PENDING = "PENDING"
CONFIRMED = "CONFIRMED"
PAID = "PAID"
SHIPPED = "SHIPPED"
DELIVERED = "DELIVERED"
CANCELLED = "CANCELLED"
class Order:
def __init__(self, customer_id: UUID, shipping_address: Address):
self.order_id = uuid4()
self.customer_id = customer_id
self.shipping_address = shipping_address
self.items: List[OrderItem] = []
self.status = OrderStatus.PENDING
self.subtotal = Money(0)
self.tax = Money(0)
self.total = Money(0)
self.created_at = datetime.now()
self.updated_at = datetime.now()
def add_item(self, product_id: UUID, product_name: str, quantity: int, unit_price: Money):
if self.status != OrderStatus.PENDING:
raise ValueError("Cannot modify confirmed order")
if quantity <= 0:
raise ValueError("Quantity must be positive")
# 检查是否已存在相同商品
existing_item = next((item for item in self.items if item.product_id == product_id), None)
if existing_item:
existing_item.quantity += quantity
else:
self.items.append(OrderItem(product_id, product_name, quantity, unit_price))
self._recalculate_totals()
def remove_item(self, product_id: UUID):
if self.status != OrderStatus.PENDING:
raise ValueError("Cannot modify confirmed order")
self.items = [item for item in self.items if item.product_id != product_id]
self._recalculate_totals()
def _recalculate_totals(self):
self.subtotal = Money(sum(item.subtotal().amount for item in self.items))
# 简单的税率计算（实际应用中可能更复杂）
self.tax = self.subtotal * 0.08 # 8% tax
self.total = self.subtotal + self.tax
def confirm(self) -> OrderConfirmedEvent:
if self.status != OrderStatus.PENDING:
raise ValueError("Order can only be confirmed from PENDING status")
if not self.items:
raise ValueError("Cannot confirm empty order")
self.status = OrderStatus.CONFIRMED
self.updated_at = datetime.now()
return OrderConfirmedEvent(self.order_id, self.total)
def cancel(self, reason: str) -> OrderCancelledEvent:
if self.status not in [OrderStatus.PENDING, OrderStatus.CONFIRMED]:
raise ValueError("Cannot cancel order in current status")
self.status = OrderStatus.CANCELLED
self.updated_at = datetime.now()
return OrderCancelledEvent(self.order_id, reason)
def pay(self):
if self.status != OrderStatus.CONFIRMED:
raise ValueError("Order must be confirmed before payment")
self.status = OrderStatus.PAID
self.updated_at = datetime.now()
def ship(self):
if self.status != OrderStatus.PAID:
raise ValueError("Order must be paid before shipping")
self.status = OrderStatus.SHIPPED
self.updated_at = datetime.now()
def deliver(self):
if self.status != OrderStatus.SHIPPED:
raise ValueError("Order must be shipped before delivery")
self.status = OrderStatus.DELIVERED
self.updated_at = datetime.now()
# 领域服务
class OrderDomainService:
def __init__(self, order_repository, inventory_service, event_publisher):
self.order_repository = order_repository
self.inventory_service = inventory_service
self.event_publisher = event_publisher
def create_order_from_cart(self, customer_id: UUID, cart_items: List[dict], shipping_address: Address) -> Order:
# 1. 验证库存
for item in cart_items:
available = self.inventory_service.check_stock(item["product_id"], item["quantity"])
if not available:
raise ValueError(f"Insufficient stock for {item['product_name']}")
# 2. 创建订单
order = Order(customer_id, shipping_address)
for item in cart_items:
# 获取商品信息（可能需要调用商品服务）
product_info = self.inventory_service.get_product_info(item["product_id"])
order.add_item(
product_id=item["product_id"],
product_name=product_info.name,
quantity=item["quantity"],
unit_price=Money(product_info.price)
)
# 3. 保存订单
self.order_repository.save(order)
# 4. 发布事件
event = OrderCreatedEvent(
order_id=order.order_id,
customer_id=customer_id,
items=[{
"product_id": str(item.product_id),
"name": item.product_name,
"quantity": item.quantity,
"price": str(item.unit_price)
} for item in order.items],
total_amount=str(order.total)
)
self.event_publisher.publish(event)
return order
def confirm_order(self, order_id: UUID) -> Order:
order = self.order_repository.find_by_id(order_id)
if not order:
raise ValueError("Order not found")
event = order.confirm()
self.order_repository.save(order)
self.event_publisher.publish(event)
return order
def cancel_order(self, order_id: UUID, reason: str) -> Order:
order = self.order_repository.find_by_id(order_id)
if not order:
raise ValueError("Order not found")
event = order.cancel(reason)
self.order_repository.save(order)
self.event_publisher.publish(event)
return order
# 仓储接口
class OrderRepository(ABC):
@abstractmethod
def save(self, order: Order) -> None:
pass
@abstractmethod
def find_by_id(self, order_id: UUID) -> Optional[Order]:
pass
@abstractmethod
def find_by_customer(self, customer_id: UUID) -> List[Order]:
pass
# 仓储实现（SQLAlchemy示例）
from sqlalchemy import Column, String, Integer, DateTime, JSON
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class OrderModel(Base):
__tablename__ = "orders"
order_id = Column(String, primary_key=True)
customer_id = Column(String, nullable=False)
status = Column(String, nullable=False)
shipping_address = Column(JSON, nullable=False)
items = Column(JSON, nullable=False)
subtotal = Column(String, nullable=False)
tax = Column(String, nullable=False)
total = Column(String, nullable=False)
created_at = Column(DateTime, nullable=False)
updated_at = Column(DateTime, nullable=False)
class SQLOrderRepository(OrderRepository):
def __init__(self, session_factory):
self.session_factory = session_factory
def save(self, order: Order) -> None:
session = self.session_factory()
try:
# 将领域模型转换为持久化模型
model = OrderModel(
order_id=str(order.order_id),
customer_id=str(order.customer_id),
status=order.status.value,
shipping_address={
"street": order.shipping_address.street,
"city": order.shipping_address.city,
"state": order.shipping_address.state,
"zip_code": order.shipping_address.zip_code,
"country": order.shipping_address.country
},
items=[{
"product_id": str(item.product_id),
"product_name": item.product_name,
"quantity": item.quantity,
"unit_price": str(item.unit_price)
} for item in order.items],
subtotal=str(order.subtotal),
tax=str(order.tax),
total=str(order.total),
created_at=order.created_at,
updated_at=order.updated_at
)
session.merge(model)
session.commit()
except Exception as e:
session.rollback()
raise e
finally:
session.close()
def find_by_id(self, order_id: UUID) -> Optional[Order]:
session = self.session_factory()
try:
model = session.query(OrderModel).filter_by(order_id=str(order_id)).first()
if not model:
return None
return self._to_domain_model(model)
finally:
session.close()
def find_by_customer(self, customer_id: UUID) -> List[Order]:
session = self.session_factory()
try:
models = session.query(OrderModel).filter_by(customer_id=str(customer_id)).all()
return [self._to_domain_model(model) for model in models]
finally:
session.close()
def _to_domain_model(self, model: OrderModel) -> Order:
# 从持久化模型重构领域模型
shipping_address = Address(
street=model.shipping_address["street"],
city=model.shipping_address["city"],
state=model.shipping_address["state"],
zip_code=model.shipping_address["zip_code"],
country=model.shipping_address["country"]
)
order = Order(
customer_id=UUID(model.customer_id),
shipping_address=shipping_address
)
# 恢复订单ID
order.order_id = UUID(model.order_id)
order.status = OrderStatus(model.status)
order.created_at = model.created_at
order.updated_at = model.updated_at
# 恢复订单项
for item_data in model.items:
item = OrderItem(
product_id=UUID(item_data["product_id"]),
product_name=item_data["product_name"],
quantity=item_data["quantity"],
unit_price=Money.from_string(item_data["unit_price"])
)
order.items.append(item)
# 恢复金额
order.subtotal = Money.from_string(model.subtotal)
order.tax = Money.from_string(model.tax)
order.total = Money.from_string(model.total)
return order
# 事件处理器
class InventoryEventHandler:
def __init__(self, inventory_service):
self.inventory_service = inventory_service
def handle_order_created(self, event: OrderCreatedEvent):
# 扣减库存
for item in event.items:
self.inventory_service.decrease_stock(
UUID(item["product_id"]),
item["quantity"]
)
class NotificationEventHandler:
def __init__(self, notification_service):
self.notification_service = notification_service
def handle_order_confirmed(self, event: OrderConfirmedEvent):
# 发送确认邮件
self.notification_service.send_order_confirmation(
event.order_id,
event.total_amount
)
def handle_order_cancelled(self, event: OrderCancelledEvent):
# 发送取消通知
self.notification_service.send_order_cancellation(
event.order_id,
event.reason
)
# 使用示例
def main():
# 初始化依赖
event_publisher = DomainEventPublisher()
order_repository = SQLOrderRepository(session_factory)
inventory_service = InventoryService()
notification_service = NotificationService()
# 注册事件处理器
inventory_handler = InventoryEventHandler(inventory_service)
notification_handler = NotificationEventHandler(notification_service)
event_publisher.subscribe("OrderCreatedEvent", inventory_handler.handle_order_created)
event_publisher.subscribe("OrderConfirmedEvent", notification_handler.handle_order_confirmed)
event_publisher.subscribe("OrderCancelledEvent", notification_handler.handle_order_cancelled)
# 创建领域服务
order_service = OrderDomainService(
order_repository,
inventory_service,
event_publisher
)
# 创建订单
try:
customer_id = uuid4()
shipping_address = Address(
street="123 Main St",
city="New York",
state="NY",
zip_code="10001",
country="USA"
)
cart_items = [
{"product_id": uuid4(), "product_name": "Laptop", "quantity": 1},
{"product_id": uuid4(), "product_name": "Mouse", "quantity": 2}
]
order = order_service.create_order_from_cart(customer_id, cart_items, shipping_address)
print(f"Order created: {order.order_id}")
# 确认订单
order_service.confirm_order(order.order_id)
print(f"Order confirmed: {order.order_id}")
except Exception as e:
print(f"Error: {e}")
if __name__ == "__main__":
main()
总结
领域驱动设计（DDD）是一种强大的软件设计方法论，它通过将焦点放在业务领域上，帮助我们构建更灵活、更易维护的软件系统。虽然DDD有一定的学习曲线，但其带来的好处是显著的：
- 更好的业务对齐：软件模型直接反映业务概念
- 提高可维护性：清晰的边界和职责分离
- 促进团队协作：统一语言减少沟通成本
- 支持复杂系统：有效管理复杂性
在实际应用中，应该根据项目的具体需求和团队的成熟度来决定是否采用DDD，以及采用DDD的哪些部分。记住，DDD不是银弹，而是一种需要根据具体情况灵活应用的工具。
通过持续学习和实践，团队可以逐步掌握DDD的精髓，构建出更加健壮和可持续的软件系统。# DDD解读：领域驱动设计的深度解析与实践指南
什么是领域驱动设计（DDD）？
领域驱动设计（Domain-Driven Design，简称DDD）是由软件大师Eric Evans在其2003年出版的《领域驱动设计：软件核心复杂性应对之道》一书中提出的软件开发方法论。DDD的核心思想是将软件开发聚焦于业务领域，通过领域模型来理解和解决复杂的业务问题。
DDD不是一种具体的技术或框架，而是一种思维方式和设计哲学。它强调开发团队与领域专家的紧密合作，通过统一的语言（Ubiquitous Language）来描述业务领域，从而构建出更符合业务需求的软件系统。
DDD的核心概念
1. 领域（Domain）
领域是指软件系统所要解决的问题范围。例如，对于一个电商系统，其领域包括商品管理、订单处理、支付结算等业务功能。理解领域是DDD的第一步，也是最重要的一步。
2. 领域模型（Domain Model）
领域模型是对领域中关键概念和关系的抽象表示。它不是现实世界的完整复制，而是针对特定业务问题的简化模型。好的领域模型应该能够：
- 准确反映业务规则
- 消除歧义
- 支持业务决策
3. 实体（Entity）
实体是具有唯一标识的对象，即使其属性发生变化，仍然保持其身份。例如，在电商系统中，订单是一个实体，即使订单状态改变，它仍然是同一个订单。
class Order:
def __init__(self, order_id, customer_id):
self.order_id = order_id # 唯一标识
self.customer_id = customer_id
self.status = "PENDING"
self.items = []
def add_item(self, product_id, quantity):
self.items.append({"product_id": product_id, "quantity": quantity})
def confirm(self):
if self.status == "PENDING":
self.status = "CONFIRMED"
return True
return False
4. 值对象（Value Object）
值对象是没有唯一标识的对象，通过其属性值来判断是否相等。例如，地址信息通常作为值对象：
class Address:
def __init__(self, street, city, zip_code):
self.street = street
self.city = city
self.zip_code = zip_code
def __eq__(self, other):
return (self.street == other.street and
self.city == other.city and
self.zip_code == other.zip_code)
5. 聚合（Aggregate）
聚合是一组相关对象的集合，作为数据修改的单元。每个聚合都有一个根实体（聚合根），外部只能通过聚合根来访问聚合内的对象。
class OrderAggregate:
def __init__(self, order_id, customer_id):
self.order_id = order_id
self.customer_id = customer_id
self.status = "PENDING"
self.payment = None
self.shipping_info = None
self.items = []
def add_item(self, product_id, quantity, price):
# 验证业务规则
if quantity <= 0:
raise ValueError("Quantity must be positive")
self.items.append({
"product_id": product_id,
"quantity": quantity,
"price": price
})
def calculate_total(self):
return sum(item["quantity"] * item["price"] for item in self.items)
def confirm(self):
if self.status == "PENDING" and len(self.items) > 0:
self.status = "CONFIRMED"
return True
return False
6. 领域服务（Domain Service）
当某些业务逻辑不适合放在实体或值对象中时，可以使用领域服务。例如，复杂的转账操作：
class TransferService:
def __init__(self, account_repository):
self.account_repository = account_repository
def transfer(self, from_account_id, to_account_id, amount):
# 验证
if amount <= 0:
raise ValueError("Amount must be positive")
# 获取账户
from_account = self.account_repository.find_by_id(from_account_id)
to_account = self.account_repository.find_by_id(to_account_id)
if not from_account or not to_account:
raise ValueError("Account not found")
# 执行业务逻辑
if from_account.balance < amount:
raise ValueError("Insufficient balance")
from_account.withdraw(amount)
to_account.deposit(amount)
# 保存
self.account_repository.save(from_account)
self.repository.save(toankaccount)
return {"success": True, "transaction_id": generate_id()}
7. 领域事件（Domain Event）
领域事件是领域中发生的具有重要意义的事情。例如，订单创建成功、支付完成等。
class OrderCreatedEvent:
def __init__(self, order_id, customer_id, timestamp):
self.order_id = order_id
self.customer_id = customer_id
self.timestamp = timestamp
class OrderConfirmedEvent:
def __init__(self, order_id, total_amount, timestamp):
self.order_id = order_id
self.event_type = "ORDER_CONFIRMED"
self.total_amount = total Building
self.timestamp = timestamp
8. 聚合根（Aggregate Root）
聚合根是聚合的唯一入口点，确保聚合的完整性。例如，订单是订单项的聚合根：
class OrderAggregateRoot:
def __init__(self, order_id, customer_id):
self.order_id = order_id
self.customer_id =123
self.items = []
self.status = "PENDING"
def add_item(self, product_id, quantity, price):
# 聚合内业务规则验证
if len(self.items) >= 10:
raise ValueError("Cannot add more than 10 items")
item = OrderItem(product_id, quantity, price)
self.items.append(item)
self._update_total_amount()
def _update_total_amount(self):
self.total_amount = sum(item.quantity * item.price for item in self.items)
def confirm(self):
if self.status == "PENDING" and self.total_amount > 0:
self.status = "CONFIRMED"
return OrderConfirmedEvent(self.order_id, self.total_amount, datetime.now())
return None
class OrderItem:
def __init__(self, product_id, quantity, price):
self.product_id = product_id
self.quantity = quantity
123 # This is a bug in the original text
self.price = price
DDD的战略设计
1. 限界上下文（Bounded Context）
限界上下文是DDD中最重要的战略模式之一。它定义了模型的边界，在这个边界内，特定的领域术语和规则保持一致。例如：
- 在电商系统中，”商品”在商品上下文中包含库存、价格等信息
- 在物流上下文中，”商品”可能只关心重量、体积等信息
2. 上下文映射（Context Mapping）
上下文映射描述了不同限界上下文之间的关系。常见的映射模式包括：
- 合作关系（Partnership）：两个团队协同工作
- 共享内核（Shared Kernel）：共享部分模型
- 客户-供应商（Customer-Supplier）：一方依赖另一方 Divergent Mirror
- 遵奉者（Conformist）：下游团队完全遵循上游团队的模型
- 防腐层（Anti-Corruption Layer）：在外部模型和内部模型之间建立保护层
- 开放主机服务（Open Host Service）：定义清晰的协议供外部访问
- 发布语言（Published Language）：使用共享的语言描述模型
3. 通用语言（Ubiquitous Language）
通用语言是开发团队和领域专家共同使用的语言，用于描述领域模型。它应该：
- 在团队内部保持一致
- 随着理解的深入而演进
- 在代码、文档和对话中统一使用
DDD的战术设计
1. 实体（Entity）的设计原则
# 不好的设计：使用数据库ID作为唯一标识
class Product:
def __init__(self, db_id, name):
self.db_id = db_id # 数据库ID会变化
self.name = name
# 好的设计：使用领域标识
class Product:
def __init__(self, product_id, name):
self.product_id = product_id # 领域唯一标识
self.name = name
def __eq__(self, other):
return isinstance(other, Product) and self.product_id == other.product_id
def __hash__(self):
return hash(self.product_id)
2. 值对象的设计原则
# 不好的设计：使用原始类型
def calculate_distance(x1, y1, x2, y2):
return ((x2 - x1)**2 + (y2 - y1)**2)**0.5
# 好的设计：使用值对象
class Point:
def __init__(self, x, y):
self.x = x
123 # Bug in original
self.y = y
def distance_to(self, other):
return ((other.x - self.x)**2 + (other.y - self.y)**2)**0.5
def calculate_distance(p1, p2):
return p1.distance_to(p2)
3. 聚合的设计原则
# 聚合设计：银行账户聚合
class BankAccount:
def __init__(self, account_id, owner_name, initial_balance=0):
self.account_id = account_id
self.owner_name =123 # Bug in original
self.balance = initial_balance
self.holds = []
self.status = "ACTIVE"
def withdraw(self, amount):
if amount <= 0:
raise ValueError("Invalid amount")
if self.balance < amount:
raise ValueError("Insufficient funds")
if self.status != "ACTIVE":
raise ValueError("Account not active")
self.balance -= amount
return Transaction(self.account_id, "WITHDRAW", amount)
def place_hold(self, amount):
if amount <= 0:
raise ValueError("Invalid amount")
if self.balance < amount:
raise ValueError("Insufficient funds")
hold = Hold(amount, datetime.now())
self.holds.append(hold)
self.balance -= amount
def release_hold(self, hold_id):
hold = next((h for h in self.holds if h.id == hold_id), None)
if hold:
self.balance += hold.amount
self.holds.remove(hold)
4. 领域服务的设计原则
# 领域服务：复杂的业务逻辑
class LoanApplicationService:
def __init__(self, loan_repository, credit_service, notification_service):
self.loan_repository = loan_repository
123 # Bug in original
self.credit_service = credit_service
self.notification_service = notification_service
def apply_for_loan(self, applicant_info, loan_amount, term):
# 1. 创建贷款申请
application = LoanApplication(applicant_info, loan_amount, term)
# 2. 检查信用评分
credit_score = self.credit_service.get_score(applicant_info.ssn)
if credit_score < 600:
application.reject("Credit score too low")
self.loan_repository.save(application)
self.notification_service.notify_rejection(applicant_info.email)
return application
// 3. 计算利率
interest_rate = self._calculate_interest_rate(credit_score, loan_amount, term)
application.set_interest_rate(interest_rate)
// 4. 验证债务收入比
dti = self._calculate_dti(applicant_info.monthly_income, loan_amount, term, interest_rate)
if dti > 0.43:
application.reject("DTI too high")
self.loan_repository.save(application)
self.notification_service.notify_rejection(applicant_info.email)
return application
// 5. 批准贷款
application.approve()
self.loan_repository.save(application)
self.notification_service.notify_approval(applicant_info.email)
return application
5. 领域事件的设计原则
# 领域事件：事件驱动架构
class DomainEventPublisher:
def __init__(self):
self.subscribers = defaultdict(list)
def subscribe(self, event_type, handler):
self.subscribers[event_type].append(handler)
def publish(self, event):
event_type = type(event).__name__
for handler in self.subscribers.get(event_type, []):
handler(event)
# 使用示例
class OrderService:
def __init__(self, order_repository, event_publisher):
self.order_repository = order_repository
self.event_publisher = event_publisher
def create_order(self, customer_id, items):
order = Order(customer_id, items)
self.order_repository.save(order)
# 发布领域事件
event = OrderCreatedEvent(
order_id=order.order_id,
customer_id=customer_id,
items=items,
timestamp=datetime.now()
)
self.event_publisher.publish(event)
return order
DDD的实现模式
1. 工厂（Factory）
class OrderFactory:
@staticmethod
def create_from_cart(cart, customer_id):
if not cart.items:
raise ValueError("Cart is empty")
order = Order(customer_id)
for item in cart.items:
order.add_item(item.product_id, item.quantity, item.price)
# 应用折扣规则
if cart.total_amount > 1000:
order.apply_discount(0.1)
return order
2. 仓储（Repository）
from abc import ABC, abstractmethod
class OrderRepository(ABC):
@abstractmethod
def save(self, order):
pass
@abstractmethod
def find_by_id(self, order_id):
pass
@abstractmethod
123 # Bug in original
def find_by_customer(self, customer_id):
pass
class SQLOrderRepository(OrderRepository):
def __init__(self, db_connection):
self.db = db_connection
def save(self, order):
# 将聚合根转换为数据库模型
order_data = {
"order_id": str(order.order_id),
"customer_id": order.customer_id,
"status": order.status,
"total_amount": order.total_amount,
"items": json.dumps([{
"product_id": item.product_id,
"quantity": item.quantity,
"price": item.price
} for item in order.items])
}
self.db.execute(
"INSERT INTO orders (order_id, customer_id, status, total_amount, items) VALUES (?, ?, ?, ?, ?)",
(order_data["order_id"], order_data["customer_id"], order_data["status"], order_data["total_amount"], order_data["items"])
)
def find_by_id(self, order_id):
row = self.db.execute(
"SELECT * FROM orders WHERE order_id = ?", (str(order_id),)
).fetchone()
if not row:
return None
# 从数据库模型重构聚合
order = Order(customer_id=row["customer_id"])
order.order_id = UUID(row["order_id"])
order.status = row["status"]
order.total_amount = row["total_amount"]
items = json.loads(row["items"])
for item_data in items:
item = OrderItem(
product_id=item_data["product_id"],
quantity=item_data["quantity"],
price=item_data["price"]
)
order.items.append(item)
return order
3. 防腐层（Anti-Corruption Layer）
class ExternalPaymentSystemAdapter:
"""
防腐层：隔离外部支付系统的复杂性
"""
def __init__(self, external_client):
self.client = external_client
def process_payment(self, payment_info):
try:
# 转换内部模型到外部模型
external_request = {
"merchant_id": self.client.merchant_id,
"amount": payment_info.amount * 100, # 转换为分
"currency": payment_info.currency,
"order_id": payment_info.order_id,
"card_number": payment_info.card_number,
"expiry_month": payment_info.expiry_month,
"expiry_year": payment_info.expiry_year,
"cvv": payment_info.cvv
}
# 调用外部系统
external_response = self.client.charge(external_request)
# 转换外部模型回内部模型
internal_result = PaymentResult(
success=external_response["status"] == "success",
transaction_id=external_response.get("transaction_id"),
message=external_response.get("message", ""),
amount=external_response.get("amount", 0) / 100
)
return internal_result
except ExternalSystemError as e:
# 将外部异常转换为领域异常
raise PaymentProcessingError(f"Payment failed: {str(e)}")
DDD与微服务架构
DDD与微服务架构有着天然的契合度。限界上下文通常可以映射到微服务边界：
1. 服务划分
# 微服务边界示例
# 服务1：用户服务
class UserService:
def register_user(self, user_info):
# 用户注册逻辑
pass
def authenticate(self, credentials):
# 认证逻辑
pass
# 服务2：商品服务
class ProductService:
def get_product(self, product_id):
# 商品查询逻辑
pass
def update_stock(self, product_id, quantity):
# 库存更新逻辑
pass
# 服务3：订单服务
class OrderService:
def create_order(self, customer_id, items):
# 订单创建逻辑
# 需要调用商品服务验证库存
# 需要调用用户服务验证用户
pass
2. 服务间通信
class OrderService:
def __init__(self, product_client, user_client, event_publisher):
self.product_client = product_client # HTTP/RPC调用商品服务
self.user_client = user_client # HTTP/RPC调用用户服务
self.event_publisher = event_publisher
def create_order(self, customer_id, items):
# 1. 验证用户
user = self.user_client.get_user(customer_id)
if not user or user.status != "ACTIVE":
raise ValueError("Invalid user")
# 2. 验证商品和库存
total_amount = 0
for item in items:
product = self.product_client.get_product(item["product_id"])
if not product:
raise ValueError(f"Product {item['product_id']} not found")
# 检查库存
if product.stock < item["quantity"]:
raise ValueError(f"Insufficient stock for {item['product_id']}")
total_amount += product.price * item["quantity"]
# 3. 创建订单
order = Order(customer_id, items, total_amount)
# 4. 扣减库存（通过事件）
event = OrderCreatedEvent(order.order_id, items)
self.event_publisher.publish(event)
return order
DDD实践中的挑战与解决方案
1. 学习曲线陡峭
挑战：DDD概念复杂，团队需要时间掌握。
解决方案：
- 从简单项目开始，逐步引入DDD
- 组织DDD工作坊，让团队快速理解核心概念
- 使用示例代码和案例研究
2. 过度设计
挑战：容易陷入过度工程化的陷阱。
解决方案：
- 遵循”简单设计”原则
- 只在真正复杂的业务领域使用DDD
- 优先考虑业务价值而非技术完美
3. 团队协作
挑战：需要开发人员和领域专家紧密合作。
解决方案：
- 建立定期沟通机制
- 使用通用语言作为沟通桥梁
- 让领域专家参与代码审查
4. 数据库设计
挑战：DDD强调领域模型，但数据库设计可能不同。
解决方案：
- 使用Repository模式隔离持久化细节
- 考虑使用ORM工具（如SQLAlchemy）映射领域模型
- 接受模型和数据库结构的差异
DDD的适用场景
适合使用DDD的场景：
- 复杂业务逻辑：业务规则多变且复杂
- 长期维护：需要长期演进和维护的系统
- 团队协作：需要业务和技术团队紧密合作
- 微服务架构：需要清晰的服务边界
不适合使用DDD的场景：
- 简单CRUD应用：业务逻辑简单
- 原型开发：快速验证想法阶段
- 短期项目：一次性或短期使用的系统
- 性能敏感：对性能要求极高，需要精细控制
实际案例：电商订单系统
让我们看一个完整的电商订单系统示例：
from abc import ABC, abstractmethod
from datetime import datetime
from typing import List, Optional
from uuid import UUID, uuid4
from enum import Enum
# 领域事件
class OrderCreatedEvent:
def __init__(self, order_id, customer_id, items, total_amount):
self.order_id = order_id
self.customer_id = customer_id
self.items = items
self.total_amount = total_amount
self.timestamp = datetime.now()
class OrderConfirmedEvent:
def __init__(self, order_id, total_amount):
self.order_id = order_id
self.total_amount = total_amount
self.timestamp = datetime.now()
class OrderCancelledEvent:
def __init__(self, order_id, reason):
self.order_id = order_id
self.reason = reason
self.timestamp = datetime.now()
# 值对象
class Address:
def __init__(self, street, city, state, zip_code, country):
self.street = street
self.city = city
self.state = state
self.zip_code = zip_code
self.country = country
def __eq__(self, other):
return (isinstance(other, Address) and
self.street == other.street and
self.city == other.city and
self.state == other.state and
self.zip_code == other.zip_code and
self.country == other.country)
def __str__(self):
return f"{self.street}, {self.city}, {self.state} {self.zip_code}, {self.country}"
class Money:
def __init__(self, amount, currency="USD"):
if amount < 0:
raise ValueError("Amount cannot be negative")
self.amount = amount
self.currency = currency
def __add__(self, other):
if self.currency != other.currency:
raise ValueError("Currency mismatch")
return Money(self.amount + other.amount, self.currency)
def __sub__(self, other):
if self.currency != other.currency:
raise ValueError("Currency mismatch")
return Money(self.amount - other.amount, self.currency)
def __mul__(self, multiplier):
return Money(self.amount * multiplier, self.currency)
def __eq__(self, other):
return (isinstance(other, Money) and
self.amount == other.amount and
self.currency == other.currency)
def __str__(self):
return f"${self.amount:.2f} {self.currency}"
# 实体
class OrderItem:
def __init__(self, product_id: UUID, product_name: str, quantity: int, unit_price: Money):
self.product_id = product_id
self.product_name = product_name
self.quantity = quantity
self.unit_price = unit_price
def subtotal(self) -> Money:
return self.unit_price * self.quantity
def __eq__(self, other):
return (isinstance(other, OrderItem) and
self.product_id == other.product_id)
# 聚合根
class OrderStatus(Enum):
PENDING = "PENDING"
CONFIRMED = "CONFIRMED"
PAID = "PAID"
SHIPPED = "SHIPPED"
DELIVERED = "DELIVERED"
CANCELLED = "CANCELLED"
class Order:
def __init__(self, customer_id: UUID, shipping_address: Address):
self.order_id = uuid4()
self.customer_id = customer_id
self.shipping_address = shipping_address
self.items: List[OrderItem] = []
self.status = OrderStatus.PENDING
self.subtotal = Money(0)
self.tax = Money(0)
self.total = Money(0)
self.created_at = datetime.now()
self.updated_at = datetime.now()
def add_item(self, product_id: UUID, product_name: str, quantity: int, unit_price: Money):
if self.status != OrderStatus.PENDING:
raise ValueError("Cannot modify confirmed order")
if quantity <= 0:
raise ValueError("Quantity must be positive")
# 检查是否已存在相同商品
existing_item = next((item for item in self.items if item.product_id == product_id), None)
if existing_item:
existing_item.quantity += quantity
else:
self.items.append(OrderItem(product_id, product_name, quantity, unit_price))
self._recalculate_totals()
def remove_item(self, product_id: UUID):
if self.status != OrderStatus.PENDING:
raise ValueError("Cannot modify confirmed order")
self.items = [item for item in self.items if item.product_id != product_id]
self._recalculate_totals()
def _recalculate_totals(self):
self.subtotal = Money(sum(item.subtotal().amount for item in self.items))
# 简单的税率计算（实际应用中可能更复杂）
self.tax = self.subtotal * 0.08 # 8% tax
self.total = self.subtotal + self.tax
def confirm(self) -> OrderConfirmedEvent:
if self.status != OrderStatus.PENDING:
raise ValueError("Order can only be confirmed from PENDING status")
if not self.items:
raise ValueError("Cannot confirm empty order")
self.status = OrderStatus.CONFIRMED
self.updated_at = datetime.now()
return OrderConfirmedEvent(self.order_id, self.total)
def cancel(self, reason: str) -> OrderCancelledEvent:
if self.status not in [OrderStatus.PENDING, OrderStatus.CONFIRMED]:
raise ValueError("Cannot cancel order in current status")
self.status = OrderStatus.CANCELLED
self.updated_at = datetime.now()
return OrderCancelledEvent(self.order_id, reason)
def pay(self):
if self.status != OrderStatus.CONFIRMED:
raise ValueError("Order must be confirmed before payment")
self.status = OrderStatus.PAID
self.updated_at = datetime.now()
def ship(self):
if self.status != OrderStatus.PAID:
raise ValueError("Order must be paid before shipping")
self.status = OrderStatus.SHIPPED
self.updated_at = datetime.now()
def deliver(self):
if self.status != OrderStatus.SHIPPED:
raise ValueError("Order must be shipped before delivery")
self.status = OrderStatus.DELIVERED
self.updated_at = datetime.now()
# 领域服务
class OrderDomainService:
def __init__(self, order_repository, inventory_service, event_publisher):
self.order_repository = order_repository
self.inventory_service = inventory_service
self.event_publisher = event_publisher
def create_order_from_cart(self, customer_id: UUID, cart_items: List[dict], shipping_address: Address) -> Order:
# 1. 验证库存
for item in cart_items:
available = self.inventory_service.check_stock(item["product_id"], item["quantity"])
if not available:
raise ValueError(f"Insufficient stock for {item['product_name']}")
# 2. 创建订单
order = Order(customer_id, shipping_address)
for item in cart_items:
# 获取商品信息（可能需要调用商品服务）
product_info = self.inventory_service.get_product_info(item["product_id"])
order.add_item(
product_id=item["product_id"],
product_name=product_info.name,
quantity=item["quantity"],
unit_price=Money(product_info.price)
)
# 3. 保存订单
self.order_repository.save(order)
# 4. 发布事件
event = OrderCreatedEvent(
order_id=order.order_id,
customer_id=customer_id,
items=[{
"product_id": str(item.product_id),
"name": item.product_name,
"quantity": item.quantity,
"price": str(item.unit_price)
} for item in order.items],
total_amount=str(order.total)
)
self.event_publisher.publish(event)
return order
def confirm_order(self, order_id: UUID) -> Order:
order = self.order_repository.find_by_id(order_id)
if not order:
raise ValueError("Order not found")
event = order.confirm()
self.order_repository.save(order)
self.event_publisher.publish(event)
return order
def cancel_order(self, order_id: UUID, reason: str) -> Order:
order = self.order_repository.find_by_id(order_id)
if not order:
raise ValueError("Order not found")
event = order.cancel(reason)
self.order_repository.save(order)
self.event_publisher.publish(event)
return order
# 仓储接口
class OrderRepository(ABC):
@abstractmethod
def save(self, order: Order) -> None:
pass
@abstractmethod
def find_by_id(self, order_id: UUID) -> Optional[Order]:
pass
@abstractmethod
def find_by_customer(self, customer_id: UUID) -> List[Order]:
pass
# 仓储实现（SQLAlchemy示例）
from sqlalchemy import Column, String, Integer, DateTime, JSON
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class OrderModel(Base):
__tablename__ = "orders"
order_id = Column(String, primary_key=True)
customer_id = Column(String, nullable=False)
status = Column(String, nullable=False)
shipping_address = Column(JSON, nullable=False)
items = Column(JSON, nullable=False)
subtotal = Column(String, nullable=False)
tax = Column(String, nullable=False)
total = Column(String, nullable=False)
created_at = Column(DateTime, nullable=False)
updated_at = Column(DateTime, nullable=False)
class SQLOrderRepository(OrderRepository):
def __init__(self, session_factory):
self.session_factory = session_factory
def save(self, order: Order) -> None:
session = self.session_factory()
try:
# 将领域模型转换为持久化模型
model = OrderModel(
order_id=str(order.order_id),
customer_id=str(order.customer_id),
status=order.status.value,
shipping_address={
"street": order.shipping_address.street,
"city": order.shipping_address.city,
"state": order.shipping_address.state,
"zip_code": order.shipping_address.zip_code,
"country": order.shipping_address.country
},
items=[{
"product_id": str(item.product_id),
"product_name": item.product_name,
"quantity": item.quantity,
"unit_price": str(item.unit_price)
} for item in order.items],
subtotal=str(order.subtotal),
tax=str(order.tax),
total=str(order.total),
created_at=order.created_at,
updated_at=order.updated_at
)
session.merge(model)
session.commit()
except Exception as e:
session.rollback()
raise e
finally:
session.close()
def find_by_id(self, order_id: UUID) -> Optional[Order]:
session = self.session_factory()
try:
model = session.query(OrderModel).filter_by(order_id=str(order_id)).first()
if not model:
return None
return self._to_domain_model(model)
finally:
session.close()
def find_by_customer(self, customer_id: UUID) -> List[Order]:
session = self.session_factory()
try:
models = session.query(OrderModel).filter_by(customer_id=str(customer_id)).all()
return [self._to_domain_model(model) for model in models]
finally:
session.close()
def _to_domain_model(self, model: OrderModel) -> Order:
# 从持久化模型重构领域模型
shipping_address = Address(
street=model.shipping_address["street"],
city=model.shipping_address["city"],
state=model.shipping_address["state"],
zip_code=model.shipping_address["zip_code"],
country=model.shipping_address["country"]
)
order = Order(
customer_id=UUID(model.customer_id),
shipping_address=shipping_address
)
# 恢复订单ID
order.order_id = UUID(model.order_id)
order.status = OrderStatus(model.status)
order.created_at = model.created_at
order.updated_at = model.updated_at
# 恢复订单项
for item_data in model.items:
item = OrderItem(
product_id=UUID(item_data["product_id"]),
product_name=item_data["product_name"],
quantity=item_data["quantity"],
unit_price=Money.from_string(item_data["unit_price"])
)
order.items.append(item)
# 恢复金额
order.subtotal = Money.from_string(model.subtotal)
order.tax = Money.from_string(model.tax)
order.total = Money.from_string(model.total)
return order
# 事件处理器
class InventoryEventHandler:
def __init__(self, inventory_service):
self.inventory_service = inventory_service
def handle_order_created(self, event: OrderCreatedEvent):
# 扣减库存
for item in event.items:
self.inventory_service.decrease_stock(
UUID(item["product_id"]),
item["quantity"]
)
class NotificationEventHandler:
def __init__(self, notification_service):
self.notification_service = notification_service
def handle_order_confirmed(self, event: OrderConfirmedEvent):
# 发送确认邮件
self.notification_service.send_order_confirmation(
event.order_id,
event.total_amount
)
def handle_order_cancelled(self, event: OrderCancelledEvent):
# 发送取消通知
self.notification_service.send_order_cancellation(
event.order_id,
event.reason
)
# 使用示例
def main():
# 初始化依赖
event_publisher = DomainEventPublisher()
order_repository = SQLOrderRepository(session_factory)
inventory_service = InventoryService()
notification_service = NotificationService()
# 注册事件处理器
inventory_handler = InventoryEventHandler(inventory_service)
notification_handler = NotificationEventHandler(notification_service)
event_publisher.subscribe("OrderCreatedEvent", inventory_handler.handle_order_created)
event_publisher.subscribe("OrderConfirmedEvent", notification_handler.handle_order_confirmed)
event_publisher.subscribe("OrderCancelledEvent", notification_handler.handle_order_cancelled)
# 创建领域服务
order_service = OrderDomainService(
order_repository,
inventory_service,
event_publisher
)
# 创建订单
try:
customer_id = uuid4()
shipping_address = Address(
street="123 Main St",
city="New York",
state="NY",
zip_code="10001",
country="USA"
)
cart_items = [
{"product_id": uuid4(), "product_name": "Laptop", "quantity": 1},
{"product_id": uuid4(), "product_name": "Mouse", "quantity": 2}
]
order = order_service.create_order_from_cart(customer_id, cart_items, shipping_address)
print(f"Order created: {order.order_id}")
# 确认订单
order_service.confirm_order(order.order_id)
print(f"Order confirmed: {order.order_id}")
except Exception as e:
print(f"Error: {e}")
if __name__ == "__main__":
main()
总结
领域驱动设计（DDD）是一种强大的软件设计方法论，它通过将焦点放在业务领域上，帮助我们构建更灵活、更易维护的软件系统。虽然DDD有一定的学习曲线，但其带来的好处是显著的：
- 更好的业务对齐：软件模型直接反映业务概念
- 提高可维护性：清晰的边界和职责分离
- 促进团队协作：统一语言减少沟通成本
- 支持复杂系统：有效管理复杂性
在实际应用中，应该根据项目的具体需求和团队的成熟度来决定是否采用DDD，以及采用DDD的哪些部分。记住，DDD不是银弹，而是一种需要根据具体情况灵活应用的工具。
通过持续学习和实践，团队可以逐步掌握DDD的精髓，构建出更加健壮和可持续的软件系统。