ENode 1.0 - 整體架構介紹

前言

今天是個開心的日子，又是周末，可以安心輕松的寫寫文章了。經過了大概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框架，框架的特色，以及使用的前提條件。

框架簡介

1. 框架名稱：enode
2. 框架特色：提供一個基於DDD設計思想，實現了CQRS + EDA + Event Sourcing + In Memory這些架構模式的，支持負載均衡的，輕量級應用開發框架。
3. 開源地址：https://github.com/tangxuehua/enode
4. 完整使用enode的一個論壇的地址：https://github.com/tangxuehua/forum
5. nuget包Id：enode

使用該框架需要了解或遵守以下幾個約定：

1. 一個command只允許導致一個聚合根的修改或一個聚合根的創建，如果違反這個規則，則框架不允許；
2. 如果一個用戶操作會涉及多個聚合根的修改，則需要通過saga （process manager）來實現；擁抱最終一致性，簡單的說就是通過將command+domain event不斷的串聯來最終實現最終一致性；如果想徹底的知道enode哪里與眾不同，可以看一下源代碼中的BankTransferExample，相信這個會讓你明白什么是我所說的事件驅動設計；
3. 框架的核心編程思想是異步消息處理加最終一致性，所以，如果你想實現強一致性需求，那這個框架不太適合，至少目前沒有提供這樣的支持；
4. 框架的設計目標不是針對企業應用開發，傳統企業應用一般訪問量不大且要求強一致性事務；enode框架更多的是針對互聯網應用，特別是為一些需要支持訪問量大、高性能、可伸縮且允許最終一致性的互聯網站點提供支持；看過：可伸縮性最佳實踐：來自eBay的經驗的人應該知道要實現一個可伸縮的互聯網應用，異步編程和最終一致性是必須的；另外，因為如果數據量一大，那我們一般會把數據分開存放，這就意味着，如果你還想實現強一致性，那就要靠分布式事務。但是很不幸，分布式事務的成本代價太高。伸縮、性能和響應延遲都受到分布式事務協調成本的反面影響，隨着依賴的資源數量和用戶訪問數量的上升，這些指標都會以幾何級數惡化。可用性亦受到限制，因為所有依賴的資源都必須就位。
5. 框架定位：目前定位於單台機器上運行的單個應用內的CQRS架構前提下的command端的實現；如果要實現多台機器多個應用之間的分布式集成，則大家需要再進一步借助ESB來與更高層的SOA架構集成；

框架架構圖：

CQRS架構圖

上面的架構圖是enode框架的內部實現架構。當然，上面這個架構圖並不是完整的CQRS架構圖，而是CQRS架構圖中command端的實現架構。完整的CQRS架構圖一般如下：

從上圖我們可以看到，傳統的CQRS架構圖，一般畫的都比大范圍，command端具體如何實現，實現方案有很多種。而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框架的架構圖以及上面的文字描述說明，我們在看一下上面最開始框架簡介中提到的框架所使用的關鍵技術。

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個我覺得比較重要的問題，我會在接下來的文章中，一一討論這些問題的解決思路。我覺得寫這種介紹框架的文章，一方面要介紹框架本身，更重要的是要告訴別人你設計以及實現框架時遇到的問題以及解決思路。要把這個分析和解決的思路寫出來，這才是對讀者意義最大的；