ASP.NET Core Web API下事件驱动型架构的实现(五):在微服务中使用自我监听模式保证数据库更新与消息派发的可靠性

By | 2018年7月14日

上一讲中,我介绍了CQRS架构中聚合与聚合根的实现,并通过单元测试验证了设计的正确性。这部分内容比较难,在实际应用过程中存在一定的门槛,所以感觉关注的读者并不是特别多。然而,CQRS本身作为一种事件驱动型架构,解决了现在流行的微服务中的一些细节问题,我认为还是非常有必要探究讨论的。这部分内容我会在另外讨论微服务架构的文章中详细注解,在这里就不多做说明了。总之,研讨CQRS架构的实现,是面向微服务进行事件驱动型架构实践的必经之路,我们会不断地深入下去。

问题分析

在本文中,我打算还是先绕过CQRS的事情,讨论一下数据一致性问题。让我们回到本系列文章的第三讲结束时的状态,在完成第三讲的内容后,我们已经可以实现一个基于RabbitMQ的事件总线,并向RabbitMQ这一真实的事件总线上派发事件。例如,在CustomersController中,有以下代码:

// 创建新的客户信息
[HttpPost]
public async Task<IActionResult> Create([FromBody] dynamic model)
{
    this.logger.LogInformation($"开始创建客户信息。");
    var name = (string)model.Name;
    if (string.IsNullOrEmpty(name))
    {
        return BadRequest("请指定客户名称。");
    }

    var email = (string)model.Email;
    if (string.IsNullOrEmpty(email))
    {
        return BadRequest("电子邮件地址不能为空。");
    }

    const string sql = "INSERT INTO [dbo].[Customers] ([CustomerId], [CustomerName]) VALUES (@Id, @Name)";
    using (var connection = new SqlConnection(connectionString))
    {
        var customer = new Model.Customer(name);
        await connection.ExecuteAsync(sql, customer);
        await this.eventBus.PublishAsync(new CustomerCreatedEvent(customer.Id, name, email));
        this.logger.LogInformation($"客户信息创建成功。");
        return Created(Url.Action("Get", new { id = customer.Id }), customer.Id);
    }
}

这部分代码逻辑上并没有什么大问题,它通过HTTP POST请求发过来的数据创建客户实体,并将实体保存到数据库中,同时向事件总线发布一条CustomerCreatedEvent事件。然而,这里是存在数据不一致的可能性的:如果CustomersController进程在执行connection.ExecuteAsync之后,而恰好在eventBus.PublishAsync执行之前异常退出了,那么,可能导致的结果是:客户实体被成功保存到数据库中,而事件派发失败。RabbitMQ等流行的消息总线机构虽然具备未投递消息的持久化机制,但很有可能此处都还没有连接上RabbitMQ,CustomerControllers进程就已经退出了,于是,需要对CustomerCreatedEvent事件进行处理的其它微服务进程就不再会捕获到这一事件,数据的不一致性就产生了。

由此引来了微服务架构(MSA)中的一个非常普遍而又比较复杂的问题:微服务间的通信。总的来说,微服务间通信有同步和异步两种方式,同步方式使用RESTful API,通过服务注册、服务发现以及API调用完成微服务之间的同步通信,这是广为人知并且也是被广泛使用的;另一种就是基于事件的异步通信方式。事实上,无论是同步还是异步,都存在着数据一致性问题,因为在经典的微服务架构中,数据存储部分都是相对独立的,技术选型也是非常复杂多样,因此,在这样的应用场景中,往往无法通过由两阶段提交模式(2PC)所实现的分布式事务来达到数据的一致性需求,这部分内容我会在另外讨论微服务的文章中详细介绍。

那么如果我们将数据库操作和事件消息派发的顺序改变一下呢?结果还是一样,还是有可能存在事件消息派发成功,而数据库更新不成功的情况,数据一致性仍然无法得到保障。因此,基于目前的代码结构,需要保证数据的一致性还是比较困难的。

解决方案

在微服务的实践中,解决这一问题的常用方法有四种:

  • 自我监听模式(Listen to Yourself):也是本文介绍的方案,这种方式的优点是实现比较简单,缺点是无法以同步的方式及时将结果返回调用方
  • 追踪数据库的事务日志(tailing transaction logs):这种方案的优点是在保证了数据一致性的前提下,能够同步地返回结果,缺点主要有两个:一是这种实现需要依赖于数据库系统,有些数据库系统并不提供事务日志的功能;另一个缺点是,由于是将数据库日志通过消息的方式发送出去,因此,事件消息本身仅能表达比较单纯的数据库层面的信息,而无法表达领域模型相关的信息(比如:发送出去的消息很可能是“在用户数据表中修改了一条记录”,而不是“用户地址发生了更改”)
  • 使用本地数据库的ACID事务:这种方案会需要在微服务本地增添额外一张保存事件的数据表,当业务数据表需要更新时,通过数据库本地事务,同时更新事件数据表,以此达到本地数据一致性,然后,通过另外一个任务调度器,定期从事件数据表读取事件并派发到事件总线。消息派发成功之后,标记记录为“已派发”状态。这样做既保证了数据一致性,同时也能同步返回结果,也不会存在所派发的消息信息与领域模型不相关的问题,但实现过程相对比较复杂,需要引入额外的任务调度器。此外,本地数据库需要选用支持ACID事务的数据库,存在一定的技术选型局限性
  • 使用事件溯源(Event Sourcing):优点是只需要发布事件消息,所有领域对象的状态都由事件描述,不存在不一致的问题;缺点是实现相对复杂,需要处理消息对等问题,另外也无法通过同步方式返回结果

接下来介绍一下自我监听模式的实现,对于第二种和第三种方案,在后续的文章中我还会继续讨论,但应该不会提供具体的实现细节。对于事件溯源,也将是我们讨论的重点。

自我监听模式(Listen to Yourself)

简单来说,自我监听模式可以用下图表示:

image

基本步骤如下:

  1. 客户向用户管理微服务发起地址变更请求
  2. 用户管理微服务不直接修改数据库,而是向消息总线发送一条地址变更消息
  3. 订单管理微服务订阅地址变更消息,在消息处理过程(Event Handler)中,更新相关订单的收货地址
  4. 用户管理微服务订阅地址变更消息,在消息处理过程中,更新用户的联系地址。由于用户管理微服务所订阅的地址变更消息正是由它自己发起的,因此,自我监听模式由此得名

总体上讲,逻辑还是比较简单,实现起来也不复杂。我们再分析一下,看这样的模型还会不会存在数据不一致的问题。

  • 如果消息总线在地址变更消息到达之前发生异常,那么地址变更消息将不会被发出,用户管理和订单管理微服务都不会更新数据库
  • 如果消息总线在地址变更消息到达之后发生异常,那么地址变更消息会同时被派发到用户管理和订单管理微服务,两者的数据库都会同时更新
  • 如果微服务在获得地址变更消息后,更新数据库时发生异常,此时消息总线无法获得微服务的Acknowledgement(确认信息),消息数据还是保存在消息总线。等下一次微服务重启,或者新的实例产生后,仍然会获得地址变更消息,并继续处理之前未完成的事件处理逻辑。市面上常见的消息总线产品都具备这种机制,确保消息的正确派发

那如果在微服务反馈Acknowledgement时出现异常呢?此时数据库已经更新完成,而消息总线却无法得知其实事件已被成功处理,于是,当新的微服务实例启动时,它仍然会再次获得事件通知,然后尝试更新数据库。因此,微服务还需要有处理事件信息不对等的情况,比如事件冗余,我会在其它文章中进行讨论。接下来看看如何在已有的ASP.NET Core Web API的项目中实现自我监听模式。

模式实现

为了模拟上述模式,我们在已有的案例项目中增加一个Notification的微服务。当有新客户注册成功时,该服务会向客户邮箱发送欢迎邮件。为了简单起见,此处我们就不发送电子邮件了,而是向日志文件中输出一条信息,表示邮件已经发送。

Notification微服务的项目结构跟案例中现有的Customer微服务的项目结构类似,在此就不多做说明了,在Visual Studio中按照Customer项目结构进行搭建就行了。除了实现一个新的微服务之外,我们还需要做一些简单的重构,第一步就是将现有的CustomerCreatedEvent事件代码从Customer微服务中移到一个公共的类库中,因为从现在开始,不仅仅是Customer微服务,Notification微服务也需要能够引用CustomerCreatedEvent事件类型,并通过事件处理器(Event Handler)来处理CustomerCreatedEvent。

Note:从架构设计的角度,理论上将CustomerCreatedEvent移到公共的类库中并不一定是必须的,道理很简单,不同微服务使用的技术可以是截然不同的,甚至是完全不同的体系架构,因此,即使从C#代码的角度将其提取成公共类型,也并不一定能够用在其它的微服务中。事实上,只需要保证流动在整个应用程序中的所有事件数据满足一定的规范,大家都懂就可以了。在这里,将CustomerCreatedEvent类型移到一个公共的类库中,主要还是为了编程方便,因为我们本身就是在用.NET和C#。

接下来,我们在Customer微服务中增加一个类,用来处理CustomerCreatedEvent事件:

public class AddNewCustomerEventHandler : EdaSample.Common.Events.EventHandler<CustomerCreatedEvent>
{
    private readonly ILogger logger;
    private readonly IOptions<MssqlConfig> config;

    public AddNewCustomerEventHandler(ILogger<AddNewCustomerEventHandler> logger, IOptions<MssqlConfig> config,
        IConfiguration configuration)
    {
        this.logger = logger;
        this.config = config;
    }

    public override async Task<bool> HandleAsync(CustomerCreatedEvent @event, CancellationToken cancellationToken = default)
    {
        const string sql = "INSERT INTO [dbo].[Customers] ([CustomerId], [CustomerName]) VALUES (@Id, @Name)";
        using (var connection = new SqlConnection(this.config.Value.ConnectionString))
        {
            var customer = new Model.Customer(@event.CustomerId, @event.CustomerName);
            await connection.ExecuteAsync(sql, customer);
            this.logger.LogInformation($"客户信息创建成功。");
        }

        return true;
    }
}

然后将该事件处理器注册到事件总线上:

public void Configure(IApplicationBuilder app, IHostingEnvironment env)
{
    var eventBus = app.ApplicationServices.GetRequiredService<IEventBus>();
    eventBus.Subscribe<CustomerCreatedEvent, CustomerCreatedEventHandler>();
    eventBus.Subscribe<CustomerCreatedEvent, AddNewCustomerEventHandler>();

    if (env.IsDevelopment())
    {
        app.UseDeveloperExceptionPage();
    }

    app.UseMvc();
}

最后,修改我们的CustomersController控制器,在创建新客户的时候,简单地发出一个CustomerCreatedEvent即可:

// 创建新的客户信息
[HttpPost]
public async Task<IActionResult> Create([FromBody] dynamic model)
{
    this.logger.LogInformation($"开始创建客户信息。");
    var name = (string)model.Name;
    if (string.IsNullOrEmpty(name))
    {
        return BadRequest("请指定客户名称。");
    }

    var email = (string)model.Email;
    if (string.IsNullOrEmpty(email))
    {
        return BadRequest("电子邮件地址不能为空。");
    }
    
    // 由于数据库更新需要通过事件处理器进行异步更新,因此无法在Controller中得到
    // 数据库更新后的Customer ID。此处通过Guid.NewGuid获得,实际中可以使用独立
    // 的Identity Service产生。
    var customerId = Guid.NewGuid();

    await this.eventBus.PublishAsync(new CustomerCreatedEvent(customerId, name, email));

    return Created(Url.Action("Get", new { id = customerId }), customerId);
}

从上面的代码可以看到这种模式的弊端了:虽然在Create方法中返回了HTTP 201 Created,但实际上Customer和Notification还在处理CustomerCreatedEvent事件,或者已经处理成功,或者已经处理失败,无论如何,客户端在得到RESTful API的返回结果时,客户创建的动作很有可能还未完成,因此,客户端无法同步地得到确切的结果,也就是说,下一步当客户端发起查询请求时,很有可能查询并不能返回任何结果。

另一个有趣的事情是,目前我们使用CustomerCreatedEvent类来搭载客户创建的事件信息,但Created这一词听起来好像是表示“已经创建”的意思,这与我们在前两章中所描述的实现并不一样:时间点变掉了。之前是可以确切表示“已经创建”的概念的,因为在发送事件消息之前,Customer微服务中的事情其实已经完成了。而此处则不然:此处仅仅是发送了一个事件就立刻返回了,而客户数据很有可能还没有创建。所以,使用CustomerCreatedEvent这一名称来定义事件消息就显得有些不太妥当。或许可以改个名字,也算是个不错的主意。

总结

本文介绍了微服务下保证数据保存和消息派发一致性的一种模式:自我监听模式。该模式的实现需要依赖于消息总线对消息派发可靠性的保证,也就是在微服务中没有作Acknowledge的情况下,事件消息将会被再次派发,就像上面所述,在这种情况下还应该预防消息被重复发送的情况。当然,除了自我监听模式之外,还有三种常见的做法来保证数据的一致性。在本系列文章中,会对最后一种:事件溯源(Event Sourcing)进行介绍,对于另外两种,由于实现方式都需要依赖于特定类型的数据库,并且架构相对不算复杂,也就不打算深入介绍了。

有关微服务的其它话题,可以参考我另外一部分关于微服务的文章。Happy Coding…

源代码的使用

本系列文章的源代码在https://github.com/daxnet/edasample这个Github Repo里,通过不同的release tag来区分针对不同章节的源代码。本文的源代码请参考chapter_5这个tag,如下:

image

(总访问量:964;当日访问量:1)

6 thoughts on “ASP.NET Core Web API下事件驱动型架构的实现(五):在微服务中使用自我监听模式保证数据库更新与消息派发的可靠性

  1. 庞建新

    最近正好要拿DDD做一个案例的项目,正好您这边有,太感谢了

    Reply
  2. 冉啟生

    前辈:
    自监听这种模式就是通过发送事件存储到消息队列,通过消息队列的重试来保证业务完整性,对吧?
    那如果客户端需要时时知道事件处理结果的话,有哪些比较好的方式解决呢?
    还是必须通过用户体验调整的形式来处理,比如界面显示正在处理?

    Reply
    1. daxnet Post author

      @冉啟生
      您好,如果需要实时反馈客户端,那么就不能使用自我监听模式来保证数据的一致性,就需要使用本文中提到的追踪数据库的事务日志方式,或者使用本地数据库ACID的方式。我后文可能会介绍这些东西,不过近期可能不会有时间。如果急用,可以上网搜索一下相关内容。感谢您的关注!

      Reply
  3. Philia

    拜读了,博主的文章质量很高。CQRS架构的优缺点和技术细节都谈到了。

    对应到实际的项目中,以订单服务为例,微服务的拆分往往会物理上分出“下单服务”和“查询订单服务”两个服务,“下单服务”又会拆分出“下单服务”和下游的“消息处理服务”。

    类似的还有互联网/银行之类的业务架构大多是物理上的CQRS,不知道您怎么看实际项目中基础框架层面的CQRS架构应用难的问题?

    Reply
    1. daxnet Post author

      您好!在实际项目中,CQRS应用困难,这是可以理解的。作为一种架构风格,我觉得需要权衡其利弊,比如,CQRS能够带来什么?又有什么是我们无法驾驭的?如果团队对于CQRS的复杂度有一定的掌控能力,并且CQRS确实能够解决项目实际问题,例如微服务架构中的事件派发与事件溯源,那么,可以考虑采用CQRS,否则,或许其它的架构风格更合适。

      Reply
  4. Pingback: Saga体系结构模式:微服务架构下跨服务事务的实现 – sunnycoding

回复 Philia 取消回复

您的电子邮箱地址不会被公开。 必填项已用*标注

此站点使用Akismet来减少垃圾评论。了解我们如何处理您的评论数据