分布式事务场景下基于事件溯源的一致性边界设计模式

distributed-transaction-es/
├── order_service/          # 订单服务（事件溯源核心）
│   ├── __init__.py
│   ├── aggregates/         # 聚合根
│   │   ├── __init__.py
│   │   └── order.py       # Order聚合根
│   ├── commands/          # 命令及其处理器
│   │   ├── __init__.py
│   │   ├── handlers.py
│   │   └── models.py      # Command对象
│   ├── queries/           # 查询及其处理器
│   │   ├── __init__.py
│   │   ├── handlers.py
│   │   └── models.py      # Query对象
│   ├── events/            # 领域事件与集成事件
│   │   ├── __init__.py
│   │   ├── domain_events.py
│   │   └── integration_events.py
│   ├── infrastructure/    # 基础设施
│   │   ├── __init__.py
│   │   ├── event_store.py # 事件存储
│   │   ├── message_bus.py # 简易消息总线
│   │   └── unit_of_work.py
│   └── app.py             # 服务入口与API
├── inventory_service/     # 库存服务（事件消费者）
│   ├── __init__.py
│   ├── event_handlers.py # 集成事件处理器
│   └── app.py
├── shared/                # 共享模型
│   └── models.py
├── requirements.txt
└── run.py                 # 主运行脚本

# 共享的数据模型
from dataclasses import dataclass
from typing import List
from uuid import UUID, uuid4

@dataclass
class OrderItem:
    """订单项"""
    product_id: str
    product_name: str
    unit_price: float
    quantity: int

    @property
    def total_price(self) -> float:
        return self.unit_price * self.quantity

# 订单聚合根 - 事件溯源的核心
from dataclasses import dataclass, field
from typing import List, Type
from uuid import UUID, uuid4
import json

from shared.models import OrderItem
from ..events.domain_events import (
    OrderCreatedEvent, OrderItemAddedEvent,
    DomainEvent
)

@dataclass
class Order:
    """订单聚合根。
    状态通过应用事件来改变，而不是直接设置字段。
    """
    id: UUID
    customer_id: str
    status: str = "PENDING"  # PENDING, CONFIRMED, CANCELLED
    items: List[OrderItem] = field(default_factory=list)
    total_amount: float = 0.0

    # 用于事件溯源的内部列表
    _changes: List[DomainEvent] = field(default_factory=list, init=False, repr=False)
    _version: int = field(default=-1, init=False, repr=False) # 当前聚合版本（基于事件）

    @classmethod
    def create(cls, customer_id: str) -> tuple['Order', List[DomainEvent]]:
        """工厂方法：创建一个新订单。
        返回聚合实例和产生的事件列表。
        """
        order_id = uuid4()
        order = cls(id=order_id, customer_id=customer_id)
        # 记录领域事件
        event = OrderCreatedEvent(
            aggregate_id=order_id,
            customer_id=customer_id
        )
        order._apply_event(event)
        order._record_change(event)
        return order, order._changes

    def add_item(self, product_id: str, product_name: str, unit_price: float, quantity: int):
        """向订单添加商品。这是聚合上的一个命令方法。"""
        if self.status != "PENDING":
            raise ValueError("只能在PENDING状态修改订单")
        # 业务规则：检查重复商品等（此处简化）
        event = OrderItemAddedEvent(
            aggregate_id=self.id,
            product_id=product_id,
            product_name=product_name,
            unit_price=unit_price,
            quantity=quantity
        )
        self._apply_event(event)
        self._record_change(event)

    def confirm(self):
        """确认订单（简化，未实现对应事件）"""
        self.status = "CONFIRMED"

    def _apply_event(self, event: DomainEvent):
        """应用一个事件到聚合的当前状态。这是状态重建的关键。"""
        if isinstance(event, OrderCreatedEvent):
            self.id = event.aggregate_id
            self.customer_id = event.customer_id
            self.status = "PENDING"
        elif isinstance(event, OrderItemAddedEvent):
            new_item = OrderItem(
                product_id=event.product_id,
                product_name=event.product_name,
                unit_price=event.unit_price,
                quantity=event.quantity
            )
            # 简化逻辑：直接添加，实际可能需要合并数量
            self.items.append(new_item)
            self.total_amount += new_item.total_price
        # 可以处理更多类型的事件...

    def _record_change(self, event: DomainEvent):
        """记录一个新产生的事件"""
        self._changes.append(event)

    def mark_changes_as_committed(self):
        """提交后清空未保存的变更列表"""
        self._changes.clear()

    @classmethod
    def load_from_history(cls, aggregate_id: UUID, history: List[DomainEvent]) -> 'Order':
        """从历史事件流中重建聚合状态。
        这是事件溯源的精华：状态不是读出来的，而是算出来的。
        """
        # 创建一个"空"聚合实例。ID是必须的，其他字段由事件填充。
        order = cls(id=aggregate_id, customer_id="") # customer_id会被第一个事件覆盖
        for event in history:
            order._apply_event(event)
            order._version += 1  # 每应用一个事件，版本+1
        order.mark_changes_as_committed()
        return order

    @property
    def changes(self) -> List[DomainEvent]:
        return self._changes.copy()

# 领域事件：发生在聚合内部，用于记录状态变更
from dataclasses import dataclass
from datetime import datetime
from typing import Any, Dict
from uuid import UUID
import json

@dataclass
class DomainEvent:
    """所有领域事件的基类"""
    aggregate_id: UUID
    occurred_on: datetime = None
    event_type: str = None

    def __post_init__(self):
        if self.occurred_on is None:
            self.occurred_on = datetime.utcnow()
        if self.event_type is None:
            self.event_type = self.__class__.__name__

    def to_dict(self) -> Dict[str, Any]:
        """序列化事件，用于存储"""
        data = self.__dict__.copy()
        data['aggregate_id'] = str(data['aggregate_id'])
        data['occurred_on'] = data['occurred_on'].isoformat()
        return data

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'DomainEvent':
        """反序列化（简化版，生产环境需更健壮）"""
        data['aggregate_id'] = UUID(data['aggregate_id'])
        data['occurred_on'] = datetime.fromisoformat(data['occurred_on'])
        # 移除event_type，因为它是类名
        data.pop('event_type', None)
        return cls(**data)

@dataclass
class OrderCreatedEvent(DomainEvent):
    customer_id: str

@dataclass
class OrderItemAddedEvent(DomainEvent):
    product_id: str
    product_name: str
    unit_price: float
    quantity: int

# 集成事件：用于跨服务通信，通常由领域事件触发
from dataclasses import dataclass
from datetime import datetime
from typing import List
from uuid import UUID
from shared.models import OrderItem

@dataclass
class IntegrationEvent:
    """集成事件基类"""
    event_id: UUID
    occurred_on: datetime
    event_type: str

    def __init__(self):
        self.event_id = UUID()
        self.occurred_on = datetime.utcnow()
        self.event_type = self.__class__.__name__

@dataclass
class OrderCreatedIntegrationEvent(IntegrationEvent):
    """订单创建集成事件，库存服务将消费此事件"""
    order_id: UUID
    customer_id: str
    items: List[OrderItem]

# 事件存储 - 核心基础设施
from typing import List, Optional
from uuid import UUID
import json

from ..events.domain_events import DomainEvent

class EventStore:
    """简化的事件存储，使用内存字典模拟。
    生产环境应使用专门数据库（如EventStoreDB, DynamoDB流等）。
    """
    def __init__(self):
        self._events = {}  # aggregate_id -> list of event dicts

    def save(self, aggregate_id: UUID, events: List[DomainEvent], expected_version: int = -1):
        """保存聚合产生的一系列新事件。
        expected_version 用于乐观并发控制。
        """
        if aggregate_id not in self._events:
            self._events[aggregate_id] = []
        current_stream = self._events[aggregate_id]
        # 简单的乐观锁检查：当前流长度应等于 expected_version + 1
        if expected_version != -1 and len(current_stream) != expected_version + 1:
            raise RuntimeError(f"并发冲突：聚合 {aggregate_id} 的期望版本是 {expected_version}, 但实际是 {len(current_stream)-1}")
        for event in events:
            event_data = event.to_dict()
            current_stream.append(event_data)

    def load(self, aggregate_id: UUID) -> List[DomainEvent]:
        """加载特定聚合的所有事件"""
        event_dicts = self._events.get(aggregate_id, [])
        events = []
        for ed in event_dicts:
            # 根据event_type动态还原事件类（简化，实际需注册表）
            class_name = ed['event_type']
            if class_name == 'OrderCreatedEvent':
                cls = OrderCreatedEvent
            elif class_name == 'OrderItemAddedEvent':
                cls = OrderItemAddedEvent
            else:
                raise ValueError(f"未知事件类型: {class_name}")
            events.append(cls.from_dict(ed))
        return events

    def get_stream_version(self, aggregate_id: UUID) -> int:
        """获取聚合事件流的当前版本（最后事件的索引）"""
        stream = self._events.get(aggregate_id, [])
        return len(stream) - 1  # -1 表示不存在

# 简易内存消息总线，用于进程内事件发布/订阅。
# 生产环境应替换为 RabbitMQ, Kafka, AWS SNS/SQS 等。
from typing import Dict, List, Callable, Any
import logging

logger = logging.getLogger(__name__)

class MessageBus:
    def __init__(self):
        self._subscribers: Dict[str, List[Callable]] = {}

    def publish(self, event_type: str, event: Any):
        """发布事件"""
        logger.info(f"发布事件: {event_type} - {event}")
        if event_type in self._subscribers:
            for handler in self._subscribers[event_type]:
                try:
                    handler(event)
                except Exception as e:
                    logger.error(f"事件处理器错误: {e}", exc_info=True)

    def subscribe(self, event_type: str, handler: Callable):
        """订阅事件"""
        if event_type not in self._subscribers:
            self._subscribers[event_type] = []
        self._subscribers[event_type].append(handler)

# 全局消息总线实例
message_bus = MessageBus()

# 命令对象
from dataclasses import dataclass
from typing import List
from uuid import UUID

@dataclass
class CreateOrderCommand:
    customer_id: str

@dataclass
class AddOrderItemCommand:
    order_id: UUID
    product_id: str
    product_name: str
    unit_price: float
    quantity: int

# 命令处理器
from uuid import UUID

from ..aggregates.order import Order
from ..infrastructure.event_store import EventStore
from ..infrastructure.message_bus import message_bus
from ..events.integration_events import OrderCreatedIntegrationEvent
from .models import CreateOrderCommand, AddOrderItemCommand

class CommandHandler:
    def __init__(self, event_store: EventStore):
        self.event_store = event_store

    def handle_create_order(self, cmd: CreateOrderCommand) -> UUID:
        """处理创建订单命令"""
        # 1. 通过聚合的工厂方法创建聚合，同时产生领域事件
        order, events = Order.create(cmd.customer_id)

        # 2. 保存事件到事件存储
        expected_version = self.event_store.get_stream_version(order.id)
        self.event_store.save(order.id, events, expected_version)

        # 3. 发布集成事件（在真实场景中，可能由单独的进程监听领域事件后发布）
        # 此处简化：直接生成并发布
        integration_event = OrderCreatedIntegrationEvent()
        integration_event.order_id = order.id
        integration_event.customer_id = order.customer_id
        integration_event.items = order.items.copy()
        message_bus.publish('OrderCreatedIntegrationEvent', integration_event)

        # 4. 标记聚合变更已提交
        order.mark_changes_as_committed()

        return order.id

    def handle_add_order_item(self, cmd: AddOrderItemCommand):
        """处理添加订单项命令"""
        # 1. 从事件存储加载历史，重建聚合
        history = self.event_store.load(cmd.order_id)
        if not history:
            raise ValueError(f"订单 {cmd.order_id} 不存在")
        order = Order.load_from_history(cmd.order_id, history)

        # 2. 在聚合上执行操作，产生新事件
        order.add_item(cmd.product_id, cmd.product_name, cmd.unit_price, cmd.quantity)

        # 3. 保存新产生的事件
        expected_version = self.event_store.get_stream_version(order.id)
        self.event_store.save(order.id, order.changes, expected_version)

        # 4. 标记聚合变更已提交
        order.mark_changes_as_committed()

# queries/models.py
from dataclasses import dataclass
from uuid import UUID

@dataclass
class GetOrderQuery:
    order_id: UUID

# queries/handlers.py
# 读模型处理器（CQRS的查询端）
from uuid import UUID
from typing import Optional, Dict, Any

# 简单的内存读模型存储
_read_model_store: Dict[UUID, Dict[str, Any]] = {}

class QueryHandler:
    @staticmethod
    def handle_get_order(query) -> Optional[Dict]:
        """处理查询：从读模型获取订单视图"""
        return _read_model_store.get(query.order_id)

    @staticmethod
    def update_read_model(order_id: UUID, snapshot: Dict[str, Any]):
        """更新读模型（由事件处理器调用）"""
        _read_model_store[order_id] = snapshot

# 订单服务 FastAPI 应用
from uuid import UUID
from fastapi import FastAPI, HTTPException
import logging

from .infrastructure.event_store import EventStore
from .infrastructure.message_bus import message_bus
from .commands.handlers import CommandHandler
from .commands.models import CreateOrderCommand, AddOrderItemCommand
from .queries.handlers import QueryHandler, _read_model_store
from .queries.models import GetOrderQuery
from .events.domain_events import OrderCreatedEvent, OrderItemAddedEvent

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 初始化基础设施和处理器
event_store = EventStore()
command_handler = CommandHandler(event_store)

# 创建FastAPI应用
app = FastAPI(title="Order Service (Event Sourcing)")

# --- 事件处理器（用于更新读模型）---
# 这些处理器订阅领域事件，并更新为查询优化的读模型
def on_order_created(event: OrderCreatedEvent):
    """当订单创建时，初始化读模型条目"""
    snapshot = {
        'order_id': event.aggregate_id,
        'customer_id': event.customer_id,
        'status': 'PENDING',
        'items': [],
        'total_amount': 0.0,
        'version': 0
    }
    QueryHandler.update_read_model(event.aggregate_id, snapshot)
    logger.info(f"读模型更新: 订单 {event.aggregate_id} 已创建")

def on_order_item_added(event: OrderItemAddedEvent):
    """当订单项添加时，更新读模型"""
    snapshot = _read_model_store.get(event.aggregate_id)
    if snapshot:
        new_item = {
            'product_id': event.product_id,
            'product_name': event.product_name,
            'unit_price': event.unit_price,
            'quantity': event.quantity
        }
        snapshot['items'].append(new_item)
        snapshot['total_amount'] += event.unit_price * event.quantity
        snapshot['version'] += 1
        logger.info(f"读模型更新: 订单 {event.aggregate_id} 添加商品 {event.product_id}")

# 订阅领域事件到内存总线（进程内）
message_bus.subscribe('OrderCreatedEvent', on_order_created)
message_bus.subscribe('OrderItemAddedEvent', on_order_item_added)

# --- HTTP API 端点 ---
@app.post("/orders", response_model=dict)
def create_order(customer_id: str):
    """创建新订单"""
    try:
        cmd = CreateOrderCommand(customer_id=customer_id)
        order_id = command_handler.handle_create_order(cmd)
        return {"order_id": order_id, "message": "Order created successfully"}
    except Exception as e:
        logger.exception("创建订单失败")
        raise HTTPException(status_code=400, detail=str(e))

@app.post("/orders/{order_id}/items")
def add_order_item(
    order_id: UUID,
    product_id: str,
    product_name: str,
    unit_price: float,
    quantity: int
):
    """向订单添加商品"""
    try:
        cmd = AddOrderItemCommand(
            order_id=order_id,
            product_id=product_id,
            product_name=product_name,
            unit_price=unit_price,
            quantity=quantity
        )
        command_handler.handle_add_order_item(cmd)
        return {"message": "Item added successfully"}
    except ValueError as e:
        raise HTTPException(status_code=404, detail=str(e))
    except Exception as e:
        logger.exception("添加订单项失败")
        raise HTTPException(status_code=400, detail=str(e))

@app.get("/orders/{order_id}")
def get_order(order_id: UUID):
    """查询订单（从读模型）"""
    query = GetOrderQuery(order_id=order_id)
    order_view = QueryHandler.handle_get_order(query)
    if not order_view:
        raise HTTPException(status_code=404, detail="Order not found")
    return order_view

@app.get("/debug/events/{order_id}")
def debug_get_events(order_id: UUID):
    """调试端点：获取订单的原始事件流"""
    events = event_store.load(order_id)
    return {"order_id": order_id, "events": [e.to_dict() for e in events]}

# 库存服务 FastAPI 应用（事件消费者）
from fastapi import FastAPI
import logging
from uuid import UUID

from order_service.infrastructure.message_bus import message_bus
from order_service.events.integration_events import OrderCreatedIntegrationEvent
from shared.models import OrderItem

logger = logging.getLogger(__name__)

app = FastAPI(title="Inventory Service")

# 模拟库存数据库
_inventory_db = {
    "product-001": {"product_name": "Laptop", "stock": 10},
    "product-002": {"product_name": "Mouse", "stock": 50},
}

# --- 集成事件处理器 ---
def on_order_created(event: OrderCreatedIntegrationEvent):
    """消费订单创建事件，扣减库存（最终一致性）"""
    logger.info(f"库存服务收到订单创建事件: 订单 {event.order_id}")
    for item in event.items:
        product_id = item.product_id
        quantity = item.quantity
        if product_id in _inventory_db:
            _inventory_db[product_id]["stock"] -= quantity
            logger.info(f"  扣减库存: 产品 {product_id} 减少 {quantity} 件，剩余 {_inventory_db[product_id]['stock']}")
        else:
            logger.warning(f"  产品 {product_id} 不在库存数据库中")

# 订阅集成事件
message_bus.subscribe('OrderCreatedIntegrationEvent', on_order_created)

# --- HTTP API 端点（供查询）---
@app.get("/inventory/{product_id}")
def get_inventory(product_id: str):
    stock_info = _inventory_db.get(product_id)
    if stock_info is None:
        return {"error": "Product not found"}
    return stock_info

@app.get("/inventory")
def list_inventory():
    return _inventory_db

# 主运行脚本 - 启动两个服务
import subprocess
import sys
import time
import threading
import uvicorn

def run_order_service():
    """运行订单服务在端口 8000"""
    uvicorn.run("order_service.app:app", host="0.0.0.0", port=8000, reload=False, log_level="info")

def run_inventory_service():
    """运行库存服务在端口 8001"""
    uvicorn.run("inventory_service.app:app", host="0.0.0.0", port=8001, reload=False, log_level="info")

if __name__ == "__main__":
    print("启动分布式事件溯源示例系统...")
    print("订单服务: http://localhost:8000")
    print("库存服务: http://localhost:8001")
    print("按 Ctrl+C 停止")

    # 使用多线程同时运行两个服务（适用于演示）
    order_thread = threading.Thread(target=run_order_service, daemon=True)
    inventory_thread = threading.Thread(target=run_inventory_service, daemon=True)

    order_thread.start()
    inventory_thread.start()

    try:
        # 主线程保持运行，直到被中断
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("\n正在关闭服务...")
        sys.exit(0)

fastapi==0.104.1
uvicorn[standard]==0.24.0
pydantic==2.5.0

pip install -r requirements.txt

python run.py

curl -X POST "http://localhost:8000/orders" \
      -H "Content-Type: application/json" \
      -d '{"customer_id": "cust-123"}'

curl "http://localhost:8000/orders/{order_id}"

curl "http://localhost:8000/debug/events/{order_id}"

curl -X POST "http://localhost:8000/orders/{order_id}/items" \
      -H "Content-Type: application/json" \
      -d '{
        "product_id": "product-001",
        "product_name": "Laptop",
        "unit_price": 1200.50,
        "quantity": 2
      }'

curl "http://localhost:8000/orders/{order_id}"

curl "http://localhost:8000/debug/events/{order_id}"

curl "http://localhost:8001/inventory/product-001"

# 1. 创建新订单，获取新的 order_id
    # 2. 立即向该订单添加商品 product-001

curl "http://localhost:8001/inventory/product-001"

# test_aggregate.py
import sys
sys.path.insert(0, '.')

from uuid import UUID
from order_service.aggregates.order import Order
from order_service.events.domain_events import OrderCreatedEvent, OrderItemAddedEvent

def test_order_creation():
    """测试聚合创建与事件生成"""
    order, events = Order.create("test-customer")
    assert isinstance(order.id, UUID)
    assert order.customer_id == "test-customer"
    assert order.status == "PENDING"
    assert len(events) == 1
    assert isinstance(events[0], OrderCreatedEvent)
    print("✓ test_order_creation passed")

def test_order_add_item():
    """测试添加商品与状态重建"""
    order, _ = Order.create("test-customer")
    order_id = order.id
    # 模拟从事件流重建
    history = [
        OrderCreatedEvent(aggregate_id=order_id, customer_id="test-customer"),
        OrderItemAddedEvent(aggregate_id=order_id, product_id="p1", product_name="Test", unit_price=10, quantity=2)
    ]
    rebuilt_order = Order.load_from_history(order_id, history)
    assert rebuilt_order.total_amount == 20.0
    assert len(rebuilt_order.items) == 1
    print("✓ test_order_add_item passed")

if __name__ == "__main__":
    test_order_creation()
    test_order_add_item()
    print("所有聚合测试通过！")

python test_aggregate.py