架构模式

是什么

用于构建可维护、可测试系统的后端架构模式：整洁架构、六边形架构和领域驱动设计。

何时使用

• - 从零开始设计新的后端系统
• 重构单体应用以提高可维护性
• 为团队建立架构标准
• 创建可测试、可模拟的代码库
• 规划微服务拆分

关键词

整洁架构、六边形架构、端口与适配器、DDD、领域驱动设计、分层、实体、用例、仓储、聚合、限界上下文

---

决策框架：选择哪种模式？

场景	推荐模式
简单的CRUD应用	无需使用（过度设计）
中等复杂度，团队标准化需求

整洁架构 | | 多个频繁变化的外部集成 | 六边形架构（端口与适配器） | | 包含大量规则的复杂业务领域 | 领域驱动设计 | | 多团队协作的大型系统 | DDD + 限界上下文 |

快速参考

整洁架构分层

┌──────────────────────────────────────┐
│ 框架与驱动（UI、数据库） │ ← 外层：可变
├──────────────────────────────────────┤
│ 接口适配器 │ ← 控制器、网关
├──────────────────────────────────────┤
│ 用例层 │ ← 应用逻辑
├──────────────────────────────────────┤
│ 实体层 │ ← 核心业务规则
└──────────────────────────────────────┘

依赖规则：依赖方向仅指向内部。内层绝不能导入外层。

六边形架构

┌─────────────┐
┌────│ 适配器 │────┐ （REST API）
│ └─────────────┘ │
▼ ▼
┌──────┐ ┌──────────┐
│ 端口 │◄────────────►│ 领域层 │
└──────┘ └──────────┘
▲ ▲
│ ┌─────────────┐ │
└────│ 适配器 │────┘ （数据库）
└─────────────┘

端口：定义领域所需功能的接口
适配器：具体实现（可替换以进行测试）

---

目录结构

app/
├── domain/ # 实体与业务规则（最内层）
│ ├── entities/
│ │ └── user.py
│ ├── value_objects/
│ │ └── email.py
│ └── interfaces/ # 端口
│ └── user_repository.py
├── use_cases/ # 应用业务规则
│ └── create_user.py
├── adapters/ # 接口实现
│ ├── repositories/
│ │ └── postgresuserrepository.py
│ └── controllers/
│ └── user_controller.py
└── infrastructure/ # 框架与外部关注点
├── database.py
└── config.py

---

模式一：整洁架构

实体（领域层）

python
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 canplaceorder(self) -> bool:
return self.is_active

端口（接口）

python
from abc import ABC, abstractmethod
from typing import Optional

class IUserRepository(ABC):
端口：定义契约，不包含实现。

@abstractmethod
async def findbyid(self, user_id: str) -> Optional[User]:
pass

@abstractmethod
async def save(self, user: User) -> User:
pass

用例（应用层）

python
@dataclass
class CreateUserRequest:
email: str
name: str

@dataclass
class CreateUserResponse:
user: Optional[User]
success: bool
error: Optional[str] = None

class CreateUserUseCase:
用例：编排业务逻辑。

def init(self, user_repository: IUserRepository):
self.userrepository = userrepository # 依赖注入

async def execute(self, request: CreateUserRequest) -> CreateUserResponse:
# 业务验证
existing = await self.userrepository.findby_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()
)

saved = await self.user_repository.save(user)
return CreateUserResponse(user=saved, success=True)

适配器（实现）

python
class PostgresUserRepository(IUserRepository):
适配器：端口的PostgreSQL实现。

def init(self, pool: asyncpg.Pool):
self.pool = pool

async def findbyid(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.toentity(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, createdat, isactive)
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.createdat, user.isactive
)
return user

---

模式二：六边形架构（端口与适配器）

适用于有多个可能变化的外部集成场景。

python

领域服务（核心）

class OrderService:
def init(
self,
order_repo: OrderRepositoryPort, # 端口
payment: PaymentGatewayPort, # 端口
notifications: NotificationPort # 端口
):
self.orders = order_repo
self.payments = payment
self.notifications = notifications

async def place_order(self, order: Order) -> OrderResult:
# 纯业务逻辑 - 无基础设施细节
if not order.is_valid():
return OrderResult(success=False, error=无效订单)

payment = await self.payments.charge(order.total, order.customer_id)
if not payment.success:
return OrderResult(success=False, error=支付失败)

order.markaspaid()
saved = await self.orders.save(order)
await self.notifications.send(order.customer_email, 订单已确认)

return OrderResult(success=True, order=saved)

适配器（测试或更换提供商时可替换）

class StripePaymentAdapter(PaymentGatewayPort): async def charge(self, amount: Money, customer: str) -> PaymentResult: # 真实的Stripe实现 ...

class MockPaymentAdapter(PaymentGatewayPort):
async def charge(self, amount: Money, customer: str) -> PaymentResult:
return PaymentResult(success=True, transaction_id=mock-123)

---

模式三：领域驱动设计

适用于包含大量规则的复杂业务领域。

值对象（不可变）

python
@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)

聚合（一致性边界）

python
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):
聚合中的业务逻辑。
if quantity > product.max_quantity:
raise ValueError(f最多允许{product.max_quantity}个)

item = OrderItem(product,