命令查詢職責分離 – CQRS

• 2022 年 11 月 30 日
• 筆記
• .NET, MASA Framework

概念

CQRS是一種與領域驅動設計和事件溯源相關的架構模式, 它的全稱是Command Query Responsibility Segregation, 又叫命令查詢職責分離, Greg Young在2010年創造了這個術語, 它是基於Bertrand Meyer 的 CQS (Command-Query Separation 命令查詢分離原則) 設計模式。

CQRS認為不論業務多複雜在最終實現的時候, 無非是讀寫操作, 因此建議將應用程式分為兩個方面, 即Command（命令）和Query（查詢）

• 命令端:

  • 關注各種業務如何處理, 更新狀態進行持久化
  • 不返回任何結果 (void)

• 查詢端:

  • 查詢, 並從不修改資料庫

CQRS的三種實現

單一資料庫的CQRS

命令與讀取操作的是同一個資料庫, 命令端通過ORM框架將實體保存到資料庫中, 查詢端通過數據訪問層獲取數據 (數據訪問層通過ORM框架或者存儲過程獲取數據)

雙資料庫的CQRS

命令與讀取操作的是不同的資料庫, 命令端通過ORM框架將實體保存到 寫庫 (Write Db), 並將本地改動推送到 讀庫 (Read Db), 查詢端通過數據訪問層訪問 讀庫 (Read Db), 使用這種模式可以帶來以下好處:

• 查詢更簡單
  • 讀操作不需要任何的完整性校驗, 也不需要外鍵約束, 可以減少鎖爭用, 我們可以針對查詢端單獨優化, 還可以使用剛好包含每個模板需要的數據的資料庫視圖，使得查詢變得更快更簡單
• 提升查詢端的使用體驗
  • 由於這種架構將讀寫徹底分離,由於一般系統是讀操作遠遠大於寫操作, 這給我們的系統帶來了巨大的性能提升, 極大的提升了客戶的使用體驗
• 關注點分離
  • 讀寫分離的模型可以使得關注點分離, 使得讀模型會變得相對簡單

事件溯源 (Event Sourcing) CQRS

通過事件溯源實現的CQRS中會將應用程式的改變都以事件的方式存儲起來, 使用這種模式可以帶來以下好處:

• 事件存儲中了完整的審計跟蹤, 後續出現問題時方便跟蹤
• 可以在任何的時間點重建實體的狀態, 它將有助於排查問題並修復問題
• 提升查詢端的使用體驗
  • 查詢端與命令端可以是完全不同的數據源, 查詢端可以針對查詢條件做針對應的優化, 或者使用像ES、Redis等用來存儲數據, 提升查詢效率
• 獨立縮放
  • 命令端與查詢端可以被獨立縮放, 減少鎖爭用

當然事情有利自然也有弊, CQRS的使用固然會帶來很多好處, 但同樣它也會給項目帶來複雜度的提升, 並且雙資料庫模式、事件溯源模式 的CQRS, 使用的是最終一致性, 這些都是我們在選擇技術方案時必須要考慮的

設計

上述文章中我們了解到了CQRS其本質上是一種讀寫分離的設計思想, 它並不是強制性的規定必須要怎樣去做, 這點與之前的IEvent (進程內事件、IIntegrationEvent (跨進程事件不同, 它並不是強制性的, 根據CQRS的設計模式我們將事件分成Command、Query

• 由於Query (查詢) 是需要有返回值的, 因此我們在繼承IEvent的同時, 還額外增加了一個Result屬性用以存儲結果, 我們希望將查詢的結果保存到Result中, 但它不是強制性的, 我們並沒有強制性要求必須要將結果保存到Result中。

• 由於Command (命令) 是沒有返回值的, 因此我們並沒有額外新增Result屬性, 我們認為命令會更新數據, 那就需要用到工作單元, 因此Command除了繼承IEvent之外, 還繼承了ITransaction，這方便了我們在Handler中的可以通過@event.UnitOfWork來管理工作單元, 而不需要通過構造函數來獲取

但MasaFramework 並沒有要求必須使用 Event Sourcing 模式 或者 雙資料庫模式 的CQRS, 具體使用哪種實現, 它取決於業務的決策者

下面就就來看看MasaFramework提供的CQRS是如何使用的

入門

• 安裝.NET 6.0

1. 新建ASP.NET Core 空項目Assignment.CqrsDemo，並安裝Masa.Contrib.Dispatcher.Events，Masa.Contrib.Dispatcher.IntegrationEvents，Masa.Contrib.Dispatcher.IntegrationEvents.Dapr，Masa.Contrib.ReadWriteSplitting.Cqrs，Masa.Contrib.Development.DaprStarter.AspNetCore

dotnet new web -o Assignment.CqrsDemo
cd Assignment.CqrsDemo

dotnet add package Masa.Contrib.Dispatcher.Events --version 0.7.0-preview.9 //使用進程內事件匯流排
dotnet add package Masa.Contrib.Dispatcher.IntegrationEvents --version 0.7.0-preview.9 //使用跨進程事件匯流排
dotnet add package Masa.Contrib.Dispatcher.IntegrationEvents.Dapr --version 0.7.0-preview.9 //使用Dapr提供pubsub能力
dotnet add package Masa.Contrib.ReadWriteSplitting.Cqrs --version 0.7.0-preview.9 //使用CQRS

dotnet add package Masa.Contrib.Development.DaprStarter.AspNetCore --version 0.7.0-preview.9  //開發環境下協助 Dapr Sidecar, 用於通過Dapr發布集成事件

2. 註冊跨進程事件匯流排、進程內事件匯流排, 修改類Program.cs

示例中未真實使用DB, 不再使用發件箱模式, 只需要使用集成事件提供的PubSub能力即可

builder.Services.AddIntegrationEventBus(dispatcherOptions =>
{
    dispatcherOptions.UseDapr();//使用 Dapr 提供的PubSub能力
    dispatcherOptions.UseEventBus();//使用進程內事件匯流排
});

3. 註冊Dapr Starter 協助管理Dapr Sidecar (開發環境使用)

if (builder.Environment.IsDevelopment())
    builder.Services.AddDaprStarter();

4. 新增加添加商品方法, 修改類Program.cs

app.MapPost("/goods/add", async (AddGoodsCommand command, IEventBus eventBus) =>
{
    await eventBus.PublishAsync(command);
});

/// <summary>
/// 添加商品參數, 用於接受商品參數
/// </summary>
public record AddGoodsCommand : Command
{
    public string Name { get; set; }

    public string Cover { get; set; }

    public decimal Price { get; set; }

    public int Count { get; set; }
}

5. 新增加查詢商品的方法, 修改類Program.cs

app.MapGet("/goods/{id}", async (Guid id, IEventBus eventBus) =>
{
    var query = new GoodsItemQuery(id);
    await eventBus.PublishAsync(query);
    return query.Result;
});

/// <summary>
/// 用於接收查詢商品資訊參數
/// </summary>
public record GoodsItemQuery : Query<GoodsItemDto>
{
    public Guid Id { get; set; } = default!;

    public override GoodsItemDto Result { get; set; }

    public GoodsItemQuery(Guid id)
    {
        Id = id;
    }
}

/// <summary>
/// 用於返回商品資訊
/// </summary>
public class GoodsItemDto
{
    public Guid Id { get; set; }

    public string Name { get; set; }

    public string Cover { get; set; }

    public decimal Price { get; set; }

    public int Count { get; set; }

    public DateTime DateTime { get; set; }
}

6. 新增Command處理程式, 添加類CommandHandler.cs

public class CommandHandler
{
    /// <summary>
    /// 將商品添加到Db，並發送跨進程事件
    /// </summary>
    /// <param name="command"></param>
    /// <param name="integrationEventBus"></param>
    [EventHandler]
    public async Task AddGoods(AddGoodsCommand command, IIntegrationEventBus integrationEventBus)
    {
        //todo: 模擬添加商品到db並發送添加商品集成事件

        var goodsId = Guid.NewGuid(); //模擬添加到db後並獲取商品id
        await integrationEventBus.PublishAsync(new AddGoodsIntegrationEvent(goodsId, command.Name, command.Cover, command.Price,
            command.Count));
    }
}

/// <summary>
/// 跨進程事件, 發送添加商品事件
/// </summary>
/// <param name="Id"></param>
/// <param name="Name"></param>
/// <param name="Cover"></param>
/// <param name="Price"></param>
/// <param name="Count"></param>
public record AddGoodsIntegrationEvent(Guid Id, string Name, string Cover, decimal Price, int Count) : IntegrationEvent
{
    public Guid Id { get; set; } = Id;

    public string Name { get; set; } = Name;

    public string Cover { get; set; } = Cover;

    public decimal Price { get; set; } = Price;

    public int Count { get; set; } = Count;

    public override string Topic { get; set; } = nameof(AddGoodsIntegrationEvent);
}

7. 新增Query處理程式, 添加類QueryHandler.cs

public class QueryHandler
{
    /// <summary>
    /// 從快取查詢商品資訊
    /// </summary>
    /// <param name="query"></param>
    /// <returns></returns>
    [EventHandler]
    public Task GetGoods(GoodsItemQuery query)
    {
        //todo: 模擬從cache獲取商品
        var goods = new GoodsItemDto();

        query.Result = goods;
        return Task.CompletedTask;
    }
}

8. 新增添加商品的跨進程事件的處理服務, 修改Program.cs

app.MapPost(
    "/integration/goods/add",
    [Topic("pubsub", nameof(AddGoodsIntegrationEvent))]
    (AddGoodsIntegrationEvent @event, ILogger<Program> logger) =>
    {
        //todo: 模擬添加商品到快取
        logger.LogInformation("添加商品到快取, {Event}", @event);
    });

// 使用 dapr 來訂閱跨進程事件
app.UseRouting();
app.UseCloudEvents();
app.UseEndpoints(endpoint =>
{
    endpoint.MapSubscribeHandler();
});

流水賬式的服務會使得Program.cs變得十分臃腫, 可以通過Masa Framework提供的MinimalAPIs來簡化Program.cs 點擊查看詳情。

總結

我們上面的例子是通過事件匯流排來完成解耦以及數據模型的同步, 使用的雙資料庫模式, 但讀庫使用的是 快取資料庫, 在Command端做商品的添加操作, 在Query端只做查詢, 且兩端分別使用各自的數據源, 兩者業務互不影響, 並且由於快取資料庫性能更強, 它將最大限度的提升性能, 使得我們有更好的使用體驗。

在Masa Framework中僅僅是通過ICommand、IQuery將讀寫分開, 但這並沒有硬性要求, 事實上你使用IEvent也是可以的, CQRS只是一種設計模式, 這點我們要清楚, 它只是告訴我們要按照一個什麼樣的標準去做, 但具體怎麼來做, 取決於業務的決策者, 除此之外, 後續Masa Framework還會增加對Event Sourcing（事件溯源）的支援, 通過事件重放, 允許我們隨時重建到對象的任何狀態

本章源碼

Assignment15

//github.com/zhenlei520/MasaFramework.Practice

CQRS架構項目：//github.com/masalabs/MASA.EShop/tree/main/src/Services/Masa.EShop.Services.Catalog

參考

• CQRS 模式
• 在微服務中應用簡化後的 CQRS 和 DDD 模式
• .NET現代化應用開發 – CQRS&類目管理程式碼剖析

開源地址

MASA.Framework：//github.com/masastack/MASA.Framework

MASA.EShop：//github.com/masalabs/MASA.EShop

MASA.Blazor：//github.com/BlazorComponent/MASA.Blazor

如果你對我們的 MASA Framework 感興趣, 無論是程式碼貢獻、使用、提 Issue, 歡迎聯繫我們