架构师小白

Posted on Mar 22

深度解读六边形架构： Ports 与 Adapters 实战

#架构 #软件设计 #hexagonalarchitecture #编程

深度解读六边形架构：Ports 与 Adapters 实战

你是否遇到过这样的困境：业务逻辑被分散在 Controller、Service、Repository 各个层级中？换个数据库、换个框架，整个系统都要伤筋动骨？

六边形架构（Hexagonal Architecture） 就是来解决这些问题的。

什么是六边形架构？

六边形架构由 Alistair Cockburn 提出，也叫端口与适配器架构（Ports and Adapters）。它的核心思想是：

将业务逻辑放在系统核心，通过端口与外部世界交互，适配器负责对接各种技术。

核心概念

1. 核心领域（Domain/Core）

这是系统的心脏，只包含业务逻辑，不依赖任何外部框架。

# 领域模型示例
class Order:
    def __init__(self, order_id, items, status="pending"):
        self.order_id = order_id
        self.items = items
        self.status = status

    def total(self):
        return sum(item.price * item.quantity for item in self.items)

    def can_cancel(self):
        return self.status in ["pending", "confirmed"]

2. 端口（Ports）

端口是核心领域与外部交互的接口，分为：

输入端口（Driving Ports）：外部驱动系统操作的接口，如 API、服务接口
输出端口（Driven Ports）：核心向外部请求帮助的接口，如 Repository、消息发送

# 输出端口示例 - 定义仓储接口
class OrderRepository(Protocol):
    def save(self, order: Order) -> None: ...
    def find_by_id(self, order_id: str) -> Order | None: ...

3. 适配器（Adapters）

适配器是端口的具体实现，连接外部技术：

主适配器（Driving Adapters）：HTTP 控制器、CLI、消息消费者
从适配器（Driven Adapters）：数据库实现、第三方 API、消息队列

# 从适配器示例 - SQLAlchemy 实现
class SQLAlchemyOrderRepository:
    def __init__(self, session):
        self.session = session

    def save(self, order: Order):
        self.session.add(order)
        self.session.commit()

为什么需要六边形架构？

传统分层	六边形架构
业务逻辑依赖框架	业务逻辑完全独立
换数据库代价大	随时替换适配器
难以测试	领域模型可单独测试
关注技术实现	关注业务价值

实战案例：订单系统

目录结构

src/
├── domain/           # 核心领域
│   ├── models/       # 领域模型
│   └── services/     # 领域服务
├── ports/            # 端口定义
│   ├── input/        # 输入端口
│   └── output/       # 输出端口
├── adapters/         # 适配器实现
│   ├── api/          # HTTP 适配器
│   └── persistence/  # 数据库适配器
└── application/      # 应用服务（用例）

代码实现

# domain/models/order.py
class Order:
    def __init__(self, items):
        self.items = items
        self.status = "pending"

    def total(self):
        return sum(i.price * i.quantity for i in self.items)

# ports/output/order_repository.py
from typing import Protocol
class OrderRepository(Protocol):
    def save(self, order: Order): ...
    def find(self, id: str): Order: ...

# application/create_order.py
class CreateOrderUseCase:
    def __init__(self, repo: OrderRepository):
        self.repo = repo

    def execute(self, order_data):
        order = Order(order_data["items"])
        self.repo.save(order)
        return order

# adapters/persistence/sqlalchemy_repo.py
class SQLAlchemyOrderRepository:
    def __init__(self, session):
        self.session = session

    def save(self, order: Order):
        self.session.add(...)

如何开始？

从核心开始：先画出你的业务模型，不要想数据库、API
定义端口：明确核心需要什么（存储、消息通知...）
实现适配器：先用一个简单的适配器，后续可以替换
渐进式改造：不需要一次性重构，先从新模块开始

总结

六边形架构帮助我们：

✅ 关注业务价值：让代码服务于业务，而非技术
✅ 保持敏捷：轻松应对技术变更
✅ 提升可测试性：核心逻辑可以单独测试
✅ 明确边界：端口和适配器各司其职

架构不是关于技术选择，而是关于如何组织代码让业务价值最大化。

DEV Community