CQRS读模型同步：微服务间查询写模型的最佳实践探讨

我采用「每个微服务读写分离数据库+HTTP CQRS API」模式，具体场景如下：

资产微服务 #

写模型 #

• AssetWriteDb（MSSQL）
• Asset类/表

public class Asset
{
 public Guid AssetId {get;set;}
 public Guid ContractId {get;set;} //引用其他微服务的合同
 ...
}

读模型 #

• AssetReadDb（MongoDB）
• AssetAggregate类/集合

public class AssetAggregate
{
 public Guid AssetId {get;set;}
 public Guid ContractId {get;get;}
 public string ContractNumber {get;set;} //来自合同微服务
 ...
}

合同微服务 #

写模型 #

• ContractWriteDb（MSSQL）
• Contract类/表

public class Contract
{
 public Guid ContractId {get;set;}
 public string ContractNumber {get;set;}
 ...
}

读模型 #

• ContractReadDb（MongoDB）
• ContractAggregate类/集合

public class ContractAggregate
{
 public Guid ContractId {get;set;}
 public string ContractNumber {get;set;}
 public int AssetCount {get;set;} //来自资产微服务
 ...
}

合同聚合同步事件处理器示例 #

public class ContractAggregateHandler :
 IHandleMessage<ContractChangedEvent> //由ContractWriteDb MSSQL仓储发布
 IHandleMessage<AssetChangedEvent> //由AssetWriteDb MSSQL仓储发布 
{
 
 public async Task Handle(ContractChangedEvent message)
 {
   await _bus.Send(new RefreshContractAggregateCommand(message.ContractId));
 }

 public async Task Handle(AssetChangedEvent message)
 {
   //由于事件仅包含AssetId，需从资产微服务获取数据，有两种方式获取ContractId：

   //调用AssetApi查询AssetAggregate集合（MongoDB）
   //var contractId = await _mediator.Send(new GetAssetContractIdQuery(message.AssetId);
   
   //调用AssetApi查询Asset表（SQL Server）
   //var contractId = await _mediator.Send(new GetAssetContractIdFromWriteDbQuery(message.AssetId);


   await _bus.Send(new RefreshContractAggregateCommand(contractId));
 }
}

问题与方案分析 #

遵循「查询需从读模型获取数据，命令需读写写模型」的规则，需确定读模型同步的最佳实践。

现有方案问题梳理 #

1. 第一种方案（查询资产读模型）

   • 核心问题：AssetChanged事件来自资产写模型，此时读模型可能未完成同步，存在数据不一致风险；若改为监听AssetAggregateRefreshedEvent触发同步，会引发聚合生成的无限循环。

2. 第二种方案（查询资产写模型）

   • 优势：数据一致性更有保障；
   • 问题：违反「查询必须从读模型获取数据」的规则，需通过特殊命名（如后缀FromWriteDbQuery）区分这类内部查询。

3. 第三种方案（直接从合同微服务查询资产写数据库）

   • 已排除，不考虑。

补充说明 #

• 公共API网关保护内部微服务，网关对外暴露的查询均正确访问MongoDB，本次问题仅聚焦于内部聚合处理及微服务间如何查询写模型。
• 同步业务逻辑（RefreshContractAggregateHandler）未编写，其逻辑为查询ContractWriteDb并投影为ContractAggregate，统计AssetCount时也存在相同问题，即需从合同微服务查询资产写数据库。

内容的提问来源于stack exchange，提问作者pfab.io