分享一个基于DDD以及事件驱动架构（EDA）的应用开发框架enode

原帖地址：http://www.cnblogs.com/netfocus/archive/2013/06/17/3139661.html

前言

今天是个开心的日子，又是周末，可以安心轻松的写写文章了。经过了大概3年的DDD理论积累，以及去年年初的第一个版本的event sourcing框架的开发以及项目实践经验，再通过今年上半年利用业余时间的设计与开发，我的enode框架终于可以和大家见面了。

自从Eric Evan提出DDD领域驱动设计以来已经过了很多年了，现在已经有很多人在学习或实践DDD。但是我发现目前能够支持DDD开发的框架还不多，至少在国内还不多。据我所知道的java和.net平台，国外比较有名的有：基于java平台的是axon framework，该框架很活跃，作者也很勤奋，该框架已经在一些实际商业项目中使用了，算比较成功；基于.net平台的是ncqrs，该框架早起比较活跃，但现在没有发展了，因为几乎没人在维护，让人很失望；国内有：banq的jdon framework可以支持DDD+CQRS+EventSourcing的开发，但是它是基于java平台的，所以对于.net平台的人，没什么实际用处；.net平台，开源的主要就是园子里的晴阳兄开发的apworks框架。晴阳兄在DDD方面，在国内的贡献很大，写了很多DDD系列的文章，框架和案例并行，很不错。当然，我所关注的紧紧是c#和java语言的框架，基于scala等其他语言实现的框架也有很多，这里就不一一例举了。

上面这么多框架都有各自的特点和优势，这里就不多做评价了，大家有兴趣的自己去看看吧。我重点想介绍的是我的enode框架，框架的特色，以及使用的前提条件。

enode框架简介

1. 框架名称：enode


2. 框架特色：提供一个基于DDD设计思想，实现了CQRS + EDA + Event Sourcing + In Memory这些架构模式的，支持负载均衡的，轻量级应用开发框架。


3. 开源地址：https://github.com/tangxuehua/enode


4. nuget包Id：enode

使用该框架前需要了解或遵守以下几个约定：

1. 一个command只允许导致一个聚合根的修改或一个聚合根的创建，如果违反这个规则，则框架不允许；


2. 如果一个用户操作会涉及多个聚合根的修改，则需要通过saga （process manager）来实现；拥抱最终一致性，简单的说就是通过将command+domain event不断的串联来最终实现最终一致性；如果想彻底的知道enode哪里与众不同，可以看一下源代码中的BankTransferSagaExample，相信这个会让你明白什么是我所说的事件驱动设计；


3. 框架的核心编程思想是异步消息处理加最终一致性，所以，如果你想实现强一致性需求，那这个框架不太适合，至少目前没有提供这样的支持；


4. 框架的设计目标不是针对企业应用开发，传统企业应用一般访问量不大且要求强一致性事务；enode框架更多的是针对互联网应用，特别是为一些需要支持访问量大、高性能、可伸缩且允许最终一致性的互联网站点提供支持；看过：可伸缩性最佳实践：来自eBay的经验的人应该知道要实现一个可伸缩的互联网应用，异步编程和最终一致性是必须的；另外，因为如果数据量一大，那我们一般会把数据分开存放，这就意味着，如果你还想实现强一致性，那就要靠分布式事务。但是很不幸，分布式事务的成本代价太高。伸缩、性能和响应延迟都受到分布式事务协调成本的反面影响，随着依赖的资源数量和用户访问数量的上升，这些指标都会以几何级数恶化。可用性亦受到限制，因为所有依赖的资源都必须就位。


5. 框架定位：目前定位于单台机器上运行的单个应用内的CQRS架构前提下的command端的实现；如果要实现多台机器多个应用之间的分布式集成，则大家需要再进一步借助ESB来与更高层的SOA架构集成；

enode框架架构图：

CQRS架构图

上面的架构图是enode框架的内部实现架构。当然，上面这个架构图并不是完整的CQRS架构图，而是CQRS架构图中command端的实现架构。完整的CQRS架构图一般如下：

从上图我们可以看到，传统的CQRS架构图，一般画的都比大范围，command端具体如何实现，实现方案有很多种。而enode框架，只是其中一种实现。

enode框架的内部实现说明

1. 首先，client会发送command给command service，command service接受到command后，会通过一个command queue router来路由该command应该放到哪个command queue，每个command queue就是一个消息队列，队列里存放command。该消息队列是本地队列，但是支持消息的持久化，也就是说command被放入队列后，就算机器挂了，下次机器重启后，消息也不会丢失。另外，command queue我们可以根据需要配置多个，上图为了示意，只画了两个；


2. command queue的出口端，有一个command processor，command processor的职责是处理command。但是command processor本身不直接处理command，直接处理command的是command processor内部的一些worker线程，每个worker线程会不断的从command queue中取出command，然后按照图中标出的5个步骤对command进行处理。可以看出，由于command processor中的worker线程都是在并行工作的，所以我们可以发现，同一时刻，会有多个command在被同时处理。为什么要这样做？因为client发送command到command queue的速度很快，比如每秒发送1W个command过来，也就是并发是1W，但是command processor如果内部只有单线程在处理command，那速度跟不上这个并发量，所以我们需要设计支持多个worker同时处理command，这样延迟就会降低；我们从架构图可以看到，command processor获取聚合根是从内存缓存（如支持分布式缓存的redis）获取，性能比较高；持久化事件，用的是MongoDB，由于mongoDB性能也很高；如果觉得事件持久化到单台MongoDB server还是有瓶颈问题，那我们可以对MongoDB server做集群，然后对事件进行sharding，将不同的event存储到不同的MongoBD Server，这样，事件的持久化也不会成为瓶颈；这样，整个command processor的处理性能理论上可以很高，当然我还没测试过集群情况下性能可以达到多少；单个mongodb server，持久化事件的性能，5K不成问题；这里有一点借此在说明下，被持久化的其实不是单个事件，而是一个事件流，即EventStream。为什么是事件流是因为单个聚合根一次可能产生不止一个领域事件，但是这些事件比如一起被持久化，所以设计思路是把这些事件设计为一个事件流，然后将这个事件流作为一条mongodb的记录插入到mongodb；事件流在mongodb中的主键是聚合根ID+事件流的版本号，通过这两个联合字段作为主键，用来实现乐观锁；假如有两个事件流都是针对同一个聚合根的，且他们的版本号相同，那插入到mongodb时，会报主键索引冲突，这就是并发冲突了。需要对command进行自动重试（enode框架会帮你自动做掉这个自动重试）来解决这个问题；


3. command processor中的worker处理完一个command后，会把产生的事件发布给一个合适的event queue。同样，内部也会有一个event queue router来路由到底该放到哪个event queue。那么event queue中的事件接下来要被如何处理呢？也就是event processor会做身事情呢？很简单，就是分发事件给所有的事件订阅者，即dispatch event to subscribers。那这些event subscribers都会做什么事情呢？一般是做两种处理：1）因为是采用CQRS架构，所以我们不能仅仅持久化领域事件，还要通过领域事件来更新CQRS的查询端数据库（这种为了更新查询库的事件订阅者老外一般叫做denormalizer）；由于更新查询库没有必要同步，所以设计event queue；2）上面提到过，有些操作会影响多个聚合根，比如银行转账，订单处理，等。这些操作本质上是一个流程，所以我们的方案是通过在领域事件的event handler中发送command来异步的实现串联整个处理流程；当然，如何实现这个流程，还是有很多问题需要讨论。我个人觉得比较靠谱的方案是通过process manager，类似BPM的思想，国外也有很多人把它叫做saga。对saga或process manager感兴趣的看官，可以看看微软的这个例子：http://msdn.microsoft.com/en-us/library/jj591569.aspx，对于如何用enode来实现一个process manager，由于信息太多，所以我接下来会写一篇文章专门系统的介绍。

回顾enode框架所使用的关键技术

基于整个enode框架的架构图以及上面的文字描述说明，我们在看一下上面最开始框架简介中提到的框架所使用的关键技术。

1. DDD：指架构图中的domain model，采用DDD的思想去分析设计实现，enode框架会提供实现DDD所必要的基类聚合根以及触发领域事件的支持；


2. CQRS：指整个enode架构实现的是CQRS架构中的command端，CQRS架构的查询端，enode框架没做任何限制，我们可以随意设计；


3. EDA：指整个编程模型的思路，都要基于事件驱动的思想，也就是领域模型的状态更改是基于响应事件的，聚合根之间的交互，也不是基于事务，而是基于事件驱动和响应；


4. Event Sourcing：中文意思是事件溯源，关于什么是事件溯源，可以看一下这篇文章。通过事件溯源，我们可以不用ORM来持久化聚合根，而是只要持久化领域事件即可，当我们要还原聚合根时只要对该聚合根进行一次事件溯源即可；


5. In Memory：是指整个domain model的所有数据都存储在内存缓存中，比如分布式缓存redis中，且缓存永远不会被释放。这样当我们要获取聚合根时，只要从内存缓存拿即可，所以叫in memory；


6. NoSQL：是指enode用到了redis，mongodb这样的nosql产品；


7. 负载均衡支持：是指，基于enode框架的应用程序，可以方便的支持负载均衡；因为应用程序本身是无状态的，in memory是存储在全局的redis分布式缓存中，独立于应用本身；而event store则是用MongoDB，同样也是全局的，且也支持集群。所以，我们可以将基于enode框架开发的应用程序部署任意多份在不同的机器，然后做负载均衡，从而让我们的应用程序支撑更高的并发访问。

框架API使用简介

框架初始化

public void Initialize()
{
var connectionString = "mongodb://localhost/EventDB";
var eventCollection = "Event";
var eventPublishInfoCollection = "EventPublishInfo";
var eventHandleInfoCollection = "EventHandleInfo";

var assemblies = new Assembly[] { Assembly.GetExecutingAssembly() };

    Configuration
        .Create()
        .UseTinyObjectContainer()
        .UseLog4Net("log4net.config")
        .UseDefaultCommandHandlerProvider(assemblies)
        .UseDefaultAggregateRootTypeProvider(assemblies)
        .UseDefaultAggregateRootInternalHandlerProvider(assemblies)
        .UseDefaultEventHandlerProvider(assemblies)

//使用MongoDB来支持持久化
        .UseDefaultEventCollectionNameProvider(eventCollection)
        .UseDefaultQueueCollectionNameProvider()
        .UseMongoMessageStore(connectionString)
        .UseMongoEventStore(connectionString)
        .UseMongoEventPublishInfoStore(connectionString, eventPublishInfoCollection)
        .UseMongoEventHandleInfoStore(connectionString, eventHandleInfoCollection)

        .UseAllDefaultProcessors(
new string[] { "CommandQueue" },
"RetryCommandQueue",
new string[] { "EventQueue" })
        .Start();
}

command定义

[Serializable]
public class ChangeNoteTitle : Command
{
public Guid NoteId { get; set; }
public string Title { get; set; }
}

发送command到ICommandService

var commandService = ObjectContainer.Resolve<ICommandService>();
commandService.Send(new ChangeNoteTitle { NoteId = noteId, Title = "Modified Note" });

Command Handler

public class ChangeNoteTitleCommandHandler : ICommandHandler<ChangeNoteTitle>
{
public void Handle(ICommandContext context, ChangeNoteTitle command)
    {
        context.Get<Note>(command.NoteId).ChangeTitle(command.Title);
    }
}

Domain Model

[Serializable]
public class Note : AggregateRoot<Guid>,
    IEventHandler<NoteCreated>,
    IEventHandler<NoteTitleChanged>
{
public string Title { get; private set; }
public DateTime CreatedTime { get; private set; }
public DateTime UpdatedTime { get; private set; }

public Note() : base() { }
public Note(Guid id, string title) : base(id)
    {
var currentTime = DateTime.Now;
        RaiseEvent(new NoteCreated(Id, title, currentTime, currentTime));
    }

public void ChangeTitle(string title)
    {
        RaiseEvent(new NoteTitleChanged(Id, title, DateTime.Now));
    }

void IEventHandler<NoteCreated>.Handle(NoteCreated evnt)
    {
        Title = evnt.Title;
        CreatedTime = evnt.CreatedTime;
        UpdatedTime = evnt.UpdatedTime;
    }
void IEventHandler<NoteTitleChanged>.Handle(NoteTitleChanged evnt)
    {
        Title = evnt.Title;
        UpdatedTime = evnt.UpdatedTime;
    }
}

 Domain Event

[Serializable]
public class NoteTitleChanged : Event
{
public Guid NoteId { get; private set; }
public string Title { get; private set; }
public DateTime UpdatedTime { get; private set; }

public NoteTitleChanged(Guid noteId, string title, DateTime updatedTime)
    {
        NoteId = noteId;
        Title = title;
        UpdatedTime = updatedTime;
    }
}

Event Handler

public class NoteEventHandler :
    IEventHandler<NoteCreated>,
    IEventHandler<NoteTitleChanged>
{
public void Handle(NoteCreated evnt)
    {
        Console.WriteLine(string.Format("Note created, title：{0}", evnt.Title));
    }
public void Handle(NoteTitleChanged evnt)
    {
        Console.WriteLine(string.Format("Note title changed, title：{0}", evnt.Title));
    }
}

后续需要讨论的关键问题

1. 既然是消息驱动，那如何保证消息不会丢失；


2. 如何保证消息至少被执行一次，且不能被重复执行；


3. 如何确保消息没执行成功就不能丢，也就是要求消息队列支持事务；


4. 因为是多线程并行持久化事件并且是多台机器集群负载均衡部署的，那如何保证领域事件被持久化的顺序与发布到事件订阅者的顺序完全一致；


5. 整个架构中，基于redis实现的memory cache以及基于mongodb实现的eventstore，是两个关键的存储点，如何确保高吞吐量和可用性；


6. 因为事件是并行持久化的，那如果遇到并发冲突如何解决？


7. 命令的重试如何实现？消息队列中的消息的重试机制如何实现？


8. 既然抛弃了强一致性的事务概念，而用process manager来实现聚合根交互，那如何具体实现一个process manager？

目前暂时想到以上8个我觉得比较重要的问题，我会在接下来的文章中，一一讨论这些问题的解决思路。我觉得写这种介绍框架的文章，一方面要介绍框架本身，更重要的是要告诉别人你设计以及实现框架时遇到的问题以及解决思路。要把这个分析和解决的思路写出来，这才是对读者意义最大的；

本文链接