CQRS 与 DDD 中如何及时更新 View

2026-10-24支付阿杰

结论

CQRS 不要求所有更新都异步，也不要求写模型和读模型使用不同的数据库。

在这个场景中，删除 Payment Method 属于业务状态变更，应在 command 成功返回前持久化到写模型。异步事件可以更新 read model、发送通知或同步其他系统，但不应承担核心业务写入。否则，command 返回“成功”只代表消息已经发出，并不代表业务操作已经完成。

如果重定向后的页面必须立即读到最新结果，通常有三种处理方式：

同步更新该 read model，直接提供强一致的交互体验。
保留异步投影，通过版本号等待 read model 追上写模型。
接受最终一致性，在界面上显示“处理中”并轮询状态。

无法同时做到完全异步解耦、完全不等待，并保证重定向后的任意查询立即看到最新数据。要实现“read-your-writes”一致性，需要选择同步更新、等待投影或 UI 补偿中的一种。

先修正命令处理流程

原流程中首先需要调整的是：

PaymentMethodRemovedMessage 的异步 handler 才从数据库删除实际记录。

如果这条记录代表 Account 当前有效的 Payment Method，那么删除记录就是业务事务的一部分，而不是 read model 投影。

流程可以改为：

Controller 发送 RemovePaymentMethod command。
Command handler 加载 Account。
Account.RemovePaymentMethod() 执行业务校验并修改聚合状态。
Repository 在同一事务中保存 Account。
同一事务将 PaymentMethodRemoved 写入 outbox。
事务提交后，command 返回成功结果和新的聚合版本号。
后台 publisher 将 outbox 中的事件发送到 service bus。
各个 projection handler 异步更新 read model。

示意代码如下：

public async Task<RemovePaymentMethodResult> Handle(
    RemovePaymentMethod command,
    CancellationToken cancellationToken)
{
    await using var transaction =
        await unitOfWork.BeginTransactionAsync(cancellationToken);

    var account = await accountRepository.GetAsync(
        command.AccountId,
        cancellationToken);

    account.RemovePaymentMethod(command.PaymentMethodId);

    await accountRepository.SaveAsync(account, cancellationToken);

    foreach (var domainEvent in account.DequeueDomainEvents())
    {
        await outbox.AddAsync(domainEvent, cancellationToken);
    }

    await transaction.CommitAsync(cancellationToken);

    return new RemovePaymentMethodResult(
        account.Id,
        account.Version);
}

这里，accountRepository.SaveAsync() 保存的是权威业务状态。outbox 则保证状态提交与事件记录不会出现一方成功、另一方失败的不一致。

异步 handler 只负责更新查询模型：

public sealed class PaymentMethodRemovedProjection
    : IHandle<PaymentMethodRemoved>
{
    public async Task Handle(
        PaymentMethodRemoved message,
        CancellationToken cancellationToken)
    {
        await readModel.DeletePaymentMethodAsync(
            message.AccountId,
            message.PaymentMethodId,
            message.AggregateVersion,
            cancellationToken);
    }
}

如果所谓的“实际记录”只是供页面查询的副本，那么由 handler 删除并没有问题。不过，此时需要明确它是最终一致的 read model，而不是 command 的权威状态。

方案一：同步更新当前页面所需的 read model

修改地址、删除付款方式等操作，用户通常希望提交后立即看到结果。最直接的处理方式是同步维护当前页面使用的 projection。

例如，command transaction 可以同时更新写模型和同一数据库中的 read table：

account.RemovePaymentMethod(command.PaymentMethodId);

await accountRepository.SaveAsync(account, cancellationToken);
await accountDetailsProjection.RemovePaymentMethodAsync(
    command.AccountId,
    command.PaymentMethodId,
    cancellationToken);

await outbox.AddAsync(
    new PaymentMethodRemoved(
        account.Id,
        command.PaymentMethodId,
        account.Version),
    cancellationToken);

await transaction.CommitAsync(cancellationToken);

事务提交后返回 303 See Other：

POST /accounts/42/payment-methods/7/remove
303 See Other
Location: /accounts/42/payment-methods

随后的 GET 就能读到更新后的数据。

这种做法并不违反 CQRS。CQRS 关注的是 command 与 query 各自承担不同职责、使用不同模型，并没有强制要求两者只能通过异步消息衔接。代价是 projection 与写事务会产生一定耦合，因此更适合数量较少、对交互一致性要求较高的核心页面。

方案二：使用版本号等待 read model 追上

如果 read model 必须通过 service bus 异步更新，可以让 command 返回一个单调递增的版本号，例如 Account.Version = 18。

重定向时带上该版本：

303 See Other
Location: /accounts/42/payment-methods?expectedVersion=18

每个 read model 记录自己已经处理到的版本：

AccountId: 42
ProjectionVersion: 18

如果 GET 请求发现 projection 版本仍是 17，可以短暂等待，例如最多等待几百毫秒：

public async Task<IActionResult> Get(
    Guid accountId,
    long? expectedVersion,
    CancellationToken cancellationToken)
{
    if (expectedVersion is not null)
    {
        var caughtUp = await projectionWaiter.WaitUntilAsync(
            accountId,
            expectedVersion.Value,
            timeout: TimeSpan.FromMilliseconds(500),
            cancellationToken);

        if (!caughtUp)
        {
            return View("Updating", new UpdatingViewModel(accountId));
        }
    }

    var model = await accountQueries.GetAsync(
        accountId,
        cancellationToken);

    return View(model);
}

投影通常还需要忽略重复事件和旧版本事件：

UPDATE AccountProjection
SET ProjectionVersion = @Version
WHERE AccountId = @AccountId
  AND ProjectionVersion < @Version;

这种方式保留了异步架构，也提供了“至少能读到自己刚才写入的数据”的语义。它需要额外处理版本管理、超时和错误，复杂度更高，但比让 UI 猜测 read model 是否已经更新更可靠。

等待时间不能没有上限。service bus 可能发生延迟，投影也可能失败，因此 GET 必须设置明确的超时时间，并准备好降级页面。

方案三：明确显示操作状态

如果操作本身就是一个长流程，例如提交订单、跨系统退款或关闭账户，那么 command 被接受并不代表操作已经完成。此时可以返回 operation ID：

202 Accepted
Location: /operations/01J8XYZ

查询结果可以是：

{
  "operationId": "01J8XYZ",
  "status": "Pending"
}

页面显示“正在处理”，再通过轮询、Server-Sent Events 或 WebSocket 获取 Completed、Rejected、Failed 等结果。

对于没有 JavaScript 的客户端，可以返回带自动刷新提示的状态页，也可以提示用户稍后刷新。对真正的异步业务流程来说，这种交互准确反映了系统当前状态。

不过，删除付款方式通常只是一个很短的本地事务，没有必要将其建模为长流程，除非该操作确实需要等待外部支付系统确认。

AJAX 与非 JavaScript 客户端如何统一

直接删除页面上的 div 属于 optimistic UI，可以改善 AJAX 的交互体验，但不能作为一致性的唯一保障。后端仍需返回明确的 command 结果，并提供可查询的状态。

两类客户端可以共用同一套语义：

AJAX 客户端在收到 command 成功结果后立即更新界面，同时在后台确认 projection 版本。
非 JavaScript 客户端使用 PRG 重定向，并携带 expectedVersion。
projection 已经追上时，直接显示正常页面。
projection 尚未追上时，先短暂等待；超时后显示“更新处理中”。
command 被拒绝时，两类客户端都显示具体的业务错误。

这样一来，JavaScript 只负责改善交互，不会改变业务正确性。

还需要注意的几个问题

区分 domain event 与 integration event

聚合可以在事务中产生 domain event。事务提交后，再通过 outbox 将需要跨进程传播的 integration event 发布到 service bus。

不要在数据库事务提交前直接向总线发送消息，否则可能出现消息已经发布、业务事务却回滚的情况。

Command 成功不能只表示“消息发送成功”

对于立即完成的业务操作，成功应表示：

业务规则已经通过；
权威状态已经持久化；
用于后续可靠发布的事件已经写入 outbox。

read model 是否完成投影可以单独表示。

Handler 必须支持幂等

service bus 通常提供至少一次投递，因此消息可能重复。删除操作、版本更新和 projection handler 都需要能够安全地重复执行。

处理并发修改

版本号既可以用于等待 projection，也可以用于 optimistic concurrency。如果两个请求同时修改同一个 Account，其中一个请求保存时应检测到版本冲突，不能静默覆盖另一个操作。

不要忽略缓存

即使 projection 已经更新，HTTP 缓存、CDN 或应用缓存仍可能返回旧数据。系统需要正确设置缓存失效策略，必要时将版本号加入 cache key。