AI原生开发:ClaudeCode与国产大模型协同实战 第9章:屠龙实战——基于事件驱动的分布式积分清算引擎

# clearing_spec.yamlentities:  ClearingTask:    fields:      task_id: { type: string, required: true, description: "幂等键，通常是订单号" }      order_amount: { type: float, required: true }      status: { type: enum, values: [PENDING, PROCESSING, SUCCESS, FAILED] }      version: { type: integer, required: true, description: "乐观锁" }contracts:  SubmitClearingAPI:    endpoint: /clearing/submit    method: POST    request:      task_id: { type: string }      order_amount: { type: float }    response:      success: { code: 202, message: "Task accepted and processing" } # 异步接受      errors:        - code: 40901, message: "Duplicate task (Idempotent)"transitions:  process_task:    from_status: PENDING    to_status: SUCCESS / FAILED    pre_conditions:      - "task does not exist (idempotent check)"      - "account balance >= order_amount"    side_effects:      - "deduct account balance"      - "add loyalty points"      - "publish ClearingSuccessEvent"

# Role你是一个精通领域驱动设计（DDD）和状态模式的结算架构师。# Task请根据我提供的 OpenSpec 规约，实现积分清算引擎的核心领域逻辑。# OpenSpec（粘贴上面的 clearing_spec.yaml）# Directive1. 使用 Python + Pydantic 定义实体。2. 必须使用状态模式实现 ClearingTask 的流转，禁止在业务逻辑中使用 if-else 判断状态。3. 必须实现基于 task_id 的幂等校验。4. 必须实现基于 version 的乐观锁机制（模拟数据库层）。5. 所有异常必须严格对应 OpenSpec 中的定义。请输出核心的领域模型、状态机逻辑和本地内存版的 Repository。

from enum import Enumfrom pydantic import BaseModelfrom typing import Optional, Dictimport threading# ================== 1. 枚举与异常 (对应 OpenSpec) ==================class TaskStatus(str, Enum):    PENDING = "PENDING"    PROCESSING = "PROCESSING"    SUCCESS = "SUCCESS"    FAILED = "FAILED"class ClearingError(Exception):    def __init__(self, code: int, message: str):        self.code = code        self.message = message# ================== 2. 领域实体 (对应 OpenSpec Entities) ==================class ClearingTask(BaseModel):    task_id: str    order_amount: float    status: TaskStatus = TaskStatus.PENDING    version: int = 1class Account(BaseModel):    user_id: str    balance: float    version: int = 1# ================== 3. 内存仓储 (带乐观锁) ==================class ClearingRepository:    def __init__(self):        self._tasks: Dict[str, ClearingTask] = {}        self._accounts: Dict[str, Account] = {}        self._lock = threading.Lock()        # 初始化一个测试账户        self._accounts["U001"] = Account(user_id="U001", balance=10000.0)    def get_task(self, task_id: str) -> Optional[ClearingTask]:        return self._tasks.get(task_id)    def save_task(self, task: ClearingTask):        self._tasks[task.task_id] = task    def deduct_balance(self, user_id: str, amount: float, expected_version: int) -> bool:        """模拟数据库层面的乐观锁扣减"""        with self._lock:            acc = self._accounts.get(user_id)            if not acc: return False            if acc.version != expected_version: return False # 乐观锁冲突            if acc.balance < amount: return False # 余额不足            acc.balance -= amount            acc.version += 1            return True# ================== 4. 状态机流转 (状态模式) ==================class ClearingStateMachine:    def __init__(self, repo: ClearingRepository):        self.repo = repo    def process(self, task: ClearingTask) -> TaskStatus:        if task.status != TaskStatus.PENDING:            raise ClearingError(50000, f"Invalid state transition from {task.status}")        task.status = TaskStatus.PROCESSING        # 执行核心扣减逻辑        success = self.repo.deduct_balance("U001", task.order_amount, 1)        if success:            task.status = TaskStatus.SUCCESS        else:            task.status = TaskStatus.FAILED        self.repo.save_task(task)        return task.status

基于我刚生成的 core_domain.py，请为我构建一个 FastAPI 的事件驱动版本：1. 暴露 POST /clearing/submit 接口，接收请求后，将 ClearingTask 投递到 asyncio.Queue 中，立刻返回 202。2. 创建一个后台 Worker，持续监听 Queue 并调用 ClearingStateMachine 进行处理。3. 必须保证在并发提交时，相同的 task_id 不会被重复投递（接口层幂等拦截，返回 40901）。4. 提供完整的、可直接运行的代码。

import asyncioimport uvicornfrom fastapi import FastAPI, HTTPExceptionfrom core_domain import ClearingTask, ClearingRepository, ClearingStateMachine, TaskStatus, ClearingErrorapp = FastAPI()repo = ClearingRepository()state_machine = ClearingStateMachine(repo)# 异步事件总线 (内存版 Kafka)clearing_queue = asyncio.Queue()# 幂等记录 (防重复投递)submitted_tasks = set()# ================== 后台 Worker ==================async def clearing_worker():    """后台消费者，不断从队列拿任务执行"""    while True:        task = await clearing_queue.get()        try:            print(f"Worker: 开始处理任务 {task.task_id}, 金额 {task.order_amount}")            result_status = state_machine.process(task)            print(f"Worker: 任务 {task.task_id} 处理完毕, 状态 {result_status}")        except Exception as e:            print(f"Worker: 任务处理异常 {e}")        finally:            clearing_queue.task_done()@app.on_event("startup")async def startup_event():    # 启动后台Worker协程    asyncio.create_task(clearing_worker())# ================== API 接口 ==================@app.post("/clearing/submit")async def submit_clearing(task_id: str, order_amount: float):    # 1. 接口层幂等拦截 (对应 OpenSpec 40901)    if task_id in submitted_tasks:        raise HTTPException(status_code=409, detail="Duplicate task (Idempotent)")    submitted_tasks.add(task_id)    # 2. 构建任务并投递    task = ClearingTask(task_id=task_id, order_amount=order_amount)    repo.save_task(task) # 先持久化初始状态    await clearing_queue.put(task)    # 3. 异步返回 202    return {"code": 202, "message": "Task accepted and processing", "task_id": task_id}@app.get("/clearing/status/{task_id}")async def get_status(task_id: str):    task = repo.get_task(task_id)    if not task:        raise HTTPException(status_code=404, detail="Task not found")    return {"task_id": task.task_id, "status": task.status, "version": task.version}if __name__ == "__main__":    uvicorn.run(app, host="0.0.0.0", port=8000)

curl "http://127.0.0.1:8000/clearing/submit?task_id=T001&order_amount=100"

现在的系统是单机内存版，无法扛住多实例部署。请帮我进行分布式改造：1. 引入 redis-py，实现基于 Redis 的分布式锁（使用 SETNX + 过期时间）。2. 将接口层的幂等控制从内存 set 改为 Redis 的 SETNX。3. 扣减余额的逻辑中，必须先获取分布式锁，再执行业务，最后释放锁。4. 考虑锁超时和异常中断的兜底逻辑。请直接改造 core_domain.py 中的相关逻辑（假设 Redis 连接已就绪）。

import redisimport uuidimport time# 假设已初始化 Redis 连接r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)class DistributedClearingService:    def __init__(self, repo):        self.repo = repo    def deduct_with_lock(self, user_id: str, amount: float, task_id: str) -> bool:        lock_key = f"lock:account:{user_id}"        lock_value = str(uuid.uuid4()) # 确保只有持有者能释放        acquire_timeout = 5  # 获取锁超时时间        lock_expire = 10     # 锁自动过期时间（防死锁）        # 1. 尝试获取分布式锁 (SETNX)        end_time = time.time() + acquire_timeout        while time.time() < end_time:            if r.set(lock_key, lock_value, nx=True, ex=lock_expire):                break            time.sleep(0.01)        else:            raise ClearingError(50300, "Failed to acquire distributed lock")        try:            # 2. 获取锁成功，执行扣减逻辑 (调用前文的 repo)            # 注意：真实场景中 repo.deduct_balance 应对接真实的数据库            success = self.repo.deduct_balance(user_id, amount, expected_version=1)            return success        finally:            # 3. 安全释放锁 (Lua 脚本，保证原子性，防止误删别人的锁)            lua_script = """            if redis.call("get", KEYS[1]) == ARGV[1] then                return redis.call("del", KEYS[1])            else                return 0            end            """            r.eval(lua_script, 1, lock_key, lock_value)