使用MediatR实现CQRS

2023-08-13 21:11 由赵榕发表于 #软件设计

CQRS和中介者模式

MediatR库主要是为了帮助开发者快速实现两种软件架构模式：CQRS和Mediator。这两种架构模式看上去似乎差不多，但还是有很多区别的。

CQRS

CQRS是Command Query Responsibility Segregation的缩写，一般称作命令查询职责分离。从字面意思理解，就是将命令（写入）和查询（读取）的责任划分到不同的模型中。

对比一下常用的 CRUD 模式（创建-读取-更新-删除），通常我们会让用户界面与负责所有四种操作的数据存储交互。而 CQRS 则将这些操作分成两种模式，一种用于查询（又称 "R"），另一种用于命令（又称 "CUD"）。

如图所示，应用程序只是将查询和命令模型分开。CQRS并没有对分离的方式做出具体的规定。可以是应用程序里面的一个类或者第三方类库，也可以是通过不同的服务器进行物理上的隔离。具体如何实现取决于应用程序的实际情况。总而言之，CQRS的核心就是将读和写分开。

看到这里是不是有种似曾相识的感觉？没错，CQRS的设计理念和数据库的读写分离一毛一样。

CQRS看上去似乎很棒，但它也是一把双刃剑，和软件开发实践中的其他东西一样，需要进行一些平衡和取舍，包括：

管理单独的系统（如果应用程序层被拆分）
数据过时（如果数据库层被拆分）
管理多个组件的复杂性

是否使用CQRS模式最终取决于我们的特定用例。良好的开发实践鼓励我们“保持简单”（KISS），因此仅在需要时使用这些模式，否则就是过度设计了。

Mediator 模式

Mediator模式只是定义了一个对象，它封装了对象之间的交互方式。两个或多个对象之间不再直接相互依赖，而是通过一个 "中介 "进行交互，"中介 "负责将这些交互发送给另一方。

如上图所示，SomeService 向Mediator发送消息，然后Mediator调用多个服务来处理该消息。任何Handler组件之间没有直接依赖关系。

中介模式之所以有用，与控制反转（Inversion of Control）等模式一样。它可以实现 "松耦合"，因为依赖关系图最小化，因此代码更简单、更容易测试。换句话说，一个组件考虑的因素越少，它就越容易开发和演进。

我们在上图中看到了服务之间没有直接依赖关系，消息的生产者不知道是那些Handler在处理它。这与消息代理在“发布/订阅”模式中的工作方式非常相似。如果我们想添加另一个处理程序，直接条件就可以了，不必修改生产者。

如何使用MediatR？

我们可以将 MediatR 视为“进程内”中介器实现方案，这有助于我们构建 CQRS 系统。用户界面和数据存储之间的所有通信都通过 MediatR 进行。

这里我们需要注意”进程内“这三个字，这是一个非常重要的限制条件，意味着您无法使用MediatR实现跨进程消息通信。如果我们想跨两个系统分离命令和查询，更好的方法是使用消息代理，例如 Kafka 、RabbitMQ或 Azure 服务总线等等。推荐学习一下MassTransit这个库。

在ASP.NET Core API项目中配置MediatR

项目设置

首先，让我们打开Visual Studio并创建一个新的 ASP.NET Core Web API应用程序。我们将它命名为CqrsMediatrExample。

安装依赖包

PM> install-package MediatR

如果是v12之前的版本，则需要再安装MediatR.Extensions.Microsoft.DependencyInjection。

注册依赖

打开Program.cs

builder.Services.AddMediatR(cfg => cfg.RegisterServicesFromAssembly(typeof(Program).Assembly));

我们必须为构造函数提供默认配置。

现在，MediatR 已配置完毕，随时可用。

在我们进入控制器创建之前，我们将修改文件：launchSettings.json

{
  "profiles": {
    "https": {
      "commandName": "Project",
      "dotnetRunMessages": true,
      "launchBrowser": false,
      "applicationUrl": "https://localhost:5001;http://localhost:5000",
      "environmentVariables": {
        "ASPNETCORE_ENVIRONMENT": "Development"
      }
    }
  }
}

添加控制器

现在我们已经安装了所有内容，让我们设置一个新的控制器，它将向 MediatR 发送消息。

在“Controllers”文件夹中，让我们添加一个名称为 ProductsController.cs的控制器

然后我们得到以下类：

[Route("api/[controller]")]
[ApiController]
public class ProductsController : ControllerBase
{
}

接下来，让我们通过构造函数注入一个IMediatR实例：

[Route("api/[controller]")]
[ApiController]
public class ProductsController : ControllerBase
{
    private readonly IMediator _mediator;
    public ProductsController(IMediator mediator)
    {
        _mediator = mediator;
    }
}

IMediatR 接口允许我们向 MediatR 发送消息，然后由 MediatR 向相关处理程序派发消息。因为我们已经安装了依赖注入软件包，所以实例会自动解析。

从 MediatR 9.0 版开始，IMediator 接口被拆分为ISender 和 IPublisher两个接口。因此，尽管我们仍然可以使用 IMediator 接口向处理程序发送请求，但是更严谨一些的做法是分别使用ISender和IPublisher分别发送不同类型的消息。

public interface ISender
{
    Task<TResponse> Send<TResponse>(IRequest<TResponse> request, CancellationToken cancellationToken = default);
    Task<object?> Send(object request, CancellationToken cancellationToken = default);
}
public interface IPublisher
{
    Task Publish(object notification, CancellationToken cancellationToken = default);
    Task Publish<TNotification>(TNotification notification, CancellationToken cancellationToken = default)
        where TNotification : INotification;
}
public interface IMediator : ISender, IPublisher
{
}

数据存储

通常，我们希望与真实的数据库进行交互。但在本文中，让我们创建一个包含此责任的Fake class，并简单地与一些 Product 实体进行交互。

但在这样做之前，我们必须创建一个简单的类：Product

public class Product
{
    public int Id { get; set; }
    public string Name { get; set; }
}

接下来我们添加一个新的类，命名为FakeDataStore：

public class FakeDataStore
{
    private static List<Product> _products;
    public FakeDataStore()
    {
        _products = new List<Product>
        {
            new Product { Id = 1, Name = "Test Product 1" },
            new Product { Id = 2, Name = "Test Product 2" },
            new Product { Id = 3, Name = "Test Product 3" }
        };
    }
    public async Task AddProduct(Product product)
    { 
        _products.Add(product);
        await Task.CompletedTask;
    }
    public async Task<IEnumerable<Product>> GetAllProducts() => await Task.FromResult(_products);
}

然后，我们需要在Program.cs将FakeDataStore配置到依赖注入：

builder.Services.AddSingleton<FakeDataStore>();

分离命令和查询

本文毕竟是关于 CQRS 的，因此让我们为此目的创建三个新文件夹：Commands、Queries和Handlers。我们将通过这三个文件夹将模型进行物理上的分隔。

使用MediatR发送请求

MediatR请求是非常简单的请求-响应样式消息，其中单个请求由单个处理程序同步处理（这里的同步并不是编程意义上的同步，而是从业务或者流程的角度触发，即发送请求后持续等待流程处理完成并且返回结果，需要和C#的async/await区别开）。这里我们做一个简单的例子来示范查询或者更新数据库。

MediatR 中有两种类型的请求。一个有返回值，另一个没有返回值。通常，这对应于读取/查询（返回值）和写入/命令（通常不返回值）。

获取产品（Query）

由于这是一个查询，让我们添加一个调用到 “Queries” 文件夹的类，并实现它：GetProductsQuery

public record GetProductsQuery() : IRequest<IEnumerable<Product>>;

这里我们创建一个名为GetProductsQuery的record对象并且继承IRequest<IEnumerable<Product>>接口，表示此查询将返回一个Product集合。

然后，在 Handlers 文件夹中，我们将创建一个新的Handler类来处理我们的查询：

public class GetProductsHandler : IRequestHandler<GetProductsQuery, IEnumerable<Product>>
{
    private readonly FakeDataStore _fakeDataStore;
    public GetProductsHandler(FakeDataStore fakeDataStore) => _fakeDataStore = fakeDataStore;
    public async Task<IEnumerable<Product>> Handle(GetProductsQuery request,
        CancellationToken cancellationToken) => await _fakeDataStore.GetAllProducts();
}

稍微分解以下，GetProductsHandler要继承IRequestHandler<GetProductsQuery, IEnumerable<Product>>，表示GetProductsHandler可以处理GetProductsQuery查询请求，并且返回一个产品列表，具体的查询逻辑在Handle方法中实现。

调用请求

要调用查询请求，只需要在ProductsController中添加一个GetProducts的Action。

[HttpGet]
public async Task<ActionResult> GetProducts()
{
    var products = await _mediator.Send(new GetProductsQuery());
    return Ok(products);
}

Too simple对不对？

来测试一下吧。

首先在IDE或者控制台中运行我们的项目。然后打开Postman并创建一个请求：

MediatR发送命令

我们在“Commands”文件夹中添加一个名为AddProductCommand的record，并且继承IRequest接口。

public record AddProductCommand(Product Product) : IRequest;

因为我们不需要返回值，所以只需要继承IRequest，不需要添加泛型参数，AddProductCommand将会自动拥有一个名为Product的属性。

注意：因为我们仅仅是为了简单且快速地示范MediatR的使用，所以此处直接使用的领域实体作为参数，在实际使用中，应当使用DTO等对象从公共Api中隐藏领域实体。

接下来，我们要在“Handlers”文件夹中添加AddProductCommand的Handler。

public class AddProductHandler : IRequestHandler<AddProductCommand> 
{ 
    private readonly FakeDataStore _fakeDataStore; 
        
    public AddProductHandler(FakeDataStore fakeDataStore) => _fakeDataStore = fakeDataStore; 
        
    public async Task Handle(AddProductCommand request, CancellationToken cancellationToken) 
    {
        await _fakeDataStore.AddProduct(request.Product); 
            
        return; 
    } 
}

调用请求

同样的，我们在ProductsController中添加一个名为AddProduct的Action来发送Command。

[HttpPost]
public async Task<ActionResult> AddProduct([FromBody]Product product)
{
    await _mediator.Send(new AddProductCommand(product));
    return StatusCode(201);
}

与上一个方法类似，只不过这次我们不需要返回任何值。

运行项目，并向Postman中添加一个新的请求：

执行完成后再次运行之前的查询请求：

新添加的数据已经出现在列表中，证明我们的代码已经按照预期执行了。

使用返回值的命令

我们的Post操作目前返回的是201状态码，并没有包含其他的信息。然而在实际应用中，客户端可能需要更多的信息，例如新添加的产品的Id等。

在此之前我们需要添加一个根据Id获取产品的功能。

在“Queries”文件夹里添加一个名为GetProductByIdQuery的record：

public record GetProductByIdQuery(int Id) : IRequest<Product>;

修改FakeDataStore使其支持根据Id查询产品信息：

public async Task<Product> GetProductById(int id) => 
    await Task.FromResult(_products.Single(p => p.Id == id));

添加一个新的Handler用于处理GetProductByIdQuery：

public class GetProductByIdHandler : IRequestHandler<GetProductByIdQuery, Product>
{
    private readonly FakeDataStore _fakeDataStore;
    public GetProductByIdHandler(FakeDataStore fakeDataStore) => _fakeDataStore = fakeDataStore;
    public async Task<Product> Handle(GetProductByIdQuery request, CancellationToken cancellationToken) =>
        await _fakeDataStore.GetProductById(request.Id);
        
}

在Controller中添加新的Get接口：

[HttpGet("{id:int}", Name = "GetProductById")]
public async Task<ActionResult> GetProductById(int id)
{
    var product = await _mediator.Send(new GetProductByIdQuery(id));
    return Ok(product);
}

好了，我们在Postman中添加一个新的请求，并测试一下：

修改命令和Handler

如果Request需要返回操作结果，只需要将Command的接口增加一个泛型参数，参数的类型为需要返回的值的类型。

public record AddProductCommand(Product Product) : IRequest<Product>;

Handler也需要做一些调整：

public class AddProductHandler : IRequestHandler<AddProductCommand, Product>
{
    private readonly FakeDataStore _fakeDataStore;
    public AddProductHandler(FakeDataStore fakeDataStore) => _fakeDataStore = fakeDataStore;
    public async Task<Product> Handle(AddProductCommand request, CancellationToken cancellationToken)
    {
        await _fakeDataStore.AddProduct(request.Product);
            
        return request.Product;
    }
}

这是一个简化到极致的例子，目的仅仅是为了演示如何使用。

最后需要修改的是Controller的Action方法：

[HttpPost]
public async Task<ActionResult> AddProduct([FromBody]Product product)
{
    var productToReturn = await _mediator.Send(new AddProductCommand(product));
    return CreatedAtRoute("GetProductById", new { id = productToReturn.Id }, productToReturn);
}

完成所有这些更改后，我们可以发送 post 请求，但这一次，我们将在响应正文中看到一个新创建的产品，并且在Header中，还会看到一个叫Location的Key，它的Value是一个连接，可以用来获取该新产品的信息：

好了，基本的新增和查询操作就到这里了，修改和删除可以按照这个套路举一反三。

MediatR通知

我们注意到，Request有且只能有一个Handler来处理，但是如果我们需要有多个Handler怎么办呢？这时候就需要用到通知了，通知的使用场景通常是在一个事件发生后，需要有多个响应。例如我们添加了产品后，需要：

发送邮件通知
作废缓存

为了演示通知的使用，我们需要修改AddProductCommand，在完成Product添加操作后发送一个通知出来。

发送电子邮件和使缓存失效超出了本文的范围，但为了演示通知的行为，让我们简单地更新我们的Fake数据来表示已处理某些内容。

打开FakeDataStore并添加一个方法：

public async Task EventOccured(Product product, string evt)
{
    _products.Single(p => p.Id == product.Id).Name = $"{product.Name} evt: {evt}";
    await Task.CompletedTask;
}

创建通知和处理程序

让我们定义一条通知消息，用于封装我们要定义的事件。

首先，让我们添加一个名为“Notifications”的新文件夹，在该文件夹中添加一个名为ProductAddedNotification的record。

public record ProductAddedNotification(Product Product) : INotification;

这个record继承了INotification，并且拥有一个Product属性。

现在，我们为通知创建两个处理程序：

public class EmailHandler : INotificationHandler<ProductAddedNotification>
{
    private readonly FakeDataStore _fakeDataStore;
    public EmailHandler(FakeDataStore fakeDataStore) => _fakeDataStore = fakeDataStore;
    public async Task Handle(ProductAddedNotification notification, CancellationToken cancellationToken)
    {
        await _fakeDataStore.EventOccured(notification.Product, "Email sent");
        await Task.CompletedTask;
    }
}

public class CacheInvalidationHandler : INotificationHandler<ProductAddedNotification>
{
    private readonly FakeDataStore _fakeDataStore;
    public CacheInvalidationHandler(FakeDataStore fakeDataStore) => _fakeDataStore = fakeDataStore;
    public async Task Handle(ProductAddedNotification notification, CancellationToken cancellationToken)
    {
        await _fakeDataStore.EventOccured(notification.Product, "Cache Invalidated");
        await Task.CompletedTask;
    }
}

这两个类做了同样的两件事：

实现INotificationHandler<ProductAddedNotification>接口表示它可以处理ProductAddedNotification通知。
在 FakeDataStore 上调用 EventOccured 方法。

在实际用例中，这些将以不同的方式实现，并且可能会采用一些外部依赖，但在这里我们只是尝试演示通知的行为。

触发通知

接下来，我们需要实际触发通知。

打开ProductsController并且修改AddProduct方法：

[HttpPost]
public async Task<ActionResult> AddProduct([FromBody]Product product)
{
    var productToReturn = await _mediator.Send(new AddProductCommand(product));
    await _mediator.Publish(new ProductAddedNotification(productToReturn));
    return CreatedAtRoute("GetProductById", new { id = productToReturn.Id }, productToReturn);
}

除了要发送AddProductCommand请求外，还需要向MediatR发送ProductAddedNotification通知，但是这次需要使用Publish方法，而不是Send。

我们也可以把通知的发送放到AddProductCommand命令Handler里面。

测试通知

运行项目，先在Postman中运行GetProducts请求。

接下来运行AddProduct请求，调用成功之后重新运行GetProducts请求。

正如预期的那样，当我们添加新产品时，两个事件都会触发并编辑名称。虽然这是一个简单且略显粗糙的例子，但这里的关键要点是，我们可以通过MediatR触发一个事件并使用不同的Handler多次处理它，而生产者不知道任何不同。

如果我们想扩展我们的工作流程来执行额外的任务，我们可以简单地添加一个新的处理程序。我们不需要修改通知本身或所述通知的发布，这再次触及了早期的可扩展性和关注点分离。

构建MediatR行为

通常，当我们构建应用程序时，我们有许多跨领域问题。其中包括授权、验证和日志记录。我们可以利用 Behavior，而不是在整个处理程序中重复此逻辑。MediatR的Behavior与ASP.NET Core中间件非常相似，它们接受请求，执行某些操作，然后（可选）传递请求。

创建Behavior

首先我们在项目下新建一个名为“Behaviors”的文件夹。然后在文件夹中添加一个类，命名为LoggingBehavior：

public class LoggingBehavior<TRequest, TResponse> : IPipelineBehavior<TRequest, TResponse>
    where TRequest : IRequest<TResponse>
{
    private readonly ILogger<LoggingBehavior<TRequest, TResponse>> _logger;
    public LoggingBehavior(ILogger<LoggingBehavior<TRequest, TResponse>> logger)
        => _logger = logger;
    public async Task<TResponse> Handle(TRequest request, RequestHandlerDelegate<TResponse> next,
        CancellationToken cancellationToken) 
    {
        _logger.LogInformation($"Handling {typeof(TRequest).Name}"); 
        var response = await next(); 
            
        _logger.LogInformation($"Handled {typeof(TResponse).Name}"); 
            
        return response;
    }
}

解释一下这段代码：

LoggingBehavior包含两个泛型参数TRequest和TResponse，继承了IPipelineBehavior<TRequest, TResponse>接口。从泛型参数可以看出，这个Behavior可以处理任何请求。
LoggingBehavior实现了Handle方法，在调用next()委托之前和之后进行日志记录。

注册Behavior

打开Program.cs，增加一行代码：

builder.Services.AddSingleton(typeof(IPipelineBehavior<,>), typeof(LoggingBehavior<,>));

测试Behavior

运行项目，打开Postman并执行任意一个请求，查看控制台输出：

OK，看到这个界面说明LoggingBehavior已经正常工作了。

我们在没有修改任何业务代码的情况下，轻松地用AOP的方式实现了日志记录。

结论

我们用了两篇文章介绍如何使用 MediatR 在核心 ASP.NET 实现 CQRS 和中介器模式。我们已经完成了请求和通知，以及如何处理行为的横切问题。

MediatR 为需要从简单的单体架构演变为更成熟的应用程序提供了一个很好的起点，它允许我们分离读取和写入关注点，并最大限度地减少代码之间的依赖关系。

这为我们采取其他几个可能的步骤提供了有利条件：

使用不同的数据库进行读取（也许可以通过扩展我们的 ProductAddedNotification 来添加第二个处理程序，将数据写入新的数据库，然后修改 GetProductsQuery 以从该数据库读取数据）
将我们的读取/写入分离到不同的应用程序中（修改 ProductAddedNotification 以发布到 Kafka/服务总线，然后让第二个应用程序从消息总线中读取）。

现在，我们的应用程序已经处于一个很好的状态，可以在需要时采取上述步骤，而不会在短期内使事情过于复杂。

点关注，不迷路。

如果您喜欢这篇文章，请不要忘记点赞、关注、转发，谢谢！如果您有任何高见，欢迎在评论区留言讨论……