标签：tin 功能 聚合 接收 添加 静态 事件 delete 它的
在《当我们在讨论CQRS时，我们在讨论些神马》中，我们讨论了当使用CQRS的过程中，需要关心的一些问题。其中与CQRS关联最为紧密的模式莫过于Event Sourcing了，CQRS与ES的结合，为我们构造高性能、可扩展系统提供了基本思路。本文将介绍
Kanasz Robert在《Introduction to CQRS》中的示例项目Diary.CQRS。
该项目为Kanasz Robert为了介绍CQRS模式而写的一个测试项目，原始项目可以通过访问《Introduction to CQRS》来获取，由于项目版本比较旧，没有使用nuget管理程序包等，导致下载以后并不能正常运行，我下载了这个项目，升级到Visual Studio 2017，重新引用了StructMap框架（使用nuget），移除了Web层报错的代码，并上传到博客园，可以从这里下载：Diary.CQRS.rar
Diary.CQRS项目的场景为日记本管理，提供了新增、编辑、删除、列表等功能，整个解决方案分为三个项目：
这是一个很好的入门项目，功能简单、结构清晰，概念覆盖全面。如果CQRS是一个城堡，那么Diary.CQRS则是打开第一重门的钥匙，接下来让我们一起推开这扇门吧。
运行项目，最先看到的是一个Web页面，如下图：
很简单，只有一个Add按钮，当我们点击以后，会进入添加的页面：
我们填上一些内容，然后点击Save按钮，就会返回到列表页，我们可以看到已添加的条目：
然后我们进行编辑操作，点击列表中的Edit按钮，跳转到编辑页面：
虽然页面中显示的是Add，但确实是Edit页面。我们编辑以后点击Save按钮，然后返回列表页即可看到编辑后的内容。
在列表页中，如果我们点击Delete按钮，则会删除改条目。
到此为止，我们已经看到了这个项目的所有页面，一个简单的CURD操作。我们继续看它的代码（在HomeController中）。
public ActionResult Index()
{
ViewBag.Model = ServiceLocator.ReportDatabase.GetItems();
return View();
}通过ServiceLocator定位ReportDatabase，并从ReportDatabase中获取所有条目。
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");
}两个方法：
public ActionResult Edit(Guid id)
{
var item = ServiceLocator.ReportDatabase.GetById(id);
var model = new DiaryItemDto()
{
Description = item.Description,
From = item.From,
Id = item.Id,
Title = item.Title,
To = item.To,
Version = 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");
}仍然是两个方法：
public ActionResult Delete(Guid id)
{
var item = ServiceLocator.ReportDatabase.GetById(id);
ServiceLocator.CommandBus.Send(new DeleteItemCommand(item.Id, item.Version));
return RedirectToAction("Index");
}对于删除操作来说，它没有视图页面，接收到请求以后，先获取该记录，创建并发送DeleteImteCommand命令，然后返回到Index页面
题外话：对于改变数据状态的操作，使用Get请求是不可取的，可能存在安全隐患
通过上面的代码，你会发现所有的操作都是从ServiceLocator发起的，通过它我们能够定位到CommandBus和ReportDatabase，从而进行相应的操作，我们在接下来会介绍ServiceLocator类。
Diary.CQRS.Configuration 项目中定义了ServiceLocator类，这个类的作用是完成IoC容器的服务注册、服务定位功能。例如我们可以通过ServiceLocator获取到CommandBus实例、获取ReportDatabase实例。
ServiceLocator使用StructureMap作为依赖注入框架，提供了服务注册、服务导航的功能。ServiceLocator类通过静态构造函数完成对服务注册和服务实例化工作：
static ServiceLocator()
{
if (!_isInitialized)
{
lock (_lockThis)
{
ContainerBootstrapper.BootstrapStructureMap();
_commandBus = ObjectFactory.GetInstance<ICommandBus>();
_reportDatabase = ObjectFactory.GetInstance<IReportDatabase>();
_isInitialized = true;
}
}
}首先调用ContainerBootstrapper.BootstrapStructureMap()方法，这个方法里面包含了对将服务添加到容器的代码；然后使用容器创建CommandBus和ReportDatabase的实例。
ServiceLocator的重要之处在于对外暴露了两个至关重要的实例，分别处理CQRS中的Command和Query。
为什么没有Event相关操作呢？到目前为止我们还没有涉及到，因为对于UI层来说，用户的意图都是通过Command表示的，而数据的状态变化才会触发Event。
在ServiceLocator中定义了获取CommandBus和ReportDatabase的方法，我们顺着这两个对象继续分析。
在基于消息的系统设计中，我们常会看到总线的身影，Command也是一种消息，所以使用总线是再合适不过的了。CommandBus就是我们在Diary.CQRS项目中用到的一种消息总线。
在Diary.CQRS中，它被定义在Messaging目录，在这个目录下面，还有与Event相关的EventBus，我们稍后再进行介绍。
CommandBus实现ICommandBus接口，ICommandBus接口的定义如下：
public interface ICommandBus
{
void Send<T>(T command) where T : Command;
}它只包含了Send方法，用来将命令发送到对应的处理程序。
CommandBus是ICommand的实现，具体代码如下：
public class CommandBus:ICommandBus
{
private readonly ICommandHandlerFactory _commandHandlerFactory;
public CommandBus(ICommandHandlerFactory commandHandlerFactory)
{
_commandHandlerFactory = commandHandlerFactory;
}
public void Send<T>(T command) where T : Command
{
var handler = _commandHandlerFactory.GetHandler<T>();
if (handler!=null)
{
handler.Execute(command);
}
else
{
throw new Exception();
}
}
}在CommandBus中，显式依赖ICommandHandlerFactory类，通过构造函数进行注入。那么 _commandHandlerFactory 的作用是什么呢？我们在Send方法中可以看到，通过 _commandHandlerFactory 可以获取到与Command对应的CommandHandler（命令处理程序），在程序的设计上，每一个Command都会有一个对应的CommandHandler，而手工判断类型、实例化处理程序显然不符合使用习惯，此处采用工厂模式来获取命令处理程序。
当获取到与Command对应的CommandHandler后，调用handler的Execute方法，执行该命令。
截止目前为止，我们又接触了三个概念：CommandHandlerFactory、CommandHandler、Command：
使用简单工厂模式，用来获取与命令对应的处理程序。它的代码在Utils文件夹中，它的作用是提供一种获取Handler的方式，所以它只能作为工具存在。
接口定义如下：
public interface ICommandHandlerFactory
{
ICommandHandler<T> GetHandler<T>() where T : Command;
}只有GetHandler一个方法，它的实现是 StructureMapCommandHandlerFactory，即通过StructureMap作为依赖注入框架来实现的，代码也比较简单，这里不再贴出来了。
命令是代表用户的意图、并包含与意图相关的数据，比如用户想要添加一条数据，这便是一个意图，于是就有了CreateItemCommand，用户要在界面上填写添加操作必须的数据，于是就有了命令的属性。
关于命令的定义如下：
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 = id;
Version = version;
}
}Command类：实现了ICommand接口，并增加了Version属性，用来标记当前操作对应的聚合跟的版本。
为什么要有版本的概念的？因为当使用ES模式的时候，数据库中的数据都是事件产生的数据镜像，保存了某个时间点的数据快照，如果要获取到最新的数据，则需要通过加载该聚合根对应的所有Event来回放到最新状态。如果引入版本的概念，每一个Event对应一个版本，而景象中的数据也有一个版本，在进行回放的时候，可以仅加载高版本的Event进行回放，节省了系统资源，并提高了运行效率。
命令处理程序，它的作用是处理与它相对应的命令，处理CQRS的核心，接口定义如下：
public interface ICommandHandler<TCommand> where TCommand : Command
{
void Execute(TCommand command);
}它接收command作为参数，执行该命令的处理逻辑。每一个命令都有一个与之对应的处理程序。
我们再重新梳理一下流程，首先用户要新增一个数据，点击保存按钮后，生成CreateItemCommand命令，随后这个命令被发送到CommandBus中，CommandBus通过CommandHandlerFactory找到该Command的处理程序，此时在CommandBus的Send方法中，我们有一个Command和CommandHandler，然后调用CommandHandler的Execute方法，即完成了该方法的处理。至此，Command的处理流程完结。
我们来看一下CreateItemCommand的代码：
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 = title;
Description = description;
From = from;
To = to;
}
}它继承自Command基类，继承后即拥有了Id和Version属性，然后又定义了几个其它的属性。它只包含数据，与该命令对应的处理程序叫做CreateItemCommandHandler，代码如下：
public class CreateItemCommandHandler : ICommandHandler<CreateItemCommand>
{
private IRepository<DiaryItem> _repository;
public CreateItemCommandHandler(IRepository<DiaryItem> repository)
{
_repository = repository;
}
public void Execute(CreateItemCommand command)
{
if (command == null)
{
throw new Exception();
}
if (_repository == null)
{
throw new Exception();
}
var aggregate = new DiaryItem(command.Id, command.Title, command.Description, command.From, command.To);
aggregate.Version = -1;
_repository.Save(aggregate, aggregate.Version);
}
}这才是我们要分析的核心，在Handler中，我们看到了Repository，看到了DiaryItem聚合：
在上面的代码中，由于是新增，所以聚合的版本为-1，然后调用仓储的Save方法进行保存。我们继续往下扒，看看仓储和聚合的实现。
对于Repository的定义，仍然先看一下接口中的定义，代码如下：
public interface IRepository<T> where T : AggregateRoot, new()
{
void Save(AggregateRoot aggregate, int expectedVersion);
T GetById(Guid id);
}在仓储中只有两个方法：
关于IRepository的实现，代码在Repository.cs中，我们拆开来进行介绍：
private readonly IEventStorage _eventStorage;
private static object _lock = new object();
public Repository(IEventStorage eventStorage)
{
_eventStorage = eventStorage;
}首先是它的构造函数，强依赖IEventStorage，通过构造函数注入。EventStorage是事件的储存仓库，有个更为熟知的名字EventStore，我们稍后进行介绍。
public T GetById(Guid id)
{
IEnumerable<Event> events;
var memento = _eventStorage.GetMemento<BaseMemento>(id);
if (memento != null)
{
events = _eventStorage.GetEvents(id).Where(e => e.Version >= memento.Version);
}
else
{
events = _eventStorage.GetEvents(id);
}
var obj = new T();
if (memento != null)
{
((IOriginator)obj).SetMemento(memento);
}
obj.LoadsFromHistory(events);
return obj;
}GetById(Guid id)方法通过Id获取一个聚合对象，获取一个聚合对象有以下几个步骤：
加载Event列表，加载到的事件列表将用来做事件回放。
如果获取到快照的话，则加载版本高于该快照版本的事件列表，如果没有获取到快照，则加载全部事件列表。此处在上面已经介绍过，通过快照的方式保存聚合对象，在获取数据时可以减少重放事件的数量，起到提高加载速度的作用。
var obj = new T();。加载历史事件，完成重放。完成这个步骤以后，聚合根将更新到最新状态。
通过这几个步骤以后，我们得到了一个最新状态的聚合根对象。
public void Save(AggregateRoot aggregate, int expectedVersion)
{
if (aggregate.GetUncommittedChanges().Any())
{
lock (_lock)
{
var item = new T();
if (expectedVersion != -1)
{
item = GetById(aggregate.Id);
if (item.Version != expectedVersion)
{
throw new Exception();
}
}
_eventStorage.Save(aggregate);
}
}
}Save方法，用来保存一个聚合根对象。在这个方法中，参数expectedVersion表示期望的版本，这里约定-1为新增的聚合根，当聚合根为新增的时候，会直接调用EventStorage中的Save方法。
关于expectedVersion参数，我们可以理解为对并发的控制，只有当expectedVersion与GetById获取到的聚合根对象的版本相同时才能进行保存操作。
在介绍Repository类的时候，我们接触了两个新的概念：EventStorage和AggregateRoot，接下来我们分别进行介绍。
AggregateRoot是聚合根，他表示一组强关联的领域对象，所有对象的状态变更只能通过聚合根来完成，这样可以保证数据的一致性，以及减少并发冲突。应用到EventSourcing模式中，聚合根的好处也是很明显的，我们所有对数据状态的变更都通过聚合根完成，而每次变更，聚合根都会生成相应的事件，在进行事件回放的时候，又通过聚合根来完成历史事件的加载。由此我们可以看到，聚合根对象应该具备生成事件、重放事件的能力。
我们来看看聚合根基类的定义，在Domain文件夹中：
public abstract class AggregateRoot : IEventProvider{
// ......
}首先这是一个抽象类，实现了IEventProvider接口，该接口的定义如下：
public interface IEventProvider
{
void LoadsFromHistory(IEnumerable<Event> history);
IEnumerable<Event> GetUncommittedChanges();
}它定义了两个方法，我们分别进行说明：
为了实现这个接口，聚合根中定义了 List<Event> _changes对象，用来临时存储所有未提交的事件，该对象在构造函数中进行初始化。
AggregateRoot中对于该事件的实现如下：
public void LoadsFromHistory(IEnumerable<Event> history)
{
foreach (var e in history)
{
ApplyChange(e, false);
}
Version = history.Last().Version;
EventVersion = Version;
}
public IEnumerable<Event> GetUncommittedChanges()
{
return _changes;
}LoadsFromHistory方法遍历历史事件，并调用ApplyChange方法更新聚合根的状态，在完成更新后设置版本号为最后一个事件的版本。GetUncommittedChanges方法比较简单，返回对象的_changes事件列表。
接下来我们看看ApplyChange方法，该方法有两个实现，代码如下：
protected void ApplyChange(Event @event)
{
ApplyChange(@event, true);
}
protected void ApplyChange(Event @event, bool isNew)
{
dynamic d = this;
d.Handle(Converter.ChangeTo(@event, @event.GetType()));
if (isNew)
{
_changes.Add(@event);
}
}这两个方法定义为protected，只能被子类访问。我们可以理解为，ApplyChange(Event @event)方法为简化操作，对第二个参数进行了默认为true的操作，然后调用ApplyChange(Event @event, bool isNew)方法。
在ApplyChange(Event @event, bool isNew)方法中，调用了聚合根的Handle方法，用来处理事件。如果isNew参数为true，则将事件添加到change列表中，如果为false，则认为是在进行事件回放，所以不进行事件的添加。
需要注意的是，聚合根的Handle方法，与EventHandler不同，当Event产生以后，首先由它对应的聚合根进行处理，因此聚合根要具备处理该事件的能力，如何具备呢？聚合根要实现IHandle接口，该接口的定义如下：
public interface IHandle<TEvent> where TEvent:Event
{
void Handle(TEvent e);
}这里可以看出，IHandle接口是泛型的，它只对一个具体的Event类型生效，在代码上的体现如下：
public class DiaryItem : AggregateRoot,
IHandle<ItemCreatedEvent>,
IHandle<ItemRenamedEvent>,
IHandle<ItemFromChangedEvent>,
IHandle<ItemToChangedEvent>,
IHandle<ItemDescriptionChangedEvent>,
IOriginator
{
//......
}最后，聚合根还定义了清除所有事件的方法，代码如下：
public void MarkChangesAsCommitted()
{
_changes.Clear();
}MarkChangesAsCommitted()方法用来清空事件列表。
终于到我们今天的另外一个核心内容了，Event是ES中的一等公民，所有的状态变更最终都以Event的形式进行存储，当我们要查看聚合根最新状态的时候，可以通过事件回放来获取。我们来看看Event的定义：
public interface IEvent
{
Guid Id { get; }
}IEvent接口定义了一个事件必须拥有唯一的Id进行标识。然后Event实现了IEvent接口：
public class Event:IEvent
{
public int Version;
public Guid AggregateId { get; set; }
public Guid Id { get; private set; }
}可以看到，除了Id属性外，还添加了两个字段Version和AggregateId。AggregateId表示该事件关联的聚合根Id，通过该Id可以获取到唯一的聚合根对象；Version表示事件发生时该事件的版本，每次产生新的事件，Version都会进行累加。
从而可以知道，在EventStorage中，聚合根Id对应的所有Event中的Version是顺序累加的，按照Version进行排序可以得到事件发生的先后顺序。
顾名思义，EventStorage是用来存储Event的地方。在Diary.CQRS中，EventStorage的定义如下：
public interface IEventStorage
{
IEnumerable<Event> GetEvents(Guid aggregateId);
void Save(AggregateRoot aggregate);
T GetMemento<T>(Guid aggregateId) where T : BaseMemento;
void SaveMemento(BaseMemento memento);
}Diary.CQRS中使用InMemory的方式实现了EventStorage，属性和构造函数如下：
private List<Event> _events;
private List<BaseMemento> _mementoes;
private readonly IEventBus _eventBus;
public InMemoryEventStorage(IEventBus eventBus)
{
_events = new List<Event>();
_mementoes = new List<BaseMemento>();
_eventBus = eventBus;
}当Event生成后，它并没有马上存入EventStorage，而是在Repository显示调用Save方法时，仓储将存储权交给了EventStorage，EventStorage是事件仓库，事件仓储在存储时进行了如下操作：
Save方法的代码如下：
public void Save(AggregateRoot aggregate)
{
var uncommittedChanges = aggregate.GetUncommittedChanges();
var version = aggregate.Version;
foreach (var @event in uncommittedChanges)
{
version++;
if (version > 2)
{
if (version % 3 == 0)
{
var originator = (IOriginator)aggregate;
var memento = originator.GetMemento();
memento.Version = version;
SaveMemento(memento);
}
}
@event.Version = version;
_events.Add(@event);
}
foreach (var @event in uncommittedChanges)
{
var desEvent = Converter.ChangeTo(@event, @event.GetType());
_eventBus.Publish(desEvent);
}
}至此Event的处理流程就算完结了。此时所有的操作都是在主库完成的，当事件被发布以后，订阅了该事件的所有Handler都将会被触发。
在Diary.CQRS项目中，EventHandler都被用来处理ReportDatabase了。
当你使用ES模式时，都存在一个严重问题，那就是数据查询的问题。当用户进行数据检索是，必然会使用各种查询条件，然而无论那种事件仓库都很难满足复杂查询。为了解决此问题，ReportDatabase就显得格外重要。
ReportDatabase的作用被定义为获取数据、应对数据查询、生成报表等，它的结构与主库不同，可以根据不同的业务场景进行定义。
ReportDatabase的数据不是通过业务逻辑进行更新的，它通过订阅Event进行更新。在本示例中ReportDatabase实现的很简单，接口定义如下：
public interface IReportDatabase
{
DiaryItemDto GetById(Guid id);
void Add(DiaryItemDto item);
void Delete(Guid id);
List<DiaryItemDto> GetItems();
}实现上，通过内存中维护一个列表，每次接收到事件以后，都对相应数据进行更新，此处不在贴出。
在上文中已经介绍过Event，而针对Event的处理，实现逻辑上与Command非常相似，唯一的区别是，命令只可以有一个对应的处理程序，而事件则可以有多个处理程序。所以在EventHandlerFactory中获取处理程序的方法返回了EventHandler列表，代码如下：
public IEnumerable<IEventHandler<T>> GetHandlers<T>() where T : Event
{
var handlers = GetHandlerType<T>();
var lstHandlers = handlers.Select(handler => (IEventHandler<T>)ObjectFactory.GetInstance(handler)).ToList();
return lstHandlers;
}在EventBus中，如果一个事件没有处理程序也不会引发错误，如果有一个或多个处理程序，则会以此调用他们的Handle方法，代码如下：
public void Publish<T>(T @event) where T : Event
{
var handlers = _eventHandlerFactory.GetHandlers<T>();
foreach (var eventHandler in handlers)
{
eventHandler.Handle(@event);
}
}Diary.CQRS是一个典型的CQRS+ES演示项目，通过对该项目的分析，我们能了解到Command、AggregateRoot、Event、EventStorage、ReportDatabase的基础知识，了解他们相互关系，尤其是如何进行事件存储、如何进行事件回放的内容。
另外，我们发现在使用CQRS+ES的过程中，项目的复杂度增加了很多，我们不可避免的要使用EventStore、Messaging等架构，从而影响那些不了解CQRS的团队成员的加入，因此在应用到实际项目的时候，要适可而止，慎重选择，避免过度设计。
由于这是一个示例，项目代码中存在很多不够严谨的地方，大家在学习的过程中应进行甄别。
由于本人的知识有限，如果内容中存在不准确或错误的地方，还请不吝赐教！
标签：tin 功能 聚合 接收 添加 静态 事件 delete 它的
原文地址：https://www.cnblogs.com/youring2/p/11074989.html
