name: 架构模式 description: 实现经过验证的后端架构模式，包括Clean Architecture、Hexagonal Architecture和Domain-Driven Design。在架构复杂后端系统或重构现有应用以提高可维护性时使用。
架构模式
掌握经过验证的后端架构模式，包括Clean Architecture、Hexagonal Architecture和Domain-Driven Design，以构建可维护、可测试和可扩展的系统。
何时使用此技能
- 从头设计新后端系统
- 重构单体应用以提高可维护性
- 为团队建立架构标准
- 从紧密耦合迁移到松散耦合架构
- 实现领域驱动设计原则
- 创建可测试和可模拟的代码库
- 规划微服务分解
核心概念
1. Clean Architecture (Uncle Bob)
层（依赖流向内）：
- 实体：核心业务模型
- 用例：应用业务规则
- 接口适配器：控制器、演示器、网关
- 框架与驱动：UI、数据库、外部服务
关键原则：
- 依赖指向内层
- 内层不知道外层
- 业务逻辑独立于框架
- 无需UI、数据库或外部服务即可测试
2. Hexagonal Architecture (端口和适配器)
组件：
- 领域核心：业务逻辑
- 端口：定义交互的接口
- 适配器：端口的实现（数据库、REST、消息队列）
优点：
- 轻松交换实现（模拟测试）
- 技术无关的核心
- 清晰的关注点分离
3. Domain-Driven Design (DDD)
战略模式：
- 限界上下文：不同领域的独立模型
- 上下文映射：上下文如何关联
- 通用语言：共享术语
战术模式：
- 实体：具有身份的对象
- 值对象：由属性定义的不可变对象
- 聚合：一致性边界
- 仓库：数据访问抽象
- 领域事件：发生的事情
Clean Architecture Pattern
目录结构
app/
├── domain/ # 实体和业务规则
│ ├── entities/
│ ├── value_objects/
│ └── interfaces/ # 抽象接口
├── use_cases/ # 应用业务规则
├── adapters/ # 接口实现
│ ├── repositories/
│ ├── controllers/
│ └── gateways/
└── infrastructure/ # 框架和外部关注点
实现示例
# domain/entities/user.py
from dataclasses import dataclass
from datetime import datetime
@dataclass
class User:
"""核心用户实体 - 无框架依赖。"""
id: str
email: str
name: str
created_at: datetime
is_active: bool = True
def deactivate(self):
"""业务规则：停用用户。"""
self.is_active = False
def can_place_order(self) -> bool:
"""业务规则：活跃用户可以下单。"""
return self.is_active
# domain/interfaces/user_repository.py
from abc import ABC, abstractmethod
from typing import Optional
class IUserRepository(ABC):
"""端口：定义契约，无实现。"""
@abstractmethod
async def find_by_id(self, user_id: str) -> Optional[User]:
pass
@abstractmethod
async def save(self, user: User) -> User:
pass
# use_cases/create_user.py
from dataclasses import dataclass
import uuid
from datetime import datetime
@dataclass
class CreateUserRequest:
email: str
name: str
class CreateUserUseCase:
"""用例：编排业务逻辑。"""
def __init__(self, user_repository: IUserRepository):
self.user_repository = user_repository
async def execute(self, request: CreateUserRequest) -> CreateUserResponse:
# 业务验证
existing = await self.user_repository.find_by_email(request.email)
if existing:
return CreateUserResponse(user=None, success=False, error="邮箱已存在")
# 创建实体
user = User(
id=str(uuid.uuid4()),
email=request.email,
name=request.name,
created_at=datetime.now(),
is_active=True
)
# 持久化
saved_user = await self.user_repository.save(user)
return CreateUserResponse(user=saved_user, success=True)
# adapters/repositories/postgres_user_repository.py
class PostgresUserRepository(IUserRepository):
"""适配器：PostgreSQL 实现。"""
def __init__(self, pool):
self.pool = pool
async def find_by_id(self, user_id: str) -> Optional[User]:
async with self.pool.acquire() as conn:
row = await conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id)
return self._to_entity(row) if row else None
async def save(self, user: User) -> User:
async with self.pool.acquire() as conn:
await conn.execute(
"""INSERT INTO users (id, email, name, created_at, is_active)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (id) DO UPDATE SET email = $2, name = $3, is_active = $5""",
user.id, user.email, user.name, user.created_at, user.is_active
)
return user
Hexagonal Architecture Pattern
# 核心领域（六边形中心）
class OrderService:
"""领域服务 - 无基础设施依赖。"""
def __init__(
self,
order_repository: OrderRepositoryPort,
payment_gateway: PaymentGatewayPort,
notification_service: NotificationPort
):
self.orders = order_repository
self.payments = payment_gateway
self.notifications = notification_service
async def place_order(self, order: Order) -> OrderResult:
# 业务逻辑
if not order.is_valid():
return OrderResult(success=False, error="订单无效")
# 使用端口（接口）
payment = await self.payments.charge(amount=order.total, customer=order.customer_id)
if not payment.success:
return OrderResult(success=False, error="支付失败")
order.mark_as_paid()
saved_order = await self.orders.save(order)
await self.notifications.send(
to=order.customer_email,
subject="订单确认",
body=f"订单 {order.id} 已确认"
)
return OrderResult(success=True, order=saved_order)
# 适配器
class StripePaymentAdapter(PaymentGatewayPort):
"""主要适配器：连接到 Stripe API。"""
async def charge(self, amount: Money, customer: str) -> PaymentResult:
# 实现
class MockPaymentAdapter(PaymentGatewayPort):
"""测试适配器：无外部依赖。"""
async def charge(self, amount: Money, customer: str) -> PaymentResult:
return PaymentResult(success=True, transaction_id="mock-123")
Domain-Driven Design Pattern
# 值对象（不可变）
@dataclass(frozen=True)
class Email:
value: str
def __post_init__(self):
if "@" not in self.value:
raise ValueError("无效邮箱")
@dataclass(frozen=True)
class Money:
amount: int # 分
currency: str
def add(self, other: "Money") -> "Money":
if self.currency != other.currency:
raise ValueError("货币不匹配")
return Money(self.amount + other.amount, self.currency)
# 实体（具有身份）
class Order:
def __init__(self, id: str, customer: Customer):
self.id = id
self.customer = customer
self.items: List[OrderItem] = []
self.status = OrderStatus.PENDING
self._events: List[DomainEvent] = []
def add_item(self, product: Product, quantity: int):
item = OrderItem(product, quantity)
self.items.append(item)
self._events.append(ItemAddedEvent(self.id, item))
def submit(self):
if not self.items:
raise ValueError("不能提交空订单")
self.status = OrderStatus.SUBMITTED
self._events.append(OrderSubmittedEvent(self.id))
# 聚合（一致性边界）
class Customer:
def __init__(self, id: str, email: Email):
self.id = id
self.email = email
self._addresses: List[Address] = []
def add_address(self, address: Address):
if len(self._addresses) >= 5:
raise ValueError("最多允许5个地址")
self._addresses.append(address)
最佳实践
- 依赖规则：依赖始终指向内层
- 接口隔离：小、专注的接口
- 业务逻辑在领域：保持框架远离核心
- 测试独立性：核心无需基础设施即可测试
- 限界上下文：清晰的领域边界
- 通用语言：一致的术语
- 薄控制器：委托给用例
- 丰富领域模型：数据与行为结合
常见陷阱
- 贫血领域：实体只有数据，无行为
- 框架耦合：业务逻辑依赖于框架
- 胖控制器：业务逻辑在控制器中
- 仓库泄漏：暴露ORM对象
- 缺少抽象：核心中的具体依赖
- 过度工程：简单CRUD使用清洁架构