CQRS+ES项目解析Diary.CQRS

在《当我们在讨论CQRS时&＃xff0c;我们在讨论些神马》中&＃xff0c;我们讨论了当使用CQRS的过程中&＃xff0c;需要关心的一些问题。其中与CQRS关联最为紧密的模式莫过于Event Sourcing了&＃xff0c;CQRS与ES的结合&＃xff0c;为我们构造高性能、可扩展系统提供了基本思路。本文将介绍
Kanasz Robert在《Introduction to CQRS》中的示例项目Diary.CQRS。

获取Diary.CQRS项目

该项目为Kanasz Robert为了介绍CQRS模式而写的一个测试项目&＃xff0c;原始项目可以通过访问《Introduction to CQRS》来获取&＃xff0c;由于项目版本比较旧&＃xff0c;没有使用nuget管理程序包等&＃xff0c;导致下载以后并不能正常运行&＃xff0c;我下载了这个项目&＃xff0c;升级到Visual Studio 2017&＃xff0c;重新引用了StructMap框架&＃xff08;使用nuget&＃xff09;&＃xff0c;移除了Web层报错的代码&＃xff0c;并上传到博客园&＃xff0c;可以从这里下载&＃xff1a;Diary.CQRS.rar

Diary.CQRS项目简介

Diary.CQRS项目的场景为日记本管理&＃xff0c;提供了新增、编辑、删除、列表等功能&＃xff0c;整个解决方案分为三个项目&＃xff1a;

• Diary.CQRS&＃xff1a;核心项目&＃xff0c;完成了EventBus、CommandBus、Domain、Storage等功能&＃xff0c;也是我们分析的重点。
• Diary.CQRS.Configuration&＃xff1a;服务配置&＃xff0c;通过ServiceLocator类进行依赖注入、服务查找功能。
• Diary.CQRS.Web&＃xff1a;用户界面&＃xff0c;MVC项目。

这是一个很好的入门项目&＃xff0c;功能简单、结构清晰&＃xff0c;概念覆盖全面。如果CQRS是一个城堡&＃xff0c;那么Diary.CQRS则是打开第一重门的钥匙&＃xff0c;接下来让我们一起推开这扇门吧。

Diary.CQRS.Web

运行项目&＃xff0c;最先看到的是一个Web页面&＃xff0c;如下图&＃xff1a;

很简单&＃xff0c;只有一个Add按钮&＃xff0c;当我们点击以后&＃xff0c;会进入添加的页面&＃xff1a;

我们填上一些内容&＃xff0c;然后点击Save按钮&＃xff0c;就会返回到列表页&＃xff0c;我们可以看到已添加的条目&＃xff1a;

然后我们进行编辑操作&＃xff0c;点击列表中的Edit按钮&＃xff0c;跳转到编辑页面&＃xff1a;

虽然页面中显示的是Add&＃xff0c;但确实是Edit页面。我们编辑以后点击Save按钮&＃xff0c;然后返回列表页即可看到编辑后的内容。

在列表页中&＃xff0c;如果我们点击Delete按钮&＃xff0c;则会删除改条目。

到此为止&＃xff0c;我们已经看到了这个项目的所有页面&＃xff0c;一个简单的CURD操作。我们继续看它的代码&＃xff08;在HomeController中&＃xff09;。

Index&＃xff1a;列表页面

public ActionResult Index()
{ViewBag.Model &＃61; ServiceLocator.ReportDatabase.GetItems();return View();
}

通过ServiceLocator定位ReportDatabase&＃xff0c;并从ReportDatabase中获取所有条目。

Add&＃xff1a;新增页面

public ActionResult Add()
{return View();
}[HttpPost]
public ActionResult Add(DiaryItemDto item)
{ServiceLocator.CommandBus.Send(new CreateItemCommand(Guid.NewGuid(), item.Title, item.Description, -1, item.From, item.To));return RedirectToAction("Index");
}

两个方法&＃xff1a;

• Add()方法&＃xff0c;处理Get请求&＃xff0c;返回新增视图&＃xff1b;
• Add(DiaryItemDto item)方法&＃xff0c;接收DiaryItemDto参数&＃xff0c;处理Post请求&＃xff0c;创建并发送CreateItemCommand命令&＃xff0c;然后返回到Index页面

Edit&＃xff1a;编辑页面

public ActionResult Edit(Guid id)
{var item &＃61; ServiceLocator.ReportDatabase.GetById(id);var model &＃61; new DiaryItemDto(){Description &＃61; item.Description,From &＃61; item.From,Id &＃61; item.Id,Title &＃61; item.Title,To &＃61; item.To,Version &＃61; item.Version};return View(model);
}[HttpPost]
public ActionResult Edit(DiaryItemDto item)
{ServiceLocator.CommandBus.Send(new ChangeItemCommand(item.Id, item.Title, item.Description, item.From, item.To, item.Version));return RedirectToAction("Index");
}

仍然是两个方法&＃xff1a;

• Edit(Guid id)方法&＃xff0c;接收Guid作为参数&＃xff0c;并从ReportDatabase中获取数据&＃xff0c;构建dto对象返回给页面
• Edit(DiaryItemDto item)方法&＃xff0c;接收DiaryItemDto对象&＃xff0c;处理Post请求&＃xff0c;接收到请求以后根据dto对象创建ChangeItemCommand命令&＃xff0c;然后返回到Index页面

Delete&＃xff1a;删除操作

public ActionResult Delete(Guid id)
{var item &＃61; ServiceLocator.ReportDatabase.GetById(id);ServiceLocator.CommandBus.Send(new DeleteItemCommand(item.Id, item.Version));return RedirectToAction("Index");
}

对于删除操作来说&＃xff0c;它没有视图页面&＃xff0c;接收到请求以后&＃xff0c;先获取该记录&＃xff0c;创建并发送DeleteImteCommand命令&＃xff0c;然后返回到Index页面

题外话&＃xff1a;对于改变数据状态的操作&＃xff0c;使用Get请求是不可取的&＃xff0c;可能存在安全隐患

通过上面的代码&＃xff0c;你会发现所有的操作都是从ServiceLocator发起的&＃xff0c;通过它我们能够定位到CommandBus和ReportDatabase&＃xff0c;从而进行相应的操作&＃xff0c;我们在接下来会介绍ServiceLocator类。

Diary.CQRS.Configuration

Diary.CQRS.Configuration 项目中定义了ServiceLocator类&＃xff0c;这个类的作用是完成IoC容器的服务注册、服务定位功能。例如我们可以通过ServiceLocator获取到CommandBus实例、获取ReportDatabase实例。

服务注册

ServiceLocator使用StructureMap作为依赖注入框架&＃xff0c;提供了服务注册、服务导航的功能。ServiceLocator类通过静态构造函数完成对服务注册和服务实例化工作&＃xff1a;

static ServiceLocator()
{if (!_isInitialized){lock (_lockThis){ContainerBootstrapper.BootstrapStructureMap();_commandBus &＃61; ObjectFactory.GetInstance();_reportDatabase &＃61; ObjectFactory.GetInstance();_isInitialized &＃61; true;}}
}

首先调用ContainerBootstrapper.BootstrapStructureMap()方法&＃xff0c;这个方法里面包含了对将服务添加到容器的代码&＃xff1b;然后使用容器创建CommandBus和ReportDatabase的实例。

• CommandBus&＃xff1a;命令总线&＃xff0c;对应Command操作&＃xff0c;用来发送命令&＃xff0c;程序中需要定义相应的命令处理器&＃xff0c;从而完成具体的操作。
• ReportDatabase&＃xff1a;报表数据库&＃xff0c;对应Query操作&＃xff0c;用来获取数据。

ServiceLocator的重要之处在于对外暴露了两个至关重要的实例&＃xff0c;分别处理CQRS中的Command和Query。

为什么没有Event相关操作呢&＃xff1f;到目前为止我们还没有涉及到&＃xff0c;因为对于UI层来说&＃xff0c;用户的意图都是通过Command表示的&＃xff0c;而数据的状态变化才会触发Event。

Diary.CQRS

在ServiceLocator中定义了获取CommandBus和ReportDatabase的方法&＃xff0c;我们顺着这两个对象继续分析。

CommandBus

在基于消息的系统设计中&＃xff0c;我们常会看到总线的身影&＃xff0c;Command也是一种消息&＃xff0c;所以使用总线是再合适不过的了。CommandBus就是我们在Diary.CQRS项目中用到的一种消息总线。

在Diary.CQRS中&＃xff0c;它被定义在Messaging目录&＃xff0c;在这个目录下面&＃xff0c;还有与Event相关的EventBus&＃xff0c;我们稍后再进行介绍。

CommandBus实现ICommandBus接口&＃xff0c;ICommandBus接口的定义如下&＃xff1a;

public interface ICommandBus
{void Send(T command) where T : Command;
}

它只包含了Send方法&＃xff0c;用来将命令发送到对应的处理程序。

CommandBus是ICommand的实现&＃xff0c;具体代码如下&＃xff1a;

public class CommandBus:ICommandBus
{private readonly ICommandHandlerFactory _commandHandlerFactory;public CommandBus(ICommandHandlerFactory commandHandlerFactory){_commandHandlerFactory &＃61; commandHandlerFactory;}public void Send(T command) where T : Command{var handler &＃61; _commandHandlerFactory.GetHandler();if (handler!&＃61;null){handler.Execute(command);}else{throw new Exception();}}
}

在CommandBus中&＃xff0c;显式依赖ICommandHandlerFactory类&＃xff0c;通过构造函数进行注入。那么 _commandHandlerFactory 的作用是什么呢&＃xff1f;我们在Send方法中可以看到&＃xff0c;通过 _commandHandlerFactory 可以获取到与Command对应的CommandHandler&＃xff08;命令处理程序&＃xff09;&＃xff0c;在程序的设计上&＃xff0c;每一个Command都会有一个对应的CommandHandler&＃xff0c;而手工判断类型、实例化处理程序显然不符合使用习惯&＃xff0c;此处采用工厂模式来获取命令处理程序。

当获取到与Command对应的CommandHandler后&＃xff0c;调用handler的Execute方法&＃xff0c;执行该命令。

截止目前为止&＃xff0c;我们又接触了三个概念&＃xff1a;CommandHandlerFactory、CommandHandler、Command&＃xff1a;

• CommandHandlerFactory&＃xff1a;命令处理程序工厂&＃xff0c;通过GetHandler方法获取到与命令对应的处理程序
• CommandHandler&＃xff1a;命令处理程序&＃xff0c;用于执行对应的命令
• Command&＃xff1a;命令&＃xff0c;描述用户的意图、并包含与意图相关的数据

CommandHandlerFactory

使用简单工厂模式&＃xff0c;用来获取与命令对应的处理程序。它的代码在Utils文件夹中&＃xff0c;它的作用是提供一种获取Handler的方式&＃xff0c;所以它只能作为工具存在。

接口定义如下&＃xff1a;

public interface ICommandHandlerFactory
{ICommandHandler GetHandler() where T : Command;
}

只有GetHandler一个方法&＃xff0c;它的实现是 StructureMapCommandHandlerFactory&＃xff0c;即通过StructureMap作为依赖注入框架来实现的&＃xff0c;代码也比较简单&＃xff0c;这里不再贴出来了。

Command和CommandHandler

命令是代表用户的意图、并包含与意图相关的数据&＃xff0c;比如用户想要添加一条数据&＃xff0c;这便是一个意图&＃xff0c;于是就有了CreateItemCommand&＃xff0c;用户要在界面上填写添加操作必须的数据&＃xff0c;于是就有了命令的属性。

关于命令的定义如下&＃xff1a;

public interface ICommand
{Guid Id { get; }
}public class Command : ICommand
{public Guid Id { get; private set; }public int Version { get; set; }public Command(Guid id, int version){Id &＃61; id;Version &＃61; version;}
}

• ICommand接口&＃xff1a;包含Id属性&＃xff0c;这个Id表示Command对应聚合的Id。聚合是领域驱动开发&＃xff08;DDD&＃xff09;的概念&＃xff0c;表示一组强关联的领域对象&＃xff0c;而对聚合中状态的变更&＃xff0c;只能通过聚合根&＃xff08;AggregateRoot&＃xff09;来完成。

• Command类&＃xff1a;实现了ICommand接口&＃xff0c;并增加了Version属性&＃xff0c;用来标记当前操作对应的聚合跟的版本。

  为什么要有版本的概念的&＃xff1f;因为当使用ES模式的时候&＃xff0c;数据库中的数据都是事件产生的数据镜像&＃xff0c;保存了某个时间点的数据快照&＃xff0c;如果要获取到最新的数据&＃xff0c;则需要通过加载该聚合根对应的所有Event来回放到最新状态。如果引入版本的概念&＃xff0c;每一个Event对应一个版本&＃xff0c;而景象中的数据也有一个版本&＃xff0c;在进行回放的时候&＃xff0c;可以仅加载高版本的Event进行回放&＃xff0c;节省了系统资源&＃xff0c;并提高了运行效率。

命令处理程序&＃xff0c;它的作用是处理与它相对应的命令&＃xff0c;处理CQRS的核心&＃xff0c;接口定义如下&＃xff1a;

public interface ICommandHandler where TCommand : Command
{void Execute(TCommand command);
}

它接收command作为参数&＃xff0c;执行该命令的处理逻辑。每一个命令都有一个与之对应的处理程序。

我们再重新梳理一下流程&＃xff0c;首先用户要新增一个数据&＃xff0c;点击保存按钮后&＃xff0c;生成CreateItemCommand命令&＃xff0c;随后这个命令被发送到CommandBus中&＃xff0c;CommandBus通过CommandHandlerFactory找到该Command的处理程序&＃xff0c;此时在CommandBus的Send方法中&＃xff0c;我们有一个Command和CommandHandler&＃xff0c;然后调用CommandHandler的Execute方法&＃xff0c;即完成了该方法的处理。至此&＃xff0c;Command的处理流程完结。

CreateItemCommand和CreateItemCommandHandler

我们来看一下CreateItemCommand的代码&＃xff1a;

public class CreateItemCommand : Command
{public string Title { get; internal set; }public string Description { get; internal set; }public DateTime From { get; internal set; }public DateTime To { get; internal set; }public CreateItemCommand(Guid aggregateId, string title,string description, int version, DateTime from, DateTime to): base(aggregateId, version){Title &＃61; title;Description &＃61; description;From &＃61; from;To &＃61; to;}
}

它继承自Command基类&＃xff0c;继承后即拥有了Id和Version属性&＃xff0c;然后又定义了几个其它的属性。它只包含数据&＃xff0c;与该命令对应的处理程序叫做CreateItemCommandHandler&＃xff0c;代码如下&＃xff1a;

public class CreateItemCommandHandler : ICommandHandler
{private IRepository _repository;public CreateItemCommandHandler(IRepository repository){_repository &＃61; repository;}public void Execute(CreateItemCommand command){if (command &＃61;&＃61; null){throw new Exception();}if (_repository &＃61;&＃61; null){throw new Exception();}var aggregate &＃61; new DiaryItem(command.Id, command.Title, command.Description, command.From, command.To);aggregate.Version &＃61; -1;_repository.Save(aggregate, aggregate.Version);}
}

这才是我们要分析的核心&＃xff0c;在Handler中&＃xff0c;我们看到了Repository&＃xff0c;看到了DiaryItem聚合&＃xff1a;

• IRepository&＃xff1a;仓储类&＃xff0c;代表数据的储存方式&＃xff0c;通过仓储能够进行数据操作
• DiaryItem&＃xff1a;领域对象&＃xff0c;聚合根&＃xff0c;所有数据状态的变更只能通过聚合根来修改

在上面的代码中&＃xff0c;由于是新增&＃xff0c;所以聚合的版本为-1&＃xff0c;然后调用仓储的Save方法进行保存。我们继续往下扒&＃xff0c;看看仓储和聚合的实现。

Repository

对于Repository的定义&＃xff0c;仍然先看一下接口中的定义&＃xff0c;代码如下&＃xff1a;

public interface IRepository where T : AggregateRoot, new()
{void Save(AggregateRoot aggregate, int expectedVersion);T GetById(Guid id);
}

在仓储中只有两个方法&＃xff1a;

• Save(AggregateRoot aggregate, int expectedVersion)&＃xff1a;保存期望版本的聚合根
• GetById(Guid id)&＃xff1a;根据聚合根Id获取聚合根

关于IRepository的实现&＃xff0c;代码在Repository.cs中&＃xff0c;我们拆开来进行介绍&＃xff1a;

private readonly IEventStorage _eventStorage;
private static object _lock &＃61; new object();public Repository(IEventStorage eventStorage)
{_eventStorage &＃61; eventStorage;
}

首先是它的构造函数&＃xff0c;强依赖IEventStorage&＃xff0c;通过构造函数注入。EventStorage是事件的储存仓库&＃xff0c;有个更为熟知的名字EventStore&＃xff0c;我们稍后进行介绍。

public T GetById(Guid id)
{IEnumerable events;var memento &＃61; _eventStorage.GetMemento(id);if (memento !&＃61; null){events &＃61; _eventStorage.GetEvents(id).Where(e &＃61;> e.Version >&＃61; memento.Version);}else{events &＃61; _eventStorage.GetEvents(id);}var obj &＃61; new T();if (memento !&＃61; null){((IOriginator)obj).SetMemento(memento);}obj.LoadsFromHistory(events);return obj;
}

GetById(Guid id)方法通过Id获取一个聚合对象&＃xff0c;获取一个聚合对象有以下几个步骤&＃xff1a;

• 首先会从EventStorage中获取到该聚合的快照&＃xff08;memento的翻译为记忆碎片、纪念品、备忘录&＃xff0c;用来聚合对象的快照&＃xff09;。

• 加载Event列表&＃xff0c;加载到的事件列表将用来做事件回放。

  如果获取到快照的话&＃xff0c;则加载版本高于该快照版本的事件列表&＃xff0c;如果没有获取到快照&＃xff0c;则加载全部事件列表。此处在上面已经介绍过&＃xff0c;通过快照的方式保存聚合对象&＃xff0c;在获取数据时可以减少重放事件的数量&＃xff0c;起到提高加载速度的作用。

• 实例化聚合根&＃xff0c;对应代码中的var obj &＃61; new T();。
• 从快照中设置聚合根的状态。在获取到快照以后&＃xff0c;如果快照不为空&＃xff0c;则调用聚合根的SetMemento方法设置为快照中的状态&＃xff0c;SetMemento方法定义在IOriginator接口中&＃xff0c;聚合根需要实现该接口。

• 加载历史事件&＃xff0c;完成重放。完成这个步骤以后&＃xff0c;聚合根将更新到最新状态。

通过这几个步骤以后&＃xff0c;我们得到了一个最新状态的聚合根对象。

public void Save(AggregateRoot aggregate, int expectedVersion)
{if (aggregate.GetUncommittedChanges().Any()){lock (_lock){var item &＃61; new T();if (expectedVersion !&＃61; -1){item &＃61; GetById(aggregate.Id);if (item.Version !&＃61; expectedVersion){throw new Exception();}}_eventStorage.Save(aggregate);}}
}

Save方法&＃xff0c;用来保存一个聚合根对象。在这个方法中&＃xff0c;参数expectedVersion表示期望的版本&＃xff0c;这里约定-1为新增的聚合根&＃xff0c;当聚合根为新增的时候&＃xff0c;会直接调用EventStorage中的Save方法。

关于expectedVersion参数&＃xff0c;我们可以理解为对并发的控制&＃xff0c;只有当expectedVersion与GetById获取到的聚合根对象的版本相同时才能进行保存操作。

在介绍Repository类的时候&＃xff0c;我们接触了两个新的概念&＃xff1a;EventStorage和AggregateRoot&＃xff0c;接下来我们分别进行介绍。

AggregateRoot

AggregateRoot是聚合根&＃xff0c;他表示一组强关联的领域对象&＃xff0c;所有对象的状态变更只能通过聚合根来完成&＃xff0c;这样可以保证数据的一致性&＃xff0c;以及减少并发冲突。应用到EventSourcing模式中&＃xff0c;聚合根的好处也是很明显的&＃xff0c;我们所有对数据状态的变更都通过聚合根完成&＃xff0c;而每次变更&＃xff0c;聚合根都会生成相应的事件&＃xff0c;在进行事件回放的时候&＃xff0c;又通过聚合根来完成历史事件的加载。由此我们可以看到&＃xff0c;聚合根对象应该具备生成事件、重放事件的能力。

我们来看看聚合根基类的定义&＃xff0c;在Domain文件夹中&＃xff1a;

public abstract class AggregateRoot : IEventProvider{// ......
}

首先这是一个抽象类&＃xff0c;实现了IEventProvider接口&＃xff0c;该接口的定义如下&＃xff1a;

public interface IEventProvider
{void LoadsFromHistory(IEnumerable history);IEnumerable GetUncommittedChanges();
}

它定义了两个方法&＃xff0c;我们分别进行说明&＃xff1a;

• LoadsFromHistory()方法&＃xff1a;加载历史事件&＃xff0c;还原聚合根的最新状态&＃xff0c;我们在Repository中已经用过这个方法。
• GetUncommittedChanges()方法&＃xff1a;获取未提交的事件。一个命令可能造成聚合根发生多次更改&＃xff0c;每次更改都会产生一个事件&＃xff0c;这些事件被暂时的保存在聚合根对象中&＃xff0c;通过该方法可以获取到未提交的事件列表。

为了实现这个接口&＃xff0c;聚合根中定义了 List _changes对象&＃xff0c;用来临时存储所有未提交的事件&＃xff0c;该对象在构造函数中进行初始化。

AggregateRoot中对于该事件的实现如下&＃xff1a;

public void LoadsFromHistory(IEnumerable history)
{foreach (var e in history){ApplyChange(e, false);}Version &＃61; history.Last().Version;EventVersion &＃61; Version;
}public IEnumerable GetUncommittedChanges()
{return _changes;
}

LoadsFromHistory方法遍历历史事件&＃xff0c;并调用ApplyChange方法更新聚合根的状态&＃xff0c;在完成更新后设置版本号为最后一个事件的版本。GetUncommittedChanges方法比较简单&＃xff0c;返回对象的_changes事件列表。

接下来我们看看ApplyChange方法&＃xff0c;该方法有两个实现&＃xff0c;代码如下&＃xff1a;

protected void ApplyChange(Event &＃64;event)
{ApplyChange(&＃64;event, true);
}protected void ApplyChange(Event &＃64;event, bool isNew)
{dynamic d &＃61; this;d.Handle(Converter.ChangeTo(&＃64;event, &＃64;event.GetType()));if (isNew){_changes.Add(&＃64;event);}
}

这两个方法定义为protected&＃xff0c;只能被子类访问。我们可以理解为&＃xff0c;ApplyChange(Event &＃64;event)方法为简化操作&＃xff0c;对第二个参数进行了默认为true的操作&＃xff0c;然后调用ApplyChange(Event &＃64;event, bool isNew)方法。

在ApplyChange(Event &＃64;event, bool isNew)方法中&＃xff0c;调用了聚合根的Handle方法&＃xff0c;用来处理事件。如果isNew参数为true&＃xff0c;则将事件添加到change列表中&＃xff0c;如果为false&＃xff0c;则认为是在进行事件回放&＃xff0c;所以不进行事件的添加。

需要注意的是&＃xff0c;聚合根的Handle方法&＃xff0c;与EventHandler不同&＃xff0c;当Event产生以后&＃xff0c;首先由它对应的聚合根进行处理&＃xff0c;因此聚合根要具备处理该事件的能力&＃xff0c;如何具备呢&＃xff1f;聚合根要实现IHandle接口&＃xff0c;该接口的定义如下&＃xff1a;

public interface IHandle where TEvent:Event
{void Handle(TEvent e);
}

这里可以看出&＃xff0c;IHandle接口是泛型的&＃xff0c;它只对一个具体的Event类型生效&＃xff0c;在代码上的体现如下&＃xff1a;

public class DiaryItem : AggregateRoot,IHandle,IHandle,IHandle,IHandle,IHandle,IOriginator
{//......
}

最后&＃xff0c;聚合根还定义了清除所有事件的方法&＃xff0c;代码如下&＃xff1a;

public void MarkChangesAsCommitted()
{_changes.Clear();
}

MarkChangesAsCommitted()方法用来清空事件列表。

Event

终于到我们今天的另外一个核心内容了&＃xff0c;Event是ES中的一等公民&＃xff0c;所有的状态变更最终都以Event的形式进行存储&＃xff0c;当我们要查看聚合根最新状态的时候&＃xff0c;可以通过事件回放来获取。我们来看看Event的定义&＃xff1a;

public interface IEvent
{Guid Id { get; }
}

IEvent接口定义了一个事件必须拥有唯一的Id进行标识。然后Event实现了IEvent接口&＃xff1a;

public class Event:IEvent
{public int Version;public Guid AggregateId { get; set; }public Guid Id { get; private set; }
}

可以看到&＃xff0c;除了Id属性外&＃xff0c;还添加了两个字段Version和AggregateId。AggregateId表示该事件关联的聚合根Id&＃xff0c;通过该Id可以获取到唯一的聚合根对象&＃xff1b;Version表示事件发生时该事件的版本&＃xff0c;每次产生新的事件&＃xff0c;Version都会进行累加。

从而可以知道&＃xff0c;在EventStorage中&＃xff0c;聚合根Id对应的所有Event中的Version是顺序累加的&＃xff0c;按照Version进行排序可以得到事件发生的先后顺序。

EventStorage

顾名思义&＃xff0c;EventStorage是用来存储Event的地方。在Diary.CQRS中&＃xff0c;EventStorage的定义如下&＃xff1a;

public interface IEventStorage
{IEnumerable GetEvents(Guid aggregateId);void Save(AggregateRoot aggregate);T GetMemento(Guid aggregateId) where T : BaseMemento;void SaveMemento(BaseMemento memento);
}

• GetEvents(Guid aggregateId)&＃xff1a;根据聚合根Id获取该聚合根的所有事件
• Save(AggregateRoot aggregate)&＃xff1a;保存方法&＃xff0c;入参为聚合根对象&＃xff0c;在实现上则是获取聚合根中所有未提交的事件&＃xff0c;随后对这些事件进行处理
• GetMemento()&＃xff1a;获取快照
• SaveMemento()&＃xff1a;存储快照

Diary.CQRS中使用InMemory的方式实现了EventStorage&＃xff0c;属性和构造函数如下&＃xff1a;

private List _events;
private List _mementoes;
private readonly IEventBus _eventBus;public InMemoryEventStorage(IEventBus eventBus)
{_events &＃61; new List();_mementoes &＃61; new List();_eventBus &＃61; eventBus;
}

• _events&＃xff1a;事件列表&＃xff0c;内存中存储事件的位置&＃xff0c;所有事件最终都会存储在该列表中
• _mementoes&＃xff1a;快照列表&＃xff0c;用于存储聚合根的某个事件版本的状态
• _eventBus&＃xff1a;事件总线&＃xff0c;用于发布任务

当Event生成后&＃xff0c;它并没有马上存入EventStorage&＃xff0c;而是在Repository显示调用Save方法时&＃xff0c;仓储将存储权交给了EventStorage&＃xff0c;EventStorage是事件仓库&＃xff0c;事件仓储在存储时进行了如下操作&＃xff1a;

• 获取聚合根中所有未提交的Event&＃xff0c;同时获取到聚合根当前的版本号
• 遍历未提交Event列表&＃xff0c;根据聚合根版本号自动为Event生成版本号&＃xff0c;保持自增长的特性&＃xff1b;
• 生成聚合根快照。示例中每3个版本生成一次&＃xff0c;并保持到事件仓储中。
• 将任务添加到事件仓库中。
• 再次遍历未提交Event列表&＃xff0c;此时将进行任务发布&＃xff0c;调用事件总线的Publish方法进行发布。

Save方法的代码如下&＃xff1a;

public void Save(AggregateRoot aggregate)
{var uncommittedChanges &＃61; aggregate.GetUncommittedChanges();var version &＃61; aggregate.Version;foreach (var &＃64;event in uncommittedChanges){version&＃43;&＃43;;if (version > 2){if (version % 3 &＃61;&＃61; 0){var originator &＃61; (IOriginator)aggregate;var memento &＃61; originator.GetMemento();memento.Version &＃61; version;SaveMemento(memento);}}&＃64;event.Version &＃61; version;_events.Add(&＃64;event);}foreach (var &＃64;event in uncommittedChanges){var desEvent &＃61; Converter.ChangeTo(&＃64;event, &＃64;event.GetType());_eventBus.Publish(desEvent);}
}

至此Event的处理流程就算完结了。此时所有的操作都是在主库完成的&＃xff0c;当事件被发布以后&＃xff0c;订阅了该事件的所有Handler都将会被触发。

在Diary.CQRS项目中&＃xff0c;EventHandler都被用来处理ReportDatabase了。

ReportDatabase

当你使用ES模式时&＃xff0c;都存在一个严重问题&＃xff0c;那就是数据查询的问题。当用户进行数据检索是&＃xff0c;必然会使用各种查询条件&＃xff0c;然而无论那种事件仓库都很难满足复杂查询。为了解决此问题&＃xff0c;ReportDatabase就显得格外重要。

ReportDatabase的作用被定义为获取数据、应对数据查询、生成报表等&＃xff0c;它的结构与主库不同&＃xff0c;可以根据不同的业务场景进行定义。

ReportDatabase的数据不是通过业务逻辑进行更新的&＃xff0c;它通过订阅Event进行更新。在本示例中ReportDatabase实现的很简单&＃xff0c;接口定义如下&＃xff1a;

public interface IReportDatabase
{DiaryItemDto GetById(Guid id);void Add(DiaryItemDto item);void Delete(Guid id);List GetItems();
}

实现上&＃xff0c;通过内存中维护一个列表&＃xff0c;每次接收到事件以后&＃xff0c;都对相应数据进行更新&＃xff0c;此处不在贴出。

EventHandler、EventHandlerFactory和EventBus

在上文中已经介绍过Event&＃xff0c;而针对Event的处理&＃xff0c;实现逻辑上与Command非常相似&＃xff0c;唯一的区别是&＃xff0c;命令只可以有一个对应的处理程序&＃xff0c;而事件则可以有多个处理程序。所以在EventHandlerFactory中获取处理程序的方法返回了EventHandler列表&＃xff0c;代码如下&＃xff1a;

public IEnumerable> GetHandlers() where T : Event
{var handlers &＃61; GetHandlerType();var lstHandlers &＃61; handlers.Select(handler &＃61;> (IEventHandler)ObjectFactory.GetInstance(handler)).ToList();return lstHandlers;
}

在EventBus中&＃xff0c;如果一个事件没有处理程序也不会引发错误&＃xff0c;如果有一个或多个处理程序&＃xff0c;则会以此调用他们的Handle方法&＃xff0c;代码如下&＃xff1a;

public void Publish(T &＃64;event) where T : Event
{var handlers &＃61; _eventHandlerFactory.GetHandlers();foreach (var eventHandler in handlers){eventHandler.Handle(&＃64;event);}
}

总结

Diary.CQRS是一个典型的CQRS&＃43;ES演示项目&＃xff0c;通过对该项目的分析&＃xff0c;我们能了解到Command、AggregateRoot、Event、EventStorage、ReportDatabase的基础知识&＃xff0c;了解他们相互关系&＃xff0c;尤其是如何进行事件存储、如何进行事件回放的内容。

另外&＃xff0c;我们发现在使用CQRS&＃43;ES的过程中&＃xff0c;项目的复杂度增加了很多&＃xff0c;我们不可避免的要使用EventStore、Messaging等架构&＃xff0c;从而影响那些不了解CQRS的团队成员的加入&＃xff0c;因此在应用到实际项目的时候&＃xff0c;要适可而止&＃xff0c;慎重选择&＃xff0c;避免过度设计。

由于这是一个示例&＃xff0c;项目代码中存在很多不够严谨的地方&＃xff0c;大家在学习的过程中应进行甄别。

由于本人的知识有限&＃xff0c;如果内容中存在不准确或错误的地方&＃xff0c;还请不吝赐教&＃xff01;