还在被框架绑架？一文看懂“六边形架构”，让你的核心业务稳如泰山！

还在被框架绑架一文看懂“六边形架构”让你的核心业务稳如泰山你有没有遇到过这样的场景辛辛苦苦写了一个项目老板突然说“明年我们要从 MySQL 换成 PostgreSQL或者从 Spring Boot 换成 Django。” 你打开代码一看好家伙业务逻辑和数据库查询、HTTP 请求、外部 API 全混在一起就像一碗杂烩面。改一个数据库驱动结果整个业务层都炸了。 这种痛苦我懂。今天我们来聊聊如何用“六边形架构”Hexagonal Architecture把核心业务“保护”起来让框架和外部服务再也绑架不了你。## 什么是六边形架构六边形架构也叫“端口与适配器架构”Ports and Adapters由 Alistair Cockburn 在 2005 年提出。它的核心思想很简单**把业务逻辑放在最中间像保护心脏一样保护它外部的一切数据库、Web 框架、消息队列、第三方 API都是可以替换的“适配器”。**想象一个六边形 - 中心是“核心业务”和“领域模型”它不依赖任何外部框架或库。 - 六边形的每一条边代表一个“端口”Port比如“用户存储接口”或“订单通知接口”。 - 六边形外侧是“适配器”Adapter比如 MySQL 适配器、REST API 适配器、邮件适配器等它们负责连接外部世界。这样设计的好处是你可以随时更换数据库、更换 Web 框架甚至替换整个 UI而核心业务代码一行都不用改。## 为什么你被框架绑架了很多开发者刚学写代码时喜欢直接打开框架如 Django、Spring Boot在 Controller 里直接写 SQL 查询或者在 Service 里直接调用第三方库。 比如这样python# 糟糕的示例业务逻辑和框架耦合from flask import Flask, requestimport mysql.connectorapp Flask(__name__)app.route(/user/int:user_id)def get_user(user_id): # 业务逻辑和数据库访问混在一起 conn mysql.connector.connect(hostlocalhost, userroot, passwordpassword, databasemydb) cursor conn.cursor() cursor.execute(SELECT id, name, email FROM users WHERE id %s, (user_id,)) row cursor.fetchone() conn.close() if row: return {id: row[0], name: row[1], email: row[2]} else: return {error: User not found}, 404这段代码的问题 1. 如果想把 MySQL 换成 PostgreSQL需要改 Controller 里的所有 SQL 语句。 2. 如果从 Flask 换成 FastAPI需要重写整个路由处理。 3. 单元测试很难写因为你必须启动数据库。 4. 业务逻辑被外部依赖“绑架”了。## 六边形架构的核心原则1.依赖反转核心业务不依赖外部而是定义接口端口让外部适配器去实现。 2.单向依赖外部适配器依赖内部端口内部从不依赖外部。 3.可测试性核心业务可以用 Mock 适配器进行纯逻辑测试无需启动数据库或 HTTP 服务器。## 代码示例 1用 Python 实现一个简单的六边形架构我们用一个“用户注册”功能来演示。核心业务是验证用户邮箱是否已存在然后保存用户。### 1. 定义核心业务领域层python# domain/user.py - 核心业务不依赖任何外部库from dataclasses import dataclassdataclassclass User: id: int name: str email: str# 端口用户存储接口Portclass UserRepository: def find_by_email(self, email: str) - User | None: raise NotImplementedError def save(self, user: User) - None: raise NotImplementedError# 核心业务逻辑Domain Serviceclass UserRegistrationService: def __init__(self, user_repo: UserRepository): self.user_repo user_repo # 依赖接口不是具体实现 def register_user(self, name: str, email: str) - User: # 业务规则邮箱不能重复 existing_user self.user_repo.find_by_email(email) if existing_user: raise ValueError(fEmail {email} is already registered.) new_user User(id0, namename, emailemail) # id 由数据库生成 self.user_repo.save(new_user) return new_user### 2. 实现适配器Adapter—— 以内存存储为例python# adapters/memory_user_repo.py - 内存适配器用于测试或开发from domain.user import User, UserRepositoryclass InMemoryUserRepository(UserRepository): def __init__(self): self.users [] self.next_id 1 def find_by_email(self, email: str) - User | None: for user in self.users: if user.email email: return user return None def save(self, user: User) - None: if user.id 0: user.id self.next_id self.next_id 1 self.users.append(user)### 3. 写一个单元测试测试核心业务python# tests/test_registration.pyfrom domain.user import UserRegistrationServicefrom adapters.memory_user_repo import InMemoryUserRepositorydef test_register_user_success(): repo InMemoryUserRepository() service UserRegistrationService(repo) user service.register_user(Alice, aliceexample.com) assert user.name Alice assert user.email aliceexample.com assert user.id 0 # 确保 ID 生成def test_register_duplicate_email(): repo InMemoryUserRepository() service UserRegistrationService(repo) service.register_user(Bob, bobexample.com) try: service.register_user(Charlie, bobexample.com) assert False, Should have raised ValueError except ValueError as e: assert already registered in str(e)运行测试pytest tests/test_registration.py。 你发现了吗测试不需要数据库不需要 Flask只需要一个简单的内存适配器。## 代码示例 2连接到真实的 MySQL 数据库如果你要上线只需要写一个 MySQL 适配器实现同样的UserRepository接口。python# adapters/mysql_user_repo.py - MySQL 适配器import mysql.connectorfrom domain.user import User, UserRepositoryclass MySQLUserRepository(UserRepository): def __init__(self, host: str, user: str, password: str, database: str): self.conn mysql.connector.connect( hosthost, useruser, passwordpassword, databasedatabase ) def find_by_email(self, email: str) - User | None: cursor self.conn.cursor() cursor.execute(SELECT id, name, email FROM users WHERE email %s, (email,)) row cursor.fetchone() cursor.close() if row: return User(idrow[0], namerow[1], emailrow[2]) return None def save(self, user: User) - None: cursor self.conn.cursor() cursor.execute( INSERT INTO users (name, email) VALUES (%s, %s), (user.name, user.email) ) self.conn.commit() user.id cursor.lastrowid # 获取自增 ID cursor.close()然后在应用的入口比如 Flask 应用中你只需要用MySQLUserRepository替换InMemoryUserRepositorypython# main.py - 应用入口from flask import Flask, request, jsonifyfrom domain.user import UserRegistrationServicefrom adapters.mysql_user_repo import MySQLUserRepositoryapp Flask(__name__)# 初始化数据库适配器repo MySQLUserRepository(hostlocalhost, userroot, passwordpassword, databasemydb)service UserRegistrationService(repo)app.route(/register, methods[POST])def register(): data request.get_json() name data.get(name) email data.get(email) try: user service.register_user(name, email) return jsonify({id: user.id, name: user.name, email: user.email}), 201 except ValueError as e: return jsonify({error: str(e)}), 400if __name__ __main__: app.run(debugTrue)看到没核心业务UserRegistrationService一行代码都没改只是换了一个适配器。 如果将来要从 Flask 换成 FastAPI你只需要换main.py中的路由代码核心业务和适配器都不用动。## 六边形架构的更多好处-框架独立你可以用任何 Web 框架Flask、FastAPI、Django Rest Framework来暴露 API核心业务不受影响。 -数据库灵活从 MySQL 到 PostgreSQL 到 MongoDB只需写新的适配器。 -测试友好核心业务可以用 Mock 适配器测试速度快不依赖外部服务。 -团队协作后端开发者可以专注于领域逻辑前端开发者可以对接 API 适配器互不干扰。## 常见误区1.过度设计如果你的项目只是一个简单的 CRUD 脚本六边形架构可能带来不必要的复杂度。它适合中大型项目或者业务逻辑复杂、需要长期维护的系统。 2.把所有东西都抽象化并不是每个外部依赖都需要一个端口。比如日志库、工具函数可以直接使用只要它们不包含业务逻辑。 3.忽略适配器的测试适配器也需要测试但测试策略不同。比如 MySQL 适配器应该用集成测试而核心业务用单元测试。## 总结六边形架构的核心思想是把业务逻辑放在最中心用接口端口隔离外部依赖用适配器连接外部世界。它让你从“框架绑架”中解脱出来让核心业务稳如泰山。 下次写项目时不妨先画一个六边形中间是你的领域模型每一条边对应一个外部服务。然后先写接口再写适配器。 你会发现代码变得更清晰、更可测试、更易维护。 记住框架只是工具业务才是核心。别再让框架绑架你的代码了