导航：首页 > 开发技术 >

使用ENode 2.0举例分析

发表于：2025-02-06 作者：千家信息网编辑

千家信息网最后更新 2025年02月06日，本篇内容介绍了"使用ENode 2.0举例分析"的有关知识，在实际案例的操作过程中，不少人都会遇到这样的困境，接下来就让小编带领大家学习一下如何处理这些情况吧！希望大家仔细阅读，能够学有所成！ENod

千家信息网最后更新 2025年02月06日使用ENode 2.0举例分析

本篇内容介绍了"使用ENode 2.0举例分析"的有关知识，在实际案例的操作过程中，不少人都会遇到这样的困境，接下来就让小编带领大家学习一下如何处理这些情况吧！希望大家仔细阅读，能够学有所成！

ENode, EQueue, Forum 开源项目地址

ENode开源地址：https://github.com/tangxuehua/enode
EQueue开源地址：https://github.com/tangxuehua/equeue
ECommon开源地址：https://github.com/tangxuehua/ecommon
Forum开源地址：https://github.com/tangxuehua/forum
Forum论坛线上地址（临时域名，以后会改为enode.me）：http://enode.cloudapp.net
Forum论坛的equeue消息数据监控统计页面：http://enode.cloudapp.net/equeueadmin

另外，项目中如果要开发引用程序集，可以通过Nuget来获取，输入关键字ENode就能看到所有相关的Package了，如下图所示：

Forum总体架构分析

Forum采用DDD+CQRS+Event Sourcing的架构。借助于ENode，使得Forum本身无须再做技术架构方面的设计了，直接使用ENode就能完成这种架构。所以我们只要明白了ENode的架构，就知道这个Forum的架构是怎样的了。以下是ENode的架构图（已经理解了这个图的朋友请直接跳过这一节）：

上图是一个CQRS架构的数据流向图。UI请求会分为两类：Command和Query。

Command用于写数据，Query用于读数据，写数据和读数据完全采用不同的架构实现。写数据支持同步和异步的方式，读数据完全走简单高效思路来实现。当我们要对系统做写操作时，如果你是用ASP.NET MVC来开发站点，那就可以在Controller中创建并发送一个Command即可。该Command会被发送到消息队列（EQueue中），然后消息队列的订阅方，也就是处理Command的进程会拉取这些Command，然后调用Command Handler完成Command的处理；Command Handler处理Command时，是调用Domain的方法来完成相关的业务逻辑操作。Domain就是DDD中的领域层，负责实现整个系统的业务逻辑。然后由于是Event Sourcing的架构，所以Domain中任何聚合根的修改都会产生相应的领域事件（Domain Event），领域事件会先被持久化到EventStore中，持久化如果没有遇到并发冲突，成功后，则会被发布（Publish）到消息队列（EQueue中），然后消息队列的订阅方，也就是处理Domain Event的进程会拉取这些Domain Event，然后调用相关的Event Handler做相关的更新，比如有些Event Handler是会更新读库（Read DB），有些是会产生新的Command，这种我把它叫做流程管理器（Process Manager，也有人叫做Saga）。当我们有时一个业务场景需要涉及到多个聚合根的修改时，我们会需要用到Process Manager。Process Manager负责对流程进行建模，它的原理是基于事件驱动的流程实现。Process Manager处理事件，然后产生响应的Command，从而完成聚合根之间的交互。一般一个流程，我们会设计一个流程聚合根以及其他的参与该流程的聚合根，Process Manager则是用于负责协调这些聚合根之间的交互。具体的例子可以看一下ENode源代码中的BankTransferSample。

关于Query端，由于都是查询，这些查询都是用于UI展示数据或者为第三方接口提供数据为目的，查询对系统无副作用。我们可以用我们自己任意喜欢的方式来实现Query端。查询面向的是Read DB。上面提到，Read DB中的数据是通过Event Handler（老外叫Denormalizer）来更新的。

所以我们可以看到，整个架构中，Command端和Query端的数据源是完全分离的。Command端***的结果就是Domain Event，Domain Event是持久化在Event Store中的；Query端的数据源就是Read DB，一般可以用关系型数据库来作为存储。CQ两端的数据同步通过Domain Event来实现。

上图的CQRS架构***的好处是在架构级别以及数据存储级别，把读写都分离了。这样我们可以方便的对读或写单独做优化。另外由于使用了Event Sourcing的架构，使得我们的Command端只要持久化了Domain Event，就意味着保存了这个Domain的所有状态。这个特性，可以让我们的框架有很多设计余地，比如不必考虑Domain Event和业务数据要强一致等问题，因为Domain Event本身就是业务数据本身了，我们通过Domain Event随时可以还原出任意时刻的Domain的状态。当我们要查询Domain的当前***数据时，就走Query端即可。当然，由于Query端是异步更新的，所以Query端的数据可能会有一点点延迟。这点也就是我们平时一直讲到的最终一致性（CQ两端的数据最终会一致）。

通过上面的架构图，我们知道，一个Command发出后会经过两个阶段的处理：1）先被某个Command Service处理（调用Domain完成业务逻辑产生Domain Event）；2）再被Event Service处理（响应Domain Event，完成Read DB的更新或者产生新的Command）；理解这两个阶段对理解下面的Forum的项目结构很有用处。

Forum项目结构分析

以上是Forum的项目工程结构，项目中包含四个宿主工程，分别是：

Forum.BrokerService：

这个工程用于宿主EQueue的Broker，整个论坛中所有的Command，Domain Event的消息，都会被放在Broker上。比如Controller发送的Command会被发送到Broker，同样Domain产生的Domain Event也会被发送到Broker；然后消费者消费消息则都是从BrokerService拉取消息。由于该宿主工程不需要和用户交互，所以我部署为Windows Service。

Forum.CommandService：

这个工程就是用于处理Command的进程，同样也部署为Windows Service。

Forum.EventService：

这个工程就是用于处理Domain Event的进程，同样也部署为Windows Service。

Forum.Web：

这个就是论坛的Web站点了，不用多讲了；这个Web站点做的事情就是发送Command或者调用Query端的查询服务查询数据；Web站点只需要依赖于Forum.Commands和Forum.QueryServices即可，因为它只需要发送Command和查询数据即可。

Forum.CommandHandlers:

所有的Command Handler都在这个工程，Command Handler的职责是处理Command，调用Domain的方法完成业务逻辑；

Forum.Commands：

所有的Command都在这个工程中，每个Command都是一个DTO，会被封装为消息发送到EQueue。

Forum.Domain：

就是论坛的领域层了，所有的聚合以、工厂、领域服务，以及领域事件等都在这个工程中。这个工程是整个Forum最有价值的地方，是业务逻辑所在的工程。

Forum.Domain.Dapper：

由于Domain中可能会定义一些接口，这些接口背后的持久化需要在外部实现；如果按照经典DDD的架构，比如仓储接口是在Domain层定义，而实现则是在基础层（Infrastructure）中。而从经典DDD的分层架构图上来看，Domain层是依赖于Infrastructure层的，但是Infrastructure层中又有一些仓储的实现类要依赖于Domain层；虽然我能理解这种双向依赖，但很容易会给不少学DDD的人带来困惑，所以我更加倾向于，把Domain看做是架构的核心，其他一切都是Domain的外围。这个思想其实和六边形架构是一个思路。就是从架构上来看，不是上层依赖于下层，而是外层依赖于内层；内层通过定义出接口，外层实现接口，内层只要面向自己定义的接口即可。所以基于这个思路，我会把Forum.Domain中定义的接口，如果用Dapper来实现，那我就定义一个Forum.Domain.Dapper这样的工程，意思是实现Forum.Domain.Dapper依赖于内层的Forum.Domain。假如以后我们有一个基于EntityFramework的实现，则只要再创建一个Forum.Domain.EntityFramework这样的工程即可。所以可以看出，Forum.Domain.Dapper这种工程司机上是Forum.Domain对外部的适配器，Forum.Domain里定义好适配接口，Forum.Domain.Dapper这种工程实现这些适配接口。基于这种思想，我们的架构就没有了上层依赖下层的概念了，而是替换为内外层的关系，内层不依赖外层，外层依赖于内层，内层与外层直接通过适配器接口来交互，或者通过Domain Event也可以。这样我们就不用再去纠结经典DDD中看似双向依赖的问题了。

Forum.Domain.Tests：

这个工程就是对Forum.Domain的一个测试工程。每个测试用例会模拟Controller发起Command，然后***检查Domain中的状态是否正确修改。

Forum.QueryServices：

这个工程定义了Query端的所有查询接口，Forum.Web站点依赖于这个工程中的查询服务接口；然后这些查询接口的实现则是放在Forum.QueryServices.Dapper中。Forum.QueryServices与Forum.QueryServices.Dapper之间的关系和Forum.Domain与Forum.Domain.Dapper之间的关系类似，这里就不在重复了。

Forum.Denormalizers.Dapper:

这个工程中的就是所有的Denormalizer，Denormalizer就是负责处理Domain Event，然后更新读库。然后由于目前使用Dapper实现数据持久化，所以工程名以Dapper结尾。

Forum.Infrastructure：

这是一个基础工程，存放所有基础的公共的东西，比如一些业务无关的服务或配置信息或全局变量等东西；需要强调的是：这里的Forum.Infrastructure和经典DDD中的Infrastructure不是同一个概念。DDD中的Infrastructure是一个逻辑上的分层，领域层中所有的技术支撑实现都在Infrastructure中；而这里的Infrastructure，则仅仅只是一些Common的基础的公用的东西，Infrastructure不是为了为其它哪一层服务的，它可以被其他任何项目使用；

好了，以上简单介绍了每个工程的作用和设计目的。下面我们来看看Forum的领域模型的设计吧！

Forum的Domain Model的设计

核心功能需求分析：

提供用户注册、登录、注销三个功能；注册用户时需要验证用户名是否唯一；
提供发帖、回帖、修改帖子、修改回复，以及回复的回复这些基本核心功能；
系统管理员可以对论坛版块进行维护；

聚合识别：识别出来的聚合有：论坛账号、帖子、回复、版块这四个。
再分析下每个聚合我们所关心的信息：账号的最少信息应该有：账号名称+密码；版块要有名称即可；帖子要有标题、内容、发帖人、发帖时间、所属版块；回复要有回复内容、回复时间、回复人、所属版块，父回复（可以为空）；
场景走查：注册就是创建账号（账号唯一性的设计后面在详细分析）；登录本质就是调用Query端的查询服务查找账号是否存在，所以不需要Domain做什么处理，注销也是；发帖就是创建帖子；回帖就是创建回复；修改帖子就是对帖子聚合根做修改；修改回复就是对回复聚合根做修改；版块添加就是创建一个版块聚合根；
关键业务规则识别：1）账号名称不能重复；2）帖子必须要有所属版块和发帖人；3）回复必须要有一个对应的帖子和回复人；

关键业务规则的实现：

如何实现账号名称不能重复？首先它是一条业务规则，所以必须在Domain里实现，而不应该在Command Handler里。然后由于Event Sourcing的架构，天生有一个缺陷就是无法实现唯一性约束这种需求。所以我们需要在Domain中显式设计出可以表达聚合根索引的东西，我把它们叫做IndexStore，表示是一种聚合根索引的存储。这个思路非常类似于在经典DDD中，我们有仓储（Repository）的概念，仓储维护了所有的聚合根；而我这里的IndexStore则是维护了聚合根的索引信息。有了这个索引信息后，我们就能在注册新账号时，在Domain中设计一个RegisterAccountService这样的领域服务，领域服务里通过AccountIndexStore来检查账号名称是否重复，如果不重复，则将当前账号名称添加到AccountIndexStore中，如果重复，则报异常。另外一个非业务的点需要考虑，那就是如何实现并发注册用户的处理。我们可以在Command Handler中实现db级别的锁（但不不需要锁整个账号表，而是锁一个其他表中的某一条记录），确保同一时刻，不会有两个Account名称添加到AccountIndexStore中；我们通过RegisterAccountService把"账号名称不能重复"的这个业务规则显式的表达出来，从而在代码级别体现领域内实现了这个业务规则。以前，如果没有用Event Sourcing，我们可能会依赖db的唯一索引来实现这个唯一性，虽然功能上也可以实现，但实际上账号名称不能重复的这个业务规则没有体现在领域内。这点也是我这次通过实现基于Event Sourcing而实现的唯一性验证而想到的点。
帖子必须要有所属版块和发帖人，这条业务规则很容易保证，只要在帖子聚合根上，对版块和发帖子判断是否为空就行了；
回复必须要有一个对应的帖子和回复人，也是同理，只要在构造函数中判断是否为空即可；

以注册新用户为例，展示代码实现

客户端JS通过angularJS提交注册信息：

$scope.submit = function () {          if (isStringEmpty($scope.newAccount.accountName)) {              $scope.errorMsg = '请输入账号。';              return false;          }          if (isStringEmpty($scope.newAccount.password)) {              $scope.errorMsg = '请输入密码。';              return false;          }          if (isStringEmpty($scope.newAccount.confirmPassword)) {              $scope.errorMsg = '请输入密码确认。';              return false;          }          if ($scope.newAccount.password != $scope.newAccount.confirmPassword) {              $scope.errorMsg = '密码输入不一致。';              return false;          }           $http({              method: 'POST',              url: '/account/register',              data: $scope.newAccount          })          .success(function (result, status, headers, config) {              if (result.success) {                  $_window.location.href = '/home/index';              } else {                  $scope.errorMsg = result.errorMsg;              }          })          .error(function (result, status, headers, config) {              $scope.errorMsg = result.errorMsg;          });      };

Controller处理请求：

[HttpPost]  [AjaxValidateAntiForgeryToken]  [AsyncTimeout(5000)]  public async Task Register(RegisterModel model, CancellationToken token)  {      var command = new RegisterNewAccountCommand(model.AccountName, model.Password);      var result = await _commandService.Execute(command, CommandReturnType.EventHandled);       if (result.Status == CommandStatus.Failed)      {          if (result.ExceptionTypeName == typeof(DuplicateAccountException).Name)          {              return Json(new { success = false, errorMsg = "该账号已被注册，请用其他账号注册。" });          }          return Json(new { success = false, errorMsg = result.ErrorMessage });      }       _authenticationService.SignIn(result.AggregateRootId, model.AccountName, false);      return Json(new { success = true });  }

CommandHandler处理Command：

[Component(LifeStyle.Singleton)]  public class AccountCommandHandler : ICommandHandler  {      private readonly ILockService _lockService;      private readonly RegisterAccountService _registerAccountService;       public AccountCommandHandler(ILockService lockService, RegisterAccountService registerAccountService)      {          _lockService = lockService;          _registerAccountService = registerAccountService;      }       public void Handle(ICommandContext context, RegisterNewAccountCommand command)      {          _lockService.ExecuteInLock(typeof(Account).Name, () =>          {              context.Add(_registerAccountService.RegisterNewAccount(command.Id, command.Name, command.Password));          });      }  }

RegisterAccountService领域服务：

/// 提供账号注册的领域服务，封装账号注册的业务规则，比如账号唯一性检查      /// 
      [Component(LifeStyle.Singleton)]      public class RegisterAccountService      {          private readonly IIdentityGenerator _identityGenerator;          private readonly IAccountIndexStore _accountIndexStore;          private readonly AggregateRootFactory _factory;           public RegisterAccountService(IIdentityGenerator identityGenerator, AggregateRootFactory factory, IAccountIndexStore accountIndexStore)          {              _identityGenerator = identityGenerator;              _factory = factory;              _accountIndexStore = accountIndexStore;          }           /// 注册新账号          /// 
          ///           ///           ///           ///           public Account RegisterNewAccount(string accountIndexId, string accountName, string accountPassword)          {              //首先创建一个新账号              var account = _factory.CreateAccount(accountName, accountPassword);               //先判断该账号是否存在              var accountIndex = _accountIndexStore.FindByAccountName(account.Name);              if (accountIndex == null)              {                  //如果不存在，则添加到账号索引                  _accountIndexStore.Add(new AccountIndex(accountIndexId, account.Id, account.Name));              }              else if (accountIndex.IndexId != accountIndexId)              {                  //如果存在但和当前的索引ID不同，则认为是账号有重复                  throw new DuplicateAccountException(accountName);              }               return account;          }      }

EventHandler处理Domain Event：

[Component(LifeStyle.Singleton)]  public class AccountEventHandler : BaseEventHandler, IEventHandler  {      public void Handle(IEventContext context, NewAccountRegisteredEvent evnt)      {          using (var connection = GetConnection())          {              connection.Insert(                  new                 {                      Id = evnt.AggregateRootId,                      Name = evnt.Name,                      Password = evnt.Password,                      CreatedOn = evnt.Timestamp,                      UpdatedOn = evnt.Timestamp,                      Version = evnt.Version                  }, Constants.AccountTable);          }      }  }

"使用ENode 2.0举例分析"的内容就介绍到这里了，感谢大家的阅读。如果想了解更多行业相关的知识可以关注网站，小编将为大家输出更多高质量的实用文章！

很赞哦！