从写代码到做架构的跃迁
很多工作3-5年的程序员会遇到一个典型的困境：代码写得越来越熟练，bug越来越少，但负责的系统却越来越难以维护。新需求来了要改很多地方，改一处代码可能影响多个模块，测试回归范围越来越大，上线越来越谨慎——系统变得“僵硬”了。
这个问题的本质是：缺乏架构视角。
写代码解决的是“如何实现”的问题，而架构解决的是“如何组织”的问题。前者让你成为一个合格的开发者，后者让你成为一个能够驾驭复杂系统的工程师。
本文将深入探讨架构落地的全流程和组件封装的实战技巧，帮助你完成从“码农”到“工程师”的关键跃迁。
一、架构的本质与价值
1.1 什么是软件架构？
软件架构是系统的高层结构，包含以下核心要素：
架构 = 组件 + 连接件 + 约束 + 决策
- 组件：系统的功能模块（订单模块、用户模块）
- 连接件：组件间的通信机制（RPC、消息队列、HTTP）
- 约束：必须遵守的规则（分层原则、依赖方向）
- 决策：设计时做出的重要选择（技术选型、数据分片策略）
1.2 架构解决的核心问题
1.复杂性管理
没有架构的100万行代码是不可维护的混沌体，有架构的100万行代码是可以理解、修改、扩展的有机体。
2.变化应对
需求永远在变。好的架构让80%的需求变更只影响20%的代码。
3.质量属性保障
性能：响应时间、吞吐量
可用性：故障恢复、容错
可扩展性：水平扩展能力
安全性：数据保护、访问控制
可维护性：代码理解、修改成本
4.团队协作边界
清晰的架构定义了模块边界，团队可以并行开发互不干扰。
1.3 架构设计的核心原则
原则一：分离关注点（Separation of Concerns）
每个模块只关注一件事，不要交叉。
❌ 错误：一个函数既查询数据库、又计算业务逻辑、还发送邮件
✅ 正确：Repository负责数据访问、Service负责业务、Notification负责通知
原则二：高内聚低耦合（High Cohesion Low Coupling）
高内聚：相关功能放在一起
低耦合：模块之间依赖尽可能少
原则三：依赖倒置（Dependency Inversion）
高层模块不依赖低层模块，都依赖抽象。
传统分层：Controller → Service → DAO → MySQL
依赖倒置：Controller → Service接口 ← Service实现 → DAO接口 ← DAO实现 → MySQL
原则四：开闭原则（Open-Closed Principle）
对扩展开放，对修改关闭。通过抽象和插件化实现。
原则五：康威定律（Conway's Law）
系统架构会反映组织沟通结构。团队怎么组织，系统就会怎么分层。
二、架构设计的分层方法论
2.1 经典四层架构
┌─────────────────────────────────────────┐
│ 表现层 (Presentation) │
│ Controller/Handler/API Gateway │
│ 职责：接收请求、参数校验、响应格式化 │
└─────────────────────────────────────────┘
↓
┌─────────────────────────────────────────┐
│ 应用层 (Application) │
│ Service/UseCase/Application Service │
│ 职责：业务流程编排、事务管理、权限控制 │
└─────────────────────────────────────────┘
↓
┌─────────────────────────────────────────┐
│ 领域层 (Domain) │
│ Entity/Value Object/Aggregate/Domain │
│ Service │
│ 职责：核心业务逻辑、领域规则 │
└─────────────────────────────────────────┘
↓
┌─────────────────────────────────────────┐
│ 基础设施层 (Infrastructure) │
│ Repository/DAO/MQ/Cache/External API │
│ 职责：数据持久化、外部通信、技术细节 │
└─────────────────────────────────────────┘
各层代码示例：
# 表现层：接收HTTP请求
# controllers/order_controller.py
from fastapi import APIRouter, Depends, HTTPException
from typing import List
router = APIRouter(prefix="/api/v1/orders", tags=["orders"])
@router.post("/", response_model=OrderResponse)
async def create_order(
request: CreateOrderRequest,
order_service: OrderService = Depends(get_order_service)
):
"""
创建订单
职责：参数校验、调用应用服务、返回响应
"""
# 1. 参数校验（简单校验）
if not request.items:
raise HTTPException(status_code=400, detail="订单不能为空")
# 2. 调用应用服务
try:
order = await order_service.create_order(
user_id=request.user_id,
items=request.items,
shipping_address=request.address
)
except InsufficientStockError as e:
raise HTTPException(status_code=409, detail=str(e))
except UserNotFoundError as e:
raise HTTPException(status_code=404, detail=str(e))
# 3. 返回响应
return OrderResponse.from_entity(order)
# 应用层：业务流程编排
# application/order_service.py
from typing import List
from datetime import datetime
class OrderService:
"""订单应用服务 - 协调领域对象和基础设施"""
def __init__(
self,
order_repository: OrderRepository,
product_service: ProductService,
payment_gateway: PaymentGateway,
inventory_service: InventoryService,
event_publisher: EventPublisher
):
self.order_repository = order_repository
self.product_service = product_service
self.payment_gateway = payment_gateway
self.inventory_service = inventory_service
self.event_publisher = event_publisher
async def create_order(
self,
user_id: str,
items: List[OrderItemRequest],
shipping_address: Address
) -> Order:
"""
创建订单流程
职责：协调各个领域对象和服务，管理事务边界
"""
# 1. 获取商品信息
product_ids = [item.product_id for item in items]
products = await self.product_service.get_products_by_ids(product_ids)
# 2. 构建订单实体（领域逻辑在实体内部）
order_items = []
for item in items:
product = products[item.product_id]
order_item = OrderItem.create(
product=product,
quantity=item.quantity,
price_at_order=product.price # 快照价格
)
order_items.append(order_item)
# 3. 创建订单聚合根
order = Order.create(
order_id=self._generate_order_id(),
user_id=user_id,
items=order_items,
shipping_address=shipping_address
)
# 4. 预留库存（调用领域服务或基础设施）
reservation_result = await self.inventory_service.reserve(
order.order_id,
[(item.product_id, item.quantity) for item in order_items]
)
if not reservation_result.success:
raise InsufficientStockError(reservation_result.failed_items)
# 5. 保存订单
await self.order_repository.save(order)
# 6. 发布领域事件
await self.event_publisher.publish(OrderCreatedEvent.from_entity(order))
return order
# 领域层：核心业务逻辑
# domain/order.py
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from typing import List, Optional
from decimal import Decimal
class OrderStatus(Enum):
PENDING = "pending"
PAID = "paid"
SHIPPED = "shipped"
DELIVERED = "delivered"
CANCELLED = "cancelled"
@dataclass
class OrderItem:
"""订单项 - 值对象"""
product_id: str
product_name: str
quantity: int
unit_price: Decimal
@classmethod
def create(cls, product: Product, quantity: int, price_at_order: Decimal):
"""创建订单项时的业务规则"""
if quantity <= 0:
raise InvalidQuantityError("商品数量必须大于0")
if quantity > 999:
raise InvalidQuantityError("单个商品数量不能超过999")
return cls(
product_id=product.id,
product_name=product.name,
quantity=quantity,
unit_price=price_at_order
)
def subtotal(self) -> Decimal:
return self.unit_price * self.quantity
class Order:
"""订单聚合根"""
def __init__(self, order_id: str, user_id: str):
self.order_id = order_id
self.user_id = user_id
self._items: List[OrderItem] = []
self._status = OrderStatus.PENDING
self._created_at = datetime.now()
self._paid_at: Optional[datetime] = None
self._shipping_address: Optional[Address] = None
self._total_amount = Decimal('0')
self._domain_events: List[DomainEvent] = []
@classmethod
def create(cls, order_id: str, user_id: str, items: List[OrderItem],
shipping_address: Address) -> 'Order':
"""创建订单的工厂方法，包含创建时的业务规则"""
if not items:
raise EmptyOrderError("订单不能为空")
order = cls(order_id, user_id)
order._items = items
order._shipping_address = shipping_address
order._total_amount = sum(item.subtotal() for item in items)
# 业务规则：订单金额最小限制
if order._total_amount < Decimal('0.01'):
raise InvalidOrderAmountError("订单金额无效")
# 发布领域事件
order._add_domain_event(OrderCreatedEvent(order_id, user_id, order._total_amount))
return order
def pay(self, payment_info: PaymentInfo):
"""支付业务逻辑"""
# 业务规则：只有待支付的订单可以支付
if self._status != OrderStatus.PENDING:
raise InvalidOrderStateError(f"订单状态为{self._status}，无法支付")
# 业务规则：支付金额必须匹配
if payment_info.amount != self._total_amount:
raise PaymentAmountMismatchError(
f"支付金额{payment_info.amount}与订单金额{self._total_amount}不符"
)
# 状态变更
self._status = OrderStatus.PAID
self._paid_at = datetime.now()
# 发布事件
self._add_domain_event(OrderPaidEvent(self.order_id, payment_info.transaction_id))
def cancel(self, reason: str):
"""取消订单业务逻辑"""
# 业务规则：已支付的订单不能直接取消（需要走退款流程）
if self._status == OrderStatus.PAID:
raise OrderPaidError("订单已支付，请通过退款流程取消")
# 业务规则：已发货的订单不能取消
if self._status in [OrderStatus.SHIPPED, OrderStatus.DELIVERED]:
raise OrderShippedError("订单已发货，无法取消")
self._status = OrderStatus.CANCELLED
self._add_domain_event(OrderCancelledEvent(self.order_id, reason))
def ship(self, tracking_number: str):
"""发货业务逻辑"""
if self._status != OrderStatus.PAID:
raise InvalidOrderStateError("只有已支付的订单才能发货")
self._status = OrderStatus.SHIPPED
self._add_domain_event(OrderShippedEvent(self.order_id, tracking_number))
@property
def status(self) -> OrderStatus:
return self._status
@property
def items(self) -> List[OrderItem]:
"""返回副本，防止外部修改"""
return list(self._items)
@property
def total_amount(self) -> Decimal:
return self._total_amount
def _add_domain_event(self, event: DomainEvent):
self._domain_events.append(event)
@property
def domain_events(self) -> List[DomainEvent]:
return list(self._domain_events)
def clear_events(self):
self._domain_events.clear()
# 基础设施层：数据持久化
# infrastructure/repositories/order_repository.py
class OrderRepository:
"""订单仓储 - 负责订单聚合的持久化"""
def __init__(self, db_session, redis_client):
self.db_session = db_session
self.cache = redis_client
async def save(self, order: Order):
"""保存订单聚合（完整保存）"""
# 保存到MySQL
order_po = OrderPO(
order_id=order.order_id,
user_id=order.user_id,
status=order.status.value,
total_amount=order.total_amount,
created_at=order.created_at,
paid_at=order._paid_at
)
await self.db_session.merge(order_po)
# 保存订单商品
for item in order.items:
item_po = OrderItemPO(
order_id=order.order_id,
product_id=item.product_id,
product_name=item.product_name,
quantity=item.quantity,
unit_price=item.unit_price
)
await self.db_session.merge(item_po)
# 更新缓存（异步）
await self.cache.setex(
f"order:{order.order_id}",
3600,
json.dumps(self._to_cache_dto(order))
)
await self.db_session.commit()
async def find_by_id(self, order_id: str) -> Optional[Order]:
"""根据ID查询订单"""
# 先查缓存
cached = await self.cache.get(f"order:{order_id}")
if cached:
return self._from_cache_dto(json.loads(cached))
# 查数据库
order_po = await self.db_session.query(OrderPO).filter_by(order_id=order_id).first()
if not order_po:
return None
item_pos = await self.db_session.query(OrderItemPO).filter_by(order_id=order_id).all()
# 重建订单聚合
order = self._reconstruct(order_po, item_pos)
# 写回缓存
await self.cache.setex(f"order:{order_id}", 3600, json.dumps(self._to_cache_dto(order)))
return order
2.2 六边形架构（端口-适配器模式）
六边形架构强调业务逻辑与技术实现的隔离，所有外部依赖通过端口（接口）连接。
┌─────────────────┐
│ HTTP适配器 │
└────────┬────────┘
│
┌──────────────┐ ┌────────┴────────┐ ┌──────────────┐
│ 数据库适配器 │◄───│ 应用核心 │───►│ 消息队列适配器 │
└──────────────┘ └────────┬────────┘ └──────────────┘
│
┌────────┴────────┐
│ 外部API适配器 │
└─────────────────┘
实现示例：
# 端口（接口）- 定义在核心层
# core/ports/repositories.py
from abc import ABC, abstractmethod
from typing import Optional
from core.entities import Order
class OrderRepositoryPort(ABC):
"""订单仓储端口 - 核心层定义的接口"""
@abstractmethod
async def save(self, order: Order) -> None:
pass
@abstractmethod
async def find_by_id(self, order_id: str) -> Optional[Order]:
pass
class PaymentServicePort(ABC):
"""支付服务端口"""
@abstractmethod
async def charge(self, amount: Decimal, payment_method: str) -> PaymentResult:
pass
# 适配器（实现）- 在基础设施层实现端口
# infrastructure/adapters/mysql_order_repository.py
class MySQLOrderRepository(OrderRepositoryPort):
"""MySQL实现的订单仓储"""
def __init__(self, session_factory):
self.session_factory = session_factory
async def save(self, order: Order) -> None:
async with self.session_factory() as session:
# MySQL特定的保存逻辑
po = self._to_persistent_object(order)
session.add(po)
await session.commit()
async def find_by_id(self, order_id: str) -> Optional[Order]:
async with self.session_factory() as session:
po = await session.get(OrderPO, order_id)
return self._to_entity(po) if po else None
# 另一个适配器实现（支持切换）
# infrastructure/adapters/redis_order_repository.py
class RedisOrderRepository(OrderRepositoryPort):
"""Redis实现的订单仓储（用于缓存场景）"""
def __init__(self, redis_client):
self.redis = redis_client
async def save(self, order: Order) -> None:
await self.redis.set(
f"order:{order.order_id}",
self._serialize(order),
ex=3600
)
async def find_by_id(self, order_id: str) -> Optional[Order]:
data = await self.redis.get(f"order:{order_id}")
return self._deserialize(data) if data else None
# 应用层 - 依赖端口而非具体实现
# core/application/order_processor.py
class OrderProcessor:
def __init__(
self,
order_repo: OrderRepositoryPort, # 依赖端口，不是具体类
payment_service: PaymentServicePort
):
self.order_repo = order_repo
self.payment_service = payment_service
async def process_order(self, order_id: str):
order = await self.order_repo.find_by_id(order_id)
if not order:
raise OrderNotFoundError()
# 业务逻辑...
result = await self.payment_service.charge(order.total_amount, "credit_card")
if result.success:
order.mark_as_paid()
await self.order_repo.save(order)
# 依赖注入 - 在启动时组装
# main.py
from infrastructure.adapters.mysql_order_repository import MySQLOrderRepository
from infrastructure.adapters.stripe_payment_service import StripePaymentService
def bootstrap():
# 组装依赖
order_repo = MySQLOrderRepository(create_session_factory())
payment_service = StripePaymentService(api_key=os.getenv("STRIPE_KEY"))
# 注入到应用服务
processor = OrderProcessor(order_repo, payment_service)
return processor
2.3 CQRS（命令查询职责分离）
将读操作和写操作分离到不同的模型和数据库中。
┌─────────────┐
│ 客户端 │
└──────┬──────┘
│
┌───┴───┐
│ 路由 │
└───┬───┘
│
┌───┴────────────────────────┐
│ │
▼ ▼
┌─────────────┐ ┌─────────────┐
│ 命令处理器 │ │ 查询处理器 │
│ (写模型) │ │ (读模型) │
└──────┬──────┘ └──────┬──────┘
│ │
▼ ▼
┌─────────────┐ ┌─────────────┐
│ 写数据库 │ │ 读数据库 │
│ (规范化) │◄─────────│ (反规范化) │
└─────────────┘ 事件同步 └─────────────┘
CQRS实现示例：
# ========== 命令侧（写模型）==========
# commands.py
from dataclasses import dataclass
from decimal import Decimal
from typing import List
@dataclass
class CreateOrderCommand:
user_id: str
items: List[OrderItemDto]
shipping_address: AddressDto
@dataclass
class CancelOrderCommand:
order_id: str
reason: str
# command_handler.py
class OrderCommandHandler:
def __init__(self, order_repo: OrderRepository, event_bus: EventBus):
self.order_repo = order_repo
self.event_bus = event_bus
async def handle_create_order(self, command: CreateOrderCommand) -> str:
"""处理创建订单命令"""
# 创建订单聚合
order = Order.create(
order_id=self._generate_id(),
user_id=command.user_id,
items=self._to_domain_items(command.items),
shipping_address=self._to_domain_address(command.shipping_address)
)
# 保存到写数据库
await self.order_repo.save(order)
# 发布事件（用于同步到读数据库）
await self.event_bus.publish(OrderCreatedEvent.from_entity(order))
return order.order_id
async def handle_cancel_order(self, command: CancelOrderCommand):
order = await self.order_repo.find_by_id(command.order_id)
if not order:
raise OrderNotFoundError()
order.cancel(command.reason)
await self.order_repo.save(order)
# 发布取消事件
await self.event_bus.publish(OrderCancelledEvent(order.order_id, command.reason))
# ========== 查询侧（读模型）==========
# read_model.py - 针对查询优化的反规范化模型
from dataclasses import dataclass
from datetime import datetime
@dataclass
class OrderReadModel:
"""订单读模型 - 为列表查询优化"""
order_id: str
user_id: str
user_name: str # 冗余用户姓名，避免JOIN
total_amount: Decimal
status: str
created_at: datetime
item_count: int
shipping_status: str
@dataclass
class OrderDetailReadModel:
"""订单详情读模型"""
order_id: str
user_id: str
user_name: str
user_phone: str
total_amount: Decimal
status: str
created_at: datetime
paid_at: Optional[datetime]
shipped_at: Optional[datetime]
items: List[OrderItemReadModel]
shipping_address: AddressDto
payment_method: Optional[str]
# query_handlers.py
class OrderQueryHandler:
def __init__(self, order_read_db: OrderReadRepository):
self.read_db = order_read_db
async def get_user_orders(
self,
user_id: str,
page: int,
page_size: int,
status: Optional[str] = None
) -> PageResult[OrderReadModel]:
"""查询用户订单列表 - 直接从读库查询"""
query = self.read_db.orders.filter_by(user_id=user_id)
if status:
query = query.filter_by(status=status)
total = await query.count()
items = await query.offset((page-1)*page_size).limit(page_size).all()
return PageResult(total=total, page=page, page_size=page_size, items=items)
async def get_order_detail(self, order_id: str) -> Optional[OrderDetailReadModel]:
"""查询订单详情 - 从读库的预连接表查询"""
# 读库已经预先JOIN好或冗余了所有需要的数据
return await self.read_db.order_details.find_by_id(order_id)
# 事件处理器 - 同步写库到读库
# event_handlers.py
class OrderReadModelUpdater:
"""事件处理器：更新读模型"""
def __init__(self, read_db: OrderReadRepository):
self.read_db = read_db
async def on_order_created(self, event: OrderCreatedEvent):
"""订单创建时，更新读数据库"""
# 从写库或事件中获取必要信息
order_detail = OrderDetailReadModel(
order_id=event.order_id,
user_id=event.user_id,
user_name=await self._get_user_name(event.user_id), # 可能从用户服务获取
user_phone=await self._get_user_phone(event.user_id),
total_amount=event.total_amount,
status="pending",
created_at=event.created_at,
paid_at=None,
shipped_at=None,
items=event.items, # 事件中已包含必要信息
shipping_address=event.shipping_address,
payment_method=None
)
# 保存到读库
await self.read_db.order_details.save(order_detail)
# 同时更新列表视图
list_view = OrderReadModel(
order_id=event.order_id,
user_id=event.user_id,
user_name=await self._get_user_name(event.user_id),
total_amount=event.total_amount,
status="pending",
created_at=event.created_at,
item_count=len(event.items),
shipping_status="pending"
)
await self.read_db.orders.save(list_view)
async def on_order_paid(self, event: OrderPaidEvent):
"""订单支付时，更新读模型"""
await self.read_db.order_details.update(
event.order_id,
status="paid",
paid_at=event.paid_at
)
await self.read_db.orders.update(
event.order_id,
status="paid"
)
