findscripter 何时使用
适用：
- 需要把「写」（命令/状态变更）与「读」（查询/报表）职责分离，各自独立演进与扩展。
- 读负载远大于写负载，需要独立扩展读侧、对读模型做反范式化以优化查询性能。
- 构建事件溯源（Event Sourcing）系统，以事件流作为唯一事实来源。
- 复杂查询、高性能报表场景，读写数据模型天然不同。
不该用（负边界）：
- 领域简单、CRUD 已足够，引入 CQRS 只会增加复杂度。
- 无法运维独立的读/写存储与同步链路。
- 处处都要求强即时一致性（读写之间不能容忍任何传播延迟）。
步骤
- 梳理读/写工作负载与一致性需求：明确哪些操作改状态、哪些只读，能容忍多大读延迟（定义一致性 SLA）。
- 划清命令模型与查询模型边界：命令表达「修改意图」，查询表达「取数请求」，二者用各自的总线与处理器，schema 不耦合。
- 实现读模型投影与同步：由投影器（Projector）消费事件、更新反范式化读模型，并维护 checkpoint 支持续传与重建。
- 处理最终一致性：对需要「读己之写」的场景，按期望版本号轮询读模型，超时则返回带 stale 警告的旧数据。
- 验证性能、恢复与故障模式：投影失败重试/跳过、投影重建、事件版本化演进。复杂模式见下方模板。
指令
- 命令侧：
Command->CommandBus.dispatch()->CommandHandler.handle()，在处理器内校验后再产生事件并event_store.append_events()。 - 查询侧：
Query->QueryBus.dispatch()->QueryHandler.handle()，仅从读模型取数，返回反范式化视图（如OrderView、PaginatedResult）。 - API 路由约定：写用 POST/PUT/DELETE 走命令总线，读用 GET 走查询总线，二者物理隔离。
- 同步：
ReadModelSynchronizer.run()持续投影；rebuild_projection()清空数据、重置 checkpoint 后从头重放重建。
示例
命令基础设施与处理器（Python）：
@dataclass
class CreateOrder(Command):
customer_id: str
items: list
shipping_address: dict
class CommandBus:
def __init__(self):
self._handlers: Dict[Type[Command], CommandHandler] = {}
def register(self, command_type, handler):
self._handlers[command_type] = handler
async def dispatch(self, command: Command) -> Any:
handler = self._handlers.get(type(command))
if not handler:
raise ValueError(f"No handler for {type(command).__name__}")
return await handler.handle(command)
class CreateOrderHandler(CommandHandler[CreateOrder]):
async def handle(self, command: CreateOrder) -> str:
if not command.items: # 在命令处理器内校验
raise ValueError("Order must have at least one item")
order = Order.create(customer_id=command.customer_id,
items=command.items,
shipping_address=command.shipping_address)
await self.event_store.append_events( # 持久化事件
stream_id=f"Order-{order.id}", stream_type="Order",
events=order.uncommitted_events)
return order.id
FastAPI 中写命令 / 读查询物理分离：
@app.post("/orders") # 命令：改状态
async def create_order(request, command_bus=Depends(get_command_bus)):
order_id = await command_bus.dispatch(CreateOrder(**request.dict()))
return {"order_id": order_id}
@app.get("/orders/{order_id}") # 查询：只读读模型
async def get_order(order_id, query_bus=Depends(get_query_bus)):
result = await query_bus.dispatch(GetOrderById(order_id=order_id))
if not result:
raise HTTPException(status_code=404, detail="Order not found")
return result
读己之写的最终一致性等待（按版本轮询，超时降级）：
async def query_after_command(self, query, expected_version, stream_id, timeout=5.0):
start = time.time()
while time.time() - start < timeout:
if await self._get_projection_version(stream_id) >= expected_version:
return await self.execute_query(query)
await asyncio.sleep(0.1)
return {"data": await self.execute_query(query), "_warning": "Data may be stale"}
注意事项
应做（Do）：
- 命令模型与查询模型分离，各按需设计；读模型反范式化以优化查询。
- 接受最终一致性与传播延迟，并明确定义可接受的滞后 SLA。
- 在命令处理器中完成校验，先校验后改状态。
- 对事件做版本化，支持 schema 演进。
不应做（Don't）：
- 不要在命令中执行查询（命令只负责写）。
- 不要耦合读/写 schema，二者应独立演进。
- 不要过度设计，从简单做起。
- 不要忽视一致性 SLA。
- 投影出错时记录日志并按策略重试或跳过，不要让整条同步链路崩溃。
关键组件职责：命令=改状态意图；命令处理器=校验并执行；事件=状态变更记录；查询=取数请求；查询处理器=从读模型取数；投影器=用事件更新读模型。
互见
- CQRS 模式（Martin Fowler）：https://martinfowler.com/bliki/CQRS.html
- 微软 CQRS 指南：https://learn.microsoft.com/azure/architecture/patterns/cqrs
- 相关：事件溯源、领域驱动设计（DDD）、读模型/物化视图、最终一致性。
采编自 sickn33/antigravity-awesome-skills（MIT 许可）。