从壹开始微服务 [ DDD ] 之终篇 ║当事件溯源 遇上 粉丝活动

2、事件溯源的执行过程

这个时候你仔细想一想，我们在和领域专家（默认他们不懂技术）讨论用户下单流程的时候，专家一定会说：客户首先选择一个商品，然后添加到购物车，确认无误下单，接着用户支付，支付成功后，就给用户发货。而我们呢，我们作为一个开发人员，和领域专家讨论的时候，自然而然的也是这么思考的，对不对！（你肯定在讨论需求的时候用的不是数据库的思维！），只不过我们后期开发的时候，拘泥于技术和数据优先的思维，不得不转向CURD的道路了，当然这个没有什么错误，我只是说明一点，事件存储真的离我们不远。

那我们平时是怎么做的呢，这里说一个特别简单的：

从这个特别简单的流程中我们可以看到，平时我们都是直接操作的 Order 这个领域聚合根，一直在修改模型状态，这个看似正常的操作下，有一些问题，是我们建立在每一步都正常执行的情况下，不过一般总会出现一些问题，特别是分布式的环境中。

然而，事件溯源与上述的情况恰好相反，它并不关心当前状态，而是关注持续不断的变化事件。

举个例子，假设我们有一个“购物车”，我们可以创建购物车，往里面添加商品或移除商品，然后结账。

购物车的生命周期可以包含如下一系列事件：

创建购物车

往购物车里添加商品

再次往购物车里添加商品

从购物车里移除商品

结账

这些就是一个购物车的生命周期，包含了一系列事件。这就是事件溯源，非常简单吧？

几乎所有的流程都可以被看成一系列事件。在与领域专家交谈时，他们不会提及“表”和“连接”，他们会将流程描述成一系列事件以及可以应用在这些事件上的规则。

3、事件溯源是如何更新实时状态的

二、事件溯源的存在意义与问题

事件溯源不是万能的，不过它可以在某一些领域发挥很大的作用，这个在以后的微服务设计中，会更能体现出来，那我们就简单说两点：

1、传统应用中出现的某些问题

从业务的角度

传统的应用中，数据库里存的是Domain Model的实例的当前状态，比如某个储户银行账户的存款数，通常是一个数字.如果考虑到如下的三个情形,我们可能付出的代价比较大：

１) 老规则：问题跟踪

如果某个储户的账户出现问题，那么我们只有从大到PB的日志中去分析用户的账户数据是如何出错的，而且我们在做日志的时候，不可能所有的都考虑到，就算是把全部数据都保存，时间都记下来，操作者都备份，那ATM机信息呢？（可能不恰当，只是说明我们总有想不到的地方），但如果一旦日志不够详细，找出问题根源基本只能靠猜了。

２) 新需求：趋势分析

历史数据的作用在于分析未来的趋势，如果仅仅从浩如烟海的日志中寻找规律，我们还得单独写逻辑，对日志进行建模，清洗，其实我们已经能接受，日志就是用来记录异常信息的，这个时候我们就很崩溃了。

３) 更奇葩：事务回滚

在介绍事务修正模式中，我们讲到某个步骤发生错误，之前的各个节点可以自己独立地完成回滚，回滚的依据就是记录的操作步骤及相关参数，根据这些有用信息就可以每个节点自行回滚到原始状态，并且在失败的时候可以retry

可见存储对于Domain Model 的各个事件还是非常有用的,尤其是对于复杂的系统，这也就是我们今天要讨论的事件溯源模式.

有许多理由使用Event Sourcing，当你浏览Greg Young的系列文章和谈话你会发现下面要点：

1. 它不是一个新概念，真实世界中许多领域都很像它，看看你的银行账户状态，比如储蓄卡，它打印出一笔笔进出明细和当前余额，这一笔笔代表了领域事件。

2.通过重播事件，我们能够得到对象的任何时刻状态(这里应该用正确术语：聚合aggregate)，比如储蓄卡每笔记录的当前余额代表你这个账户聚合对象的某刻时刻的状态，这可能会极大地帮助我们理解领域知识，当前状态是怎么来，因为什么改变？方便调试关键问题的错误

3.领域中当前状态和存储数据库中的数据没有任何耦合，而传统上我们都是将应用状态存储到数据库中，比如储蓄卡当前余额100元存储到数据库中，现在我们存储导致余额的进出事件了，存款了多少钱，取

款了多少钱，这一笔笔领域事件都会记录在数据库中。

4.Append-only追加模型存储这些事件，易于扩展，这样我们无论读写都有很好地性能，读取能够转为查询优化，也可以转为写优化（因为没有读，写得很快），读写分离。

5除了可以存储用户意图数据，也就是操作事件，事件存储顺序能够用来分析用户正在做什么，通往大数据。

6.我们能避免了对象与关系数据库的不匹配。

7.审计日志是免费的，一次审计日志所有变化，因为没有状态改变，只有事件。

这样不会浪费时间吗？

一点也不。一般来说，要执行约束，只需要获得事件的一个很小子集。通过简单的数据库查询就可以获得有用的历史事件，在加载完这些事件后重放它们，把它们“投射”出来，以此构建你的数据集。这样的操作其实是很快的，因为你使用的是本地的处理器，而不是执行一系列SQL查询（跨域网络的调用要比本地操作慢得多，至少会相差两个数量等级）。

你可以在后台构建数据集，然后把中间结果保存在数据库里。这样，用户就可以在很短的时间内查询到这些数据。

4、什么时候使用这种模式

这种模式在以下几种场景中是最理想的解决方案：

• 当你想获得数据的“意图”，“目的”或者“原因”的时候。 例如，一个客户的实体改变可能用一系列的类似于”搬家“，”注销账户“或者”死亡“等事件类型。
• 并发更新数据时候非常需要减少或者完全避免冲突的时候。
• 当你需要保存已经发生的事件，并且能够重播他们来还原到某个状态、使用这些事件去回滚系统的某些变化或者仅仅是历史或者审查记录的时候。例如 ，当一个任务包括几个步骤，你可能需要执行一个撤销更新的操作然后重播过去的每个步骤来回到稳定的状态。
• 当使用事件是一些应用程序的某些操作的天然属性，并且需要很少的额外扩展或者实施的时候。
• 当你需要把插入，更新数据和需要执行这些操作的应用程序解耦开的时候。用这种模式可以提高UI的性能，或者把这些事件分发给其他的监听者，比如有些应用程序或系统，它们在一些事件发生的时候必须做出一些反应。例如，将一个工资系统和一个报销系统结合起来，这样的话当报销系统更新一个事件给事件数据库，数据库对此做出的相应事件就可以被报销系统和工资系统共享。
• 当要求变更或者——当和CQRS配合使用的时候——你需要适配一个读的模型或者视图来显示数据，而你想要更灵活地改变物化视图的格式和实体数据的时候。
• 当和CQRS配合使用的时候，并且当一个读模型被更新时能接受数据的最终一致性问题，或者说从一系列的事件序列中生成实体对性能的影响可以被接受。

这种模式在以下几种场景中可能并不适用：

• 小而简单的，业务逻辑简单或者根本没有业务逻辑，或者领域概念的，一般传统的增删改查（CURD）就能实现功能的业务领域，或者系统。
• 需要实时一致和实时更新数据的系统。
• 不需要审查，历史和回滚的系统。
• 并发更新数据可能性非常小的系统。例如，只增加数据不更新数据的系统。

5、ES 和 CQRS 的关系

CQRS与事件溯源有着相辅相成的关系。CQRS允许事件溯源作为领域的数据存储机制。然而，使用事件溯源的一个最大的缺点是，你无法向你的系统提出类似“请告诉我所有名字为Greg的用户”这样的问题，这是由于事件溯源无法提供对象的当前状态而引起的。CQRS唯一支持的查询就是：GetById - 通过ID来获得某个聚合。下图为基于CQRS/ES的应用系统结构：

CQRS经常和事件溯源模式结合使用

基于CQRS的系统使用分离的读和写模型，每一个都对应相应的任务并且一般储存在不同的数据库中。当和事件溯源模式一起使用的时候，一系列的事件存储相当于“写”模型，是所有信息的可信赖来源（authoritative source ）。基于CQRS的系统的读模型提供了数据的物化视图，经常是一种高度格式化的视图形式。这些视图对应相应的界面并且展示了应用程序的需求，帮助最大化展示和查询效率。

使用一系列的事件当作“写”而不是某一个时间点的数据，避免了更新的冲突并且最大化性能和系统的伸缩性，这些事件可以被异步地产生被用来展示数据的物化视图。

因为事件数据库是所有信息的可信赖来源，当系统改进的时候，有可能删除物化视图并且展示所有过去的时间来产生一个新的数据，或者当读模型必须改变的时候。物化视图是一个长久的数据缓存。

当将CQRS和事件溯源模式结合起来的时候，考虑以下几点：

• 对于任何的读写分离储存的系统，这些系统基于事件溯源模式都是“最终一致”的。因此在事件产生和数据存储之间会有一些延迟。
• 这种模式会造成一些额外的复杂度，因为代码必须要能够初始化和处理事件，然后组合或者更新相应的读写模型需要视图或者对象。这种复杂度会对让系统的实现变得有些困难，需要重新学习一些概念和一个不同的设计系统的方式。然而事件溯源可以让为领域建模，让重建视图或者对象更加容易。
• 生成物化视图

CQRS最核心的概念是Command、Event，“将数据(Data)看做是事实(Fact)。每个事实都是过去的痕迹，虽然这种过去可以遗忘，但却无法改变。” 这一思想直接发展了Event Source，即将这些事件的发生过程记录下来，使得我们可以追溯业务流程。CQRS对设计者的影响，是将领域逻辑，尤其是业务流程，皆看做是一种领域对象状态迁移的过程。这一点与REST将HTTP应用协议看做是应用状态迁移的引擎，有着异曲同工之妙。

1、必须自己实现事务的统一commit和rollback：这个是无论哪一种方式，都必须面对的问题。完全逃不掉。在DDD中有一个叫Saga的概念，专门用于统理这种复杂交互业务的，CQRS/ES架构下，由于本身就是最终一致性，所以都实现了Saga，可以使用该机制来做微服务下的transaction治理。

 

 

2、请求幂等：请求发送后，由于各种原因，未能收到正确响应，而被请求端已经正确执行了操作。如果这时重发请求，则会造成重复操作。
CQRS/ES架构下通过AggregateRootId、Version、CommandId三种标识来识别相同command，目前的开源框架都实现了幂等支持。

 

 

3、并发：单点上，CQRS/ES中按事件的先来后到严格执行，内存中Aggregate的状态由单一线程原子操作进行改变。
多节点上，通过EventStore的broker机制，毫秒级将事件复制到其他节点，保证同步性，同时支持版本回退。(Eventuate)

namespace Christ3D.Domain.Core.Events
{
    /// <summary>
    /// 抽象类Message，用来获取我们事件执行过程中的类名
    /// 然后并且添加聚合根
    /// </summary>
    public abstract class Message : IRequest
    {
        public string MessageType { get; protected set; }
        public Guid AggregateId { get; protected set; }

        protected Message()
        {
            MessageType = GetType().Name;
        }
    }
}

同时在该文件夹下，新建 存储事件 模型StoredEvent.cs

    public class StoredEvent : Event
    {
        /// <summary>
        /// 构造方式实例化
        /// </summary>
        /// <param name="theEvent"></param>
        /// <param name="data"></param>
        /// <param name="user"></param>
        public StoredEvent(Event theEvent, string data, string user)
        {
            Id = Guid.NewGuid();
            AggregateId = theEvent.AggregateId;
            MessageType = theEvent.MessageType;
            Data = data;
            User = user;
        }

        // 为了EFCore能正确CodeFirst
        protected StoredEvent() { }
        // 事件存储Id
        public Guid Id { get; private set; }
        // 存储的数据
        public string Data { get; private set; }
        // 用户信息
        public string User { get; private set; }
    }

namespace Christ3D.Infra.Data.Mappings
{    
    /// <summary>
    /// 事件存储模型Map
    /// </summary>
    public class StoredEventMap : IEntityTypeConfiguration<StoredEvent>
    {
        public void Configure(EntityTypeBuilder<StoredEvent> builder)
        {
            builder.Property(c => c.Timestamp)
                .HasColumnName("CreationDate");

            builder.Property(c => c.MessageType)
                .HasColumnName("Action")
                .HasColumnType("varchar(100)");
        }
    }
}

2、然后再上下文文件夹 Context 下，新建事件存储Sql上下文 EventStoreSQLContext.cs

namespace Christ3D.Infra.Data.Context
{
    /// <summary>
    /// 事件存储数据库上下文，继承 DbContext
    /// 
    /// </summary>
    public class EventStoreSQLContext : DbContext
    {
        // 事件存储模型
        public DbSet<StoredEvent> StoredEvent { get; set; }

        protected override void OnModelCreating(ModelBuilder modelBuilder)
        {
            modelBuilder.ApplyConfiguration(new StoredEventMap());

            base.OnModelCreating(modelBuilder);
        }

        protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder)
        {
            // 获取链接字符串
            var config = new ConfigurationBuilder()
                .SetBasePath(Directory.GetCurrentDirectory())
                .AddJsonFile("appsettings.json")
                .Build();

            // 使用默认的sql数据库连接
            optionsBuilder.UseSqlServer(config.GetConnectionString("DefaultConnection"));
        }
    }
}

这里要说明下，因为已经创建了两个上下文，以后迁移的时候，就要加上 上下文名称 了：

namespace Christ3D.Infra.Data.Repository.EventSourcing
{
    /// <summary>
    /// 事件存储仓储接口
    /// 继承IDisposable ，可手动回收
    /// </summary>
    public interface IEventStoreRepository : IDisposable
    {
        void Store(StoredEvent theEvent);
        IList<StoredEvent> All(Guid aggregateId);
    }
}

2、然后对上边的接口进行实现

namespace Christ3D.Infra.Data.Repository.EventSourcing
{
    /// <summary>
    /// 事件仓储数据库仓储实现类
    /// </summary>
    public class EventStoreSQLRepository : IEventStoreRepository
    {
        // 注入事件存储数据库上下文
        private readonly EventStoreSQLContext _context;

        public EventStoreSQLRepository(EventStoreSQLContext context)
        {
            _context = context;
        }

        /// <summary>
        /// 根据聚合id 获取全部的事件
        /// 这个聚合是指领域模型的聚合根模型
        /// </summary>
        /// <param name="aggregateId"> 聚合根id 比如：订单模型id</param>
        /// <returns></returns>
        public IList<StoredEvent> All(Guid aggregateId)
        {
            return (from e in _context.StoredEvent where e.AggregateId == aggregateId select e).ToList();
        }

        /// <summary>
        /// 将命令事件持久化
        /// </summary>
        /// <param name="theEvent"></param>
        public void Store(StoredEvent theEvent)
        {
            _context.StoredEvent.Add(theEvent);
            _context.SaveChanges();
        }

        /// <summary>
        /// 手动回收
        /// </summary>
        public void Dispose()
        {
            _context.Dispose();
        }
    }
}

这个时候，我们的事件存储模型、上下文和仓储层已经建立好了，也就是说我们可以对我们的事件模型进行持久化了，接下来就是在建立服务了，用来调用仓储的服务，就好像我们的应用服务层的概念。