引言

最近在读微服务架构先驱，Java 开发者社区的意见领袖 Chris Richardson 的《微服务架构设计模式》。作者在第二章讲到微服务的两种拆分方式 —— 根据业务进行拆分和根据子域进行拆分。

在阅读第二章的过程中，我很自然地有一些疑问：

这两种拆分方式有什么区别？
领域驱动设计（Domain-Driven Design 下文用简称 DDD）提出的一些晦涩概念比如限界上下文、子域等等是什么意思？
大家都知道要按业务划分服务：毕竟业务是稳定的，而实现是不断变化的；将实现的复杂度隔离在微服务的 API 中，有利于隔离变化带来的稳定性风险。那么按照子域进行拆分有什么独特优势吗？

目前，这两种拆分方式在我脑中只是抽象的概念，我希望通过具体的示例来理解概念。本文以在线书店业务为例，给出了业务拆分和子域拆分 base 版本的代码实现（这里结合子域拆分的代码实现了介绍一些 DDD的概念），来对比两种划分方式的核心区别；然后设想三个需求迭代的场景，对比了两种划分方式下我们需要做哪些修改以及对应的影响有哪些；最后总结 DDD 的适用场景和核心优势。

在线书店业务

下面我将提供在线书店业务的两份基础代码实现，通过具体代码对比业务划分与子领域划分的差异，并结合代码解释DDD的核心概念。

按业务功能划分的实现

这种方式按照功能模块划分服务，每个服务负责相关的所有操作，业务流程在服务中直接实现。

*# 订单服务 - 负责所有与订单相关的功能*
class OrderService:
    def **__init__**(*self*, *inventory_service*, *payment_service*, *logistics_service*):
        *self*.inventory_service = *inventory_service*
        *self*.payment_service = *payment_service*
        *self*.logistics_service = *logistics_service*
        *self*.orders = {}  *# 存储订单信息的简易数据库*

    def **create_order**(*self*, *user_id*, *books*):
        """创建订单的完整流程"""
        *# 1. 检查库存*
        *for* book *in* *books*:
            *if* not *self*.inventory_service.check_stock(book["id"], book["quantity"]):
                *raise* Exception(f"书籍《{book['id']}》库存不足")

        *# 2. 创建订单记录*
        order_id = f"order_{len(*self*.orders) + 1}"
        *# 从库存服务获取价格信息*
        total_amount = 0
        *for* book *in* *books*:
            book_info = *self*.inventory_service.get_book_info(book["id"])
            *if* not book_info:
                *raise* Exception(f"书籍《{book['id']}》不存在")
            total_amount += book_info["price"] * book["quantity"]

        *self*.orders[order_id] = {
            "order_id": order_id,
            "user_id": *user_id*,
            "books": *books*,
            "total_amount": total_amount,
            "status": "PENDING"
        }
        print(f"订单创建成功: {order_id}, 总金额: {total_amount}")

        *# 3. 扣减库存*
        *for* book *in* *books*:
            *self*.inventory_service.reduce_stock(book["id"], book["quantity"])

        *# 4. 处理支付*
        payment_result = *self*.payment_service.process_payment(*user_id*, order_id, total_amount)
        *if* not payment_result["success"]:
            *# 支付失败，回滚库存*
            *for* book *in* *books*:
                *self*.inventory_service.increase_stock(book["id"], book["quantity"])
            *self*.orders[order_id]["status"] = "PAYMENT_FAILED"
            *raise* Exception(f"支付失败: {payment_result['message']}")

        *# 5. 标记订单为已支付*
        *self*.orders[order_id]["status"] = "PAID"

        *# 6. 生成物流单*
        *self*.logistics_service.create_shipment(order_id, *user_id*, *books*)

        *return* *self*.orders[order_id]

    def **get_order**(*self*, *order_id*):
        """查询订单信息"""
        *return* *self*.orders.get(*order_id*)

*# 库存服务 - 负责所有与库存相关的功能*
class InventoryService:
    def **__init__**(*self*):
        *# 初始化一些书籍库存*
        *self*.inventory = {
            "Python编程": {"quantity": 100, "price": 89},
            "DDD实战": {"quantity": 50, "price": 79},
            "设计模式": {"quantity": 30, "price": 69}
        }

    def **check_stock**(*self*, *book_id*, *quantity*):
        """检查库存是否充足"""
        book = *self*.inventory.get(*book_id*)
        *return* book and book["quantity"] >= *quantity*

    def **reduce_stock**(*self*, *book_id*, *quantity*):
        """减少库存"""
        *if* *self*.check_stock(*book_id*, *quantity*):
            *self*.inventory[*book_id*]["quantity"] -= *quantity*
            print(f"库存扣减成功: 《{*book_id*}》减少 {*quantity*} 本")
            *return* True
        *return* False

    def **increase_stock**(*self*, *book_id*, *quantity*):
        """增加库存"""
        *if* *book_id* in *self*.inventory:
            *self*.inventory[*book_id*]["quantity"] += *quantity*
            print(f"库存增加成功: 《{*book_id*}》增加 {*quantity*} 本")
            *return* True
        *return* False

    def **get_book_info**(*self*, *book_id*):
        """获取书籍信息"""
        *return* *self*.inventory.get(*book_id*)

*# 支付服务 - 负责所有与支付相关的功能*
class PaymentService:
    def **process_payment**(*self*, *user_id*, *order_id*, *amount*):
        """处理支付"""
        print(f"处理用户 {*user_id*} 的订单 {*order_id*} 支付，金额: {*amount*} 元")
        *# 模拟支付处理*
        *return* {"success": True, "message": "支付成功", "transaction_id": f"trans_{*order_id*}"}

*# 物流服务 - 负责所有与物流相关的功能*
class LogisticsService:
    def **__init__**(*self*):
        *self*.shipments = {}  *# 存储物流信息*

    def **create_shipment**(*self*, *order_id*, *user_id*, *books*):
        """创建物流单"""
        shipment_id = f"ship_{*order_id*}"
        *self*.shipments[shipment_id] = {
            "shipment_id": shipment_id,
            "order_id": *order_id*,
            "user_id": *user_id*,
            "books": *books*,
            "status": "PREPARING"
        }
        print(f"物流单创建成功: {shipment_id}，准备发货")
        *return* shipment_id

*# 业务流程演示*
*if* __name__ == "__main__":
    *# 初始化服务*
    inventory_service = InventoryService()
    payment_service = PaymentService()
    logistics_service = LogisticsService()
    order_service = OrderService(inventory_service, payment_service, logistics_service)

    *# 购买书籍*
    user_id = "user_001"
    books_to_buy = [
        {"id": "Python编程", "quantity": 2},
        {"id": "DDD实战", "quantity": 1}
    ]

    *try*:
        *# 创建订单（包含了完整的业务流程）*
        order = order_service.create_order(user_id, books_to_buy)
        print(f"最终订单状态: {order['status']}")

        *# 查询订单*
        retrieved_order = order_service.get_order(order["order_id"])
        print(f"查询订单: {retrieved_order}")
    *except* Exception *as* e:
        print(f"下单失败: {str(e)}")

按业务功能划分的代码特点：

服务划分：按照功能职责分为OrderService、InventoryService、PaymentService和LogisticsService
流程实现：核心业务流程（创建订单→检查库存→扣减库存→处理支付→创建物流单）直接在OrderService.create_order方法中实现
依赖关系：服务之间直接依赖，OrderService需要知道并调用其他服务的方法
数据存储：每个服务管理自己的数据（如orders、inventory等）
业务规则：业务规则分散在各个服务的方法中，没有集中封装

按DDD子领域划分的实现

*# ------------------------------------------------------*
*# 1. 共享内核 - 通用值对象和异常*
*# ------------------------------------------------------*
class Money:
    """金额值对象（无唯一标识，不可变）"""
    def **__init__**(*self*, *amount*, *currency*="CNY"):
        *self*.amount = round(*amount*, 2)
        *self*.currency = *currency*

    def **__add__**(*self*, *other*):
        *if* *self*.currency != *other*.currency:
            *raise* ValueError("货币类型不同，无法相加")
        *return* Money(*self*.amount + *other*.amount, *self*.currency)

    def **__repr__**(*self*):
        *return* f"{*self*.amount} {*self*.currency}"

class BookId:
    """书籍ID值对象"""
    def **__init__**(*self*, *value*):
        *self*.value = *value*

    def **__eq__**(*self*, *other*):
        *return* isinstance(*other*, BookId) and *self*.value == *other*.value

    def **__hash__**(*self*):
        """添加哈希方法，使BookId实例可作为字典键"""
        *return* hash(*self*.value)

    def **__repr__**(*self*):
        *return* *self*.value

*# ------------------------------------------------------*
*# 2. 订单子域 - 包含订单相关的领域模型*
*# ------------------------------------------------------*
class OrderItem:
    """订单项值对象"""
    def **__init__**(*self*, *book_id*, *quantity*, *unit_price*):
        *self*.book_id = *book_id*  *# BookId对象*
        *self*.quantity = *quantity*
        *self*.unit_price = *unit_price*  *# Money对象*

        *if* *quantity* <= 0:
            *raise* ValueError("订单项数量必须大于0")

    @property
    def total_price(*self*):
        """计算订单项总价"""
        *return* Money(*self*.unit_price.amount * *self*.quantity)

class Order:
    """订单聚合根（包含订单的核心业务规则）"""
    def **__init__**(*self*, *order_id*, *user_id*):
        *self*.order_id = *order_id*
        *self*.user_id = *user_id*
        *self*.items = []
        *self*.status = "CREATED"  *# 订单状态：CREATED → PAID → SHIPPED → DELIVERED*
        *self*.created_at = datetime.now()

    def **add_item**(*self*, *book_id*, *quantity*, *unit_price*):
        """添加订单项（领域行为）"""
        *self*.items.append(OrderItem(*book_id*, *quantity*, *unit_price*))

    def **calculate_total**(*self*):
        """计算订单总价（领域行为）"""
        total = Money(0)
        *for* item *in* *self*.items:
            total += item.total_price
        *return* total

    def **confirm_payment**(*self*):
        """确认支付（领域行为，包含状态转换规则）"""
        *if* *self*.status != "CREATED":
            *raise* Exception(f"订单状态为{*self*.status}，无法确认支付")
        *self*.status = "PAID"
        *self*.paid_at = datetime.now()

    def **__repr__**(*self*):
        *return* f"Order(id={*self*.order_id}, status={*self*.status}, total={*self*.calculate_total()})"

class OrderRepository:
    """订单仓储（负责订单的持久化）"""
    def **__init__**(*self*):
        *self*.orders = {}  *# 模拟数据库*

    def **save**(*self*, *order*):
        """保存订单"""
        *self*.orders[*order*.order_id] = *order*

    def **get_by_id**(*self*, *order_id*):
        """根据ID获取订单"""
        *return* *self*.orders.get(*order_id*)

class OrderDomainService:
    """订单领域服务（协调订单相关的领域对象）"""
    def **__init__**(*self*, *repository*):
        *self*.repository = *repository*

    def **create_order**(*self*, *user_id*, *book_items*):
        """创建订单"""
        order_id = f"order_{len(*self*.repository.orders) + 1}"
        order = Order(order_id, *user_id*)

        *for* item *in* *book_items*:
            order.add_item(item["book_id"], item["quantity"], item["unit_price"])

        *self*.repository.save(order)
        *return* order

*# ------------------------------------------------------*
*# 3. 库存子域 - 包含库存相关的领域模型*
*# ------------------------------------------------------*
class Inventory:
    """库存聚合根"""
    def **__init__**(*self*, *book_id*, *quantity*):
        *self*.book_id = *book_id*  *# BookId对象*
        *self*.quantity = *quantity*  *# 当前库存数量*

    def **check_availability**(*self*, *required_quantity*):
        """检查库存是否充足（领域行为）"""
        *return* *self*.quantity >= *required_quantity*

    def **reserve**(*self*, *quantity*):
        """预留库存（领域行为）"""
        *if* not *self*.check_availability(*quantity*):
            *return* False
        *self*.quantity -= *quantity*
        *return* True

    def **release**(*self*, *quantity*):
        """释放库存（领域行为）"""
        *self*.quantity += *quantity*
        *return* True

    def **__repr__**(*self*):
        *return* f"Inventory(book={*self*.book_id}, quantity={*self*.quantity})"

class InventoryRepository:
    """库存仓储"""
    def **__init__**(*self*):
        *# 初始化库存数据*
        *self*.inventories = {
            BookId("Python编程"): Inventory(BookId("Python编程"), 100),
            BookId("DDD实战"): Inventory(BookId("DDD实战"), 50),
            BookId("设计模式"): Inventory(BookId("设计模式"), 30)
        }

    def **get_by_book_id**(*self*, *book_id*):
        """根据书籍ID获取库存"""
        *return* *self*.inventories.get(*book_id*)

    def **save**(*self*, *inventory*):
        """保存库存信息"""
        *self*.inventories[*inventory*.book_id] = *inventory*

class InventoryDomainService:
    """库存领域服务"""
    def **__init__**(*self*, *repository*):
        *self*.repository = *repository*

    def **check_and_reserve**(*self*, *book_id*, *quantity*):
        """检查并预留库存"""
        inventory = *self*.repository.get_by_book_id(*book_id*)
        *if* not inventory:
            *raise* Exception(f"书籍 {*book_id*} 不存在")

        *if* inventory.reserve(*quantity*):
            *self*.repository.save(inventory)
            *return* True
        *return* False

    def **release_stock**(*self*, *book_id*, *quantity*):
        """释放库存"""
        inventory = *self*.repository.get_by_book_id(*book_id*)
        *if* inventory:
            inventory.release(*quantity*)
            *self*.repository.save(inventory)
            *return* True
        *return* False

*# ------------------------------------------------------*
*# 4. 支付子域和物流子域（简化实现）*
*# ------------------------------------------------------*
class PaymentDomainService:
    """支付领域服务"""
    def **process_payment**(*self*, *user_id*, *order_id*, *amount*):
        """处理支付"""
        print(f"处理用户 {*user_id*} 的订单 {*order_id*} 支付，金额: {*amount*}")
        *return* {"success": True, "transaction_id": f"trans_{*order_id*}"}

class LogisticsDomainService:
    """物流领域服务"""
    def **create_shipment**(*self*, *order_id*, *user_id*, *items*):
        """创建物流单"""
        shipment_id = f"ship_{*order_id*}"
        print(f"创建物流单 {shipment_id}，订单 {*order_id*}，用户 {*user_id*}")
        *return* shipment_id

*# ------------------------------------------------------*
*# 5. 应用服务 - 协调各子域完成业务流程*
*# ------------------------------------------------------*
class OrderApplicationService:
    """订单应用服务（协调不同子域完成下单流程）"""
    def **__init__**(*self*, *order_service*, *inventory_service*, *payment_service*, *logistics_service*):
        *self*.order_service = *order_service*
        *self*.inventory_service = *inventory_service*
        *self*.payment_service = *payment_service*
        *self*.logistics_service = *logistics_service*

    def **place_order**(*self*, *user_id*, *book_ids_with_quantity*):
        """处理下单流程（协调各子域）"""
        *# 1. 准备书籍信息（从库存子域获取）*
        book_items = []
        *for* book_info *in* *book_ids_with_quantity*:
            book_id = BookId(book_info["id"])
            quantity = book_info["quantity"]

            *# 获取书籍库存信息*
            inventory = *self*.inventory_service.repository.get_by_book_id(book_id)
            *if* not inventory:
                *raise* Exception(f"书籍《{book_id}》不存在")

            *# 假设价格存储在库存中（实际中可能有专门的商品子域）*
            book_items.append({
                "book_id": book_id,
                "quantity": quantity,
                "unit_price": Money(89 *if* book_id.value == "Python编程" *else*
                                   79 *if* book_id.value == "DDD实战" *else* 69)
            })

        *# 2. 创建订单（订单子域）*
        order = *self*.order_service.create_order(*user_id*, book_items)
        print(f"创建订单: {order}")

        *try*:
            *# 3. 预留库存（库存子域）*
            *for* item *in* book_items:
                *if* not *self*.inventory_service.check_and_reserve(item["book_id"], item["quantity"]):
                    *raise* Exception(f"书籍《{item['book_id']}》库存不足")

            *# 4. 处理支付（支付子域）*
            total_amount = order.calculate_total()
            payment_result = *self*.payment_service.process_payment(
                *user_id*, order.order_id, total_amount
            )
            *if* not payment_result["success"]:
                *raise* Exception("支付失败")

            *# 5. 确认订单支付状态（订单子域的领域行为）*
            order.confirm_payment()
            *self*.order_service.repository.save(order)
            print(f"订单 {order.order_id} 支付确认成功")

            *# 6. 创建物流单（物流子域）*
            *self*.logistics_service.create_shipment(order.order_id, *user_id*, book_items)

            *return* order

        *except* Exception *as* e:
            *# 发生异常，释放已预留的库存*
            *for* item *in* book_items:
                *self*.inventory_service.release_stock(item["book_id"], item["quantity"])
            print(f"下单失败: {str(e)}, 已回滚库存")
            *raise*

*# 业务流程演示*
*if* __name__ == "__main__":
    *from* datetime *import* datetime  *# 导入datetime模块*

    *# 初始化各子域的仓储和服务*
    order_repo = OrderRepository()
    order_service = OrderDomainService(order_repo)

    inventory_repo = InventoryRepository()
    inventory_service = InventoryDomainService(inventory_repo)

    payment_service = PaymentDomainService()
    logistics_service = LogisticsDomainService()

    *# 初始化应用服务*
    order_app_service = OrderApplicationService(
        order_service, inventory_service, payment_service, logistics_service
    )

    *# 执行下单流程*
    user_id = "user_001"
    books_to_buy = [
        {"id": "Python编程", "quantity": 2},
        {"id": "DDD实战", "quantity": 1}
    ]

    *try*:
        order = order_app_service.place_order(user_id, books_to_buy)
        print(f"下单成功: {order}")

        *# 查询订单*
        retrieved_order = order_repo.get_by_id(order.order_id)
        print(f"查询订单: {retrieved_order}")
    *except* Exception *as* e:
        print(f"操作失败: {str(e)}")

这种方式按照业务领域边界划分，通过领域模型封装业务规则，各子域通过限界上下文隔离。

DDD 核心概念

上述实现中包含了多个核心概念，体现在代码的各个部分：

子域（Subdomain）：
- 订单子域：包含Order、OrderItem等类，处理与订单相关的所有业务
- 库存子域：包含Inventory等类，专注于库存管理
- 支付子域和物流子域：处理各自领域的业务逻辑
限界上下文（Bounded Context）：
- 每个子域构成一个限界上下文，有明确的边界
- 上下文内的领域模型独立演化，不直接依赖其他上下文的内部实现
领域模型（Domain Model）：
- 实体（Entity）：如Order、Inventory，有唯一标识，状态可变化
- 值对象（Value Object）：如Money、BookId、OrderItem，无唯一标识，不可变，用于描述事物的属性
聚合（Aggregate）与聚合根（Aggregate Root）：
- Order是聚合根，包含多个OrderItem值对象，形成一个订单聚合
- Inventory是聚合根，代表一个库存聚合
- 聚合根负责维护聚合内的业务规则和一致性
领域服务（Domain Service）：
- 如OrderDomainService、InventoryDomainService
- 用于协调领域对象完成单个领域内的复杂操作
应用服务（Application Service）：
- 如OrderApplicationService
- 不包含业务规则，仅负责协调多个子域完成跨领域的业务流程
仓储（Repository）：
- 如OrderRepository、InventoryRepository
- 负责领域对象的持久化，隔离领域模型与数据访问逻辑
通用语言（Ubiquitous Language）：
- 代码中的类名、方法名直接反映业务术语（如confirm_payment、reserve）
- 开发团队和业务人员使用统一的语言沟通

两种划分方式的核心区别

对比维度	业务功能划分	DDD子领域划分
核心组织单元	功能模块（如订单服务、库存服务）	子域和领域模型（如订单子域、库存子域）
业务规则位置	分散在服务的流程方法中	封装在领域对象的行为中
服务职责	直接实现完整业务流程	领域服务专注单一领域逻辑，应用服务协调跨域流程
变更影响范围	一处变更可能影响多个服务	变更通常局限在单个子域内
与业务的映射关系	映射业务功能	映射业务本质和规则
适用场景	简单业务、稳定需求	复杂业务、频繁变化的需求

通过这两份代码可以看出，DDD虽然在初始实现时需要更多代码，但它通过清晰的领域边界和封装的业务规则，为系统的长期演进和维护提供了更好的支持。当业务需求发生变化时，DDD的设计能将变更控制在最小范围内，大大降低维护成本。

在线书店业务的需求迭代

下面我们还是以在线书店业务为例，从具体的业务需求迭代中感受两者的差异

初始场景（基础版）

功能：创建订单→扣库存→支付→生成物流单
规则：无折扣、无库存预留时效、不支持拆单

变化1：新增会员折扣规则（规则类变化）

需求：会员下单享受9折，非会员无折扣
规则：折扣计算需依赖用户会员状态

传统方式的修改（耦合问题初现）

传统设计中，价格计算逻辑散落在订单服务的流程里，修改需侵入核心流程：

# 传统OrderService.create_order（需修改）
def create_order(self, user_id, books):
    # ... 原有逻辑 ...

    # 新增：计算折扣（直接嵌入订单流程）
    user = self.user_service.get_user(user_id)  # 新增依赖
    discount = 0.9 if user.is_member() else 1.0  # 业务规则散落在流程中

    # 原有总价计算需修改
    total_amount = 0
    *for* book *in* *books*:
        book_info = *self*.inventory_service.get_book_info(book["id"])
        *if* not book_info:
            *raise* Exception(f"书籍《{book['id']}》不存在")
        total_amount += book_info["price"] * book["quantity"] * discount
    # ... 后续支付逻辑依赖total_amount ...

变更影响范围：

直接修改核心流程方法create_order（风险高，可能引入 bug）
订单服务新增对用户服务的依赖（耦合度上升）
若后续折扣规则调整（如会员等级折扣），需再次修改create_order方法（规则与流程强绑定）。

DDD方式的修改（规则内聚）

DDD中，价格计算是订单子域的核心规则，修改被限制在订单模型内：

# 订单子域的Order类（仅修改此处）
class Order:
    def __init__(self, order_id, user_id, user_service):  # 依赖注入用户服务
        self.user_service = user_service  # 领域模型可依赖其他子域的接口
        # ... 原有属性 ...

    def **calculate_total**(*self*):
        """计算订单总价（领域行为）"""
        total = Money(0)
        *for* item *in* *self*.items:
            total += item.total_price

        """修改总价计算逻辑，封装折扣规则"""
        base_total = sum(item.total_price for item in self.items)
        # 调用用户服务接口（不依赖实现）
        user = self.user_service.get_user(self.user_id)
        *return* total * (0.9 if user.is_member() else 1.0)

变更影响范围：

仅修改订单子域的Order类（规则内聚，不影响流程）
订单服务与用户服务通过接口交互，无紧耦合（依赖注入）
库存、支付、物流子域完全无感知（零修改）

变化2：库存预留时效（子域规则变化）

需求：库存扣减后需预留15分钟，超时未支付自动释放
规则：需记录库存预留时间，定时检查释放

传统方式的修改（耦合扩散）

传统设计中，库存服务需新增时效管理，且订单服务需感知这一变化：

# 1. 传统InventoryService需大幅修改
class InventoryService:
    def reduce_stock(self, book_id, quantity):
        # ... 原有逻辑 ...

        # 新逻辑：记录预留时间
        reservation_id = f"res_{book_id}_{time.time()}"
        self.reservations[reservation_id] = {
            "book_id": book_id,
            "quantity": quantity,
            "expire_time": time.time() + 900  # 15分钟
        }
        return reservation_id  # 新增返回值（影响所有调用者）

# 2. 传统OrderService需同步修改（耦合扩散）
def create_order(self, user_id, books):
    # ... 原有逻辑 ...
    # 扣库存时需保存reservation_id，以便后续确认或释放
    reservation_ids = []  # 新增变量
    for book in books:
        res_id = self.inventory_service.reduce_stock(book["id"], book["quantity"])  # 接口变更
        reservation_ids.append(res_id)

    # 支付成功后需确认库存预留
    if payment_result["success"]:
        for res_id in reservation_ids:
            self.inventory_service.confirm_reservation(res_id)  # 新增接口调用
    else:
        for res_id in reservation_ids:
            self.inventory_service.release_reservation(res_id)  # 新增接口调用

变更影响范围：

库存服务接口变更（reduce_stock返回值修改），所有调用该接口的服务（订单、退货等）必须同步修改
订单服务的create_order方法新增大量与库存预留相关的逻辑（职责边界模糊，订单服务被迫处理库存细节）
若后续调整时效（如改为 30 分钟），需同时检查库存服务和所有依赖它的服务（维护成本高）

DDD方式的修改（边界隔离）

DDD中，库存预留是库存子域的内部规则，订单子域无需感知细节：

# 1. 库存子域：仅内部扩展，接口不变
class Inventory:
    def __init__(self, book_id, quantity):
        self.book_id = book_id
        self.quantity = quantity
        self.reservations = []  # 新增：存储预留记录（内部属性）

    def reserve(self, quantity):  # 方法逻辑修改，接口参数不变
        expire_time = time.time() + 900  # 15分钟
        self.reservations.append({"quantity": quantity, "expire_time": expire_time, "status": "PENDING"})
        self.quantity -= quantity
        return True

    def release_expired(self):  # 新增内部方法，不影响外部
        now = time.time()
        for res in self.reservations:
            if res["status"] == "PENDING" and res["expire_time"] < now:
                self.quantity += res["quantity"]
                res["status"] = "EXPIRED"

# 2. 库存领域服务：接口不变，内部调用新方法
class InventoryDomainService:
    def check_and_reserve(self, book_id, quantity):
        inventory = self.repository.get_by_book_id(book_id)
        inventory.release_expired()  # 新增内部调用
        return inventory.reserve(quantity)  # 接口参数/返回值不变

变更影响范围：

仅修改库存子域的Inventory类和InventoryDomainService（完全内聚）
订单子域、支付子域、应用服务均无需修改（接口稳定，无感知）
若调整时效，仅需修改Inventory类中的expire_time计算（零扩散风险）

变化3：订单拆分（流程类变化）

需求：订单中不同仓库的书籍需拆分为多个子订单，分别发货
规则：父订单创建后拆分子订单，各子订单独立扣库存、独立物流

传统方式的修改（耦合爆炸）

传统设计中，订单服务需处理拆分逻辑，且与库存、物流服务深度耦合：

# 1. 订单服务：核心流程完全重构（风险极高）
class OrderService:
    def create_order(self, user_id, books):
        # 新增：按仓库拆分书籍（依赖仓库服务，新增耦合）
        warehouse_books = self.warehouse_service.group_by_warehouse(books)

        # 新增：创建父订单和子订单（流程复杂度激增）
        parent_order_id = f"parent_{user_id}"
        child_orders = [{"id": f"child_{parent_order_id}_{i}", "books": books}
                       for i, (_, books) in enumerate(warehouse_books.items())]

        # 库存服务需按仓库扣减（接口必须修改）
        for child in child_orders:
            for book in child["books"]:
                self.inventory_service.reduce_stock(book["id"], book["quantity"], child["warehouse"])

        # 物流服务需按子订单创建（接口必须修改）
        for child in child_orders:
            self.logistics_service.create_shipment(child["id"], user_id, child["books"], child["warehouse"])

变更影响范围：

订单服务的create_order方法完全重构（原有逻辑被颠覆，易引入 bug）
库存服务、物流服务必须修改接口以支持仓库参数（耦合扩散至多个服务）
支付服务虽不修改，但需理解父订单与子订单的关系（认知成本上升）
后续若新增如 “子订单合并” 等需求，需再次重构所有相关服务（扩展性极差）

DDD方式的修改（聚焦核心变更）

在订单子域中内聚拆分规则，没有侵入到流程类或者应用服务中

# 订单子域：仅新增类，不修改原有Order
class ParentOrder:
    def __init__(self, order_id, user_id):
        self.order_id = order_id
        self.user_id = user_id
        self.child_orders = []  # 子订单集合（聚合内管理）

    def split_by_warehouse(self, warehouse_map):
        """按仓库拆分（核心规则内聚在订单子域）"""
        # 1. 按仓库分组订单项（规则封装）
        warehouse_items = self._group_items_by_warehouse(warehouse_map)

        # 2. 为每个仓库创建子订单（内部行为）
        for warehouse_id, items in warehouse_items.items():
            child = ChildOrder(
                order_id=f"child_{self.order_id}_{warehouse_id}",
                user_id=self.user_id,
                parent_id=self.order_id,
                warehouse_id=warehouse_id
            )
            child.add_items(items)  # 复用Order的基础能力
            self.child_orders.append(child)

    def calculate_total(self):
        """聚合级规则：父订单总价=子订单总和"""
        return sum(child.calculate_total() for child in self.child_orders)


class ChildOrder(Order):
    """继承基础订单能力，仅新增必要属性"""
    def __init__(self, order_id, user_id, parent_id, warehouse_id):
        super().__init__(order_id, user_id)  # 复用原有订单逻辑
        self.parent_id = parent_id
        self.warehouse_id = warehouse_id  # 子订单特有属性

在库存子域中新增接口，保持向后兼容

# 库存子域：仅扩展接口，核心逻辑不变
class InventoryDomainService:
    # 原有接口保留（兼容非拆分订单）
    def check_and_reserve(self, book_id, quantity):
        pass  # 核心逻辑不变

    # 新增接口：支持仓库维度（扩展不修改）
    def check_and_reserve_by_warehouse(self, book_id, quantity, warehouse_id):
        """按仓库预留库存（仅新增逻辑，不影响原有方法）"""
        inventory = self.repository.get_by(book_id, warehouse_id)  # 按仓库查询
        return inventory.reserve(quantity)  # 复用原有reserve核心逻辑

在应用服务中，仅协调不包含子域规则

# 应用服务：纯流程协调，无业务规则
class OrderApplicationService:
    def place_order(self, user_id, books):
        # 1. 创建父订单（调用订单子域）
        parent_order = self.order_service.create_parent_order(user_id，books)

        # 2. 拆分订单（调用订单子域的拆分能力）
        warehouse_map = self.warehouse_service.get_mapping()  # 依赖抽象接口
        parent_order.split_by_warehouse(warehouse_map)

        # 3. 协调库存、支付、物流子域（仅调用接口）
        for child in parent_order.child_orders:
            self.inventory_service.check_and_reserve_by_warehouse(
                child.items, child.warehouse_id  # 传递仓库信息
            )

        self.payment_service.pay(parent_order.calculate_total())  # 父单统一支付
        self.logistics_service.create_shipments(parent_order.child_orders)  # 子单分物流

变更影响范围：

订单拆分仅修改订单子域（新增 2 个类）和库存子域（新增 1 个接口），支付、物流等子域完全不动。对比业务拆分模式中 "改一处动全身"，DDD 通过子域边界隔离了变化。（变更范围小）
拆分逻辑（按仓库分组、子单创建）封装在ParentOrder中，而非散落在应用服务的流程里。后续若调整拆分规则（如按重量拆分），仅需修改split_by_warehouse方法，其他代码不受影响（业务规则内聚）
支付子域完全无修改（无需协调更多的团队，更加敏捷）
类名和方法名直接反映业务术语（ParentOrder、split_by_warehouse），业务人员能直接理解代码意图；父子订单的关系和拆分规则通过代码结构显性化，而非隐藏在复杂的流程逻辑中。（业务代码直接映射，降低团队的协作成本）

为什么 DDD 在简单场景看不到优势？

从以上示例中可以看到 DDD 的这么多显著优势，那么为什么实际开发中，很多时候我们并没有选择使用 DDD 方法论来进行领域建模呢？

DDD 的价值与业务复杂度和变化频率正相关：

当业务简单（规则少、流程固定）：传统方式开发更快，DDD的“领域建模”反而显得冗余
当业务复杂（多规则、跨流程、高频变更）：DDD通过“领域模型封装规则+限界上下文隔离变化”，让每次修改的影响范围最小化

就像建房子：

盖小木屋（简单业务）：直接钉木板最快，无需设计图纸
盖摩天大楼（复杂业务）：必须先有结构设计（领域模型）和分区规划（限界上下文），否则每加一层都可能塌

为什么 DDD 和微服务架构天然契合？

DDD 的子域和限界上下文的概念和微服务架构的服务恰好匹配
DDD 的领域模型由独立团队开发这与微服务架构中自治化团队负责一个松散耦合的微服务的想法不谋而合
子域有用于自己的领域模型，这可以帮助消灭代码中的上帝类，帮助进行服务拆分

总结：DDD的优势在长期演进中显现

传统方式的代码在需求迭代中极易发展成 “大泥球”：服务间互相依赖、业务规则散落在流程中、每次修改都要牵动多个服务
DDD方式的代码仍保持清晰：业务规则在领域模型中内聚、子域边界明确、变化被限制在最小范围内。

这就是DDD的核心价值 —— 让复杂系统在持续变化中保持可维护性。对于需要长期迭代的业务系统（比如电商、金融、ERP），这种优势会随着时间推移愈发明显。