enode框架step by step之saga的思想与实现
enode框架系列step by step文章系列索引：
因为enode框架的思想是，一次修改只能新建或修改一个聚合根；那么，如果一个用户请求要涉及多个聚合根的新建或修改该怎么办呢？本文的目的就是要分析清楚这个问题在enode框架下是如何解决的。如果想直接通过看代码的朋友，可以直接下载源代码，源码中共有三个例子，BankTransferSagaSample这个例子就是本文所用的例子。
saga的由来
saga这个术语，可能很多人都还很陌生。saga的提出，最早是为了解决可能会长时间运行的分布式事务（long-running process）的问题。所谓long-running的分布式事务，是指那些企业业务流程，需要跨应用、跨企业来完成某个事务，甚至在事务流程中还需要有手工操作的参与，这类事务的完成时间可能以分计，以小时计，甚至可能以天计。这类事务如果按照事务的ACID的要求去设计，势必造成系统的可用性大大的降低。试想一个由两台服务器一起参与的事务，服务器A发起事务，服务器B参与事务，B的事务需要人工参与，所以处理时间可能很长。如果按照ACID的原则，要保持事务的隔离性、一致性，服务器A中发起的事务中使用到的事务资源将会被锁定，不允许其他应用访问到事务过程中的中间结果，直到整个事务被提交或者回滚。这就造成事务A中的资源被长时间锁定，系统的可用性将不可接受。
而saga，则是一种基于补偿的消息驱动的用于解决long-running process的一种解决方案。目标是为了在确保系统高可用的前提下尽量确保数据的一致性。还是上面的例子，如果用saga来实现，那就是这样的流程：服务器A的事务先执行，如果执行顺利，那么事务A就先行提交；如果提交成功，那么就开始执行事务B，如果事务B也执行顺利，则事务B也提交，整个事务就算完成。但是如果事务B执行失败，那事务B本身需要回滚，这时因为事务A已经提交，所以需要执行一个补偿操作，将已经提交的事务A执行的操作作反操作，恢复到未执行前事务A的状态。这样的基于消息驱动的实现思路，就是saga。我们可以看出，saga是牺牲了数据的强一致性，仅仅实现了最终一致性，但是提高了系统整体的可用性。
CQRS架构下的saga (process manager)
上面一段，我们知道了saga的由来，现在我们再看一下CQRS架构下，saga是用来做什么的。虽然都叫saga，但是实际上在CQRS架构下，人们往往用saga来解决DDD中多个聚合或多个bounded context之间的通信问题。DDD中有bounded context的概念。一个bounded context代表一个上下文边界，一个bounded context中可能包含一个或多个聚合。而saga就是用来实现bounded context之间的通信，或者是聚合之间的通信。在经典DDD中，我们通常用领域服务来实现多个聚合的协调，并最终通过事务的方式来提交所有聚合的修改；这样做的后果是，1：用到了事务；2.一个事务涉及了多个聚合的更改；这样做没什么不好，在条件允许的情况下（比如不会出现分布式事务的情况下或者并发修改的请求数不高的情况下），这样做没什么特别大的问题。唯一的问题是，这样做会增加并发冲突的几率。现在的web应用，往往都是多用户在同时向系统发送各种处理请求，所以我们不难想到，一个事务中涉及到的聚合越多，那并发冲突的可能性就越高。不仅如此，如果你的聚合很大，包含了很多的子实体和很多的方法，那该聚合持久化时产生并发冲突的几率也会相对较高；而系统的并发冲突将直接影响系统的可用性；所以，一般的建议是，我们应该尽量将聚合设计的小，且尽量一次只修改一个聚合；这样我们就不需要事务，还能把并发冲突的可能性降到最低；当然，单个聚合持久化时也还会存在并发冲突，但这点相对很容易解决，因为单个聚合是数据一致性的最小单元，所以我们可以完全不需要事务，通过乐观锁就能解决并发覆盖的问题；关于这个问题的讨论，大家如果还有兴趣或者想了解的更加深入，我推荐看一下Effective Aggregate Design这篇论文，共三个部分，其作者是《implementing domain-driven design》一书的作者。
所以，通过上面的分析，我们知道了“聚合应该设计的小，且一次修改只修改一个聚合”这样一条不成文的原则。当然你一定有很多理由认为不应该这样，欢迎大家讨论。那么如果要遵循这样的原则，那我们需要一种机制来解决多个聚合之间的通信的问题。在CQRS的架构下，人们也都把这种机制叫做saga，但因为这种CQRS架构下的saga的语义已经不是上面一段中介绍的saga了。所以，微软也把cqrs架构下的saga叫做process manager，具体可以看一下微软的一个CQRS架构的开源项目的例子；process manager这个名字我们应该很容易理解，即流程管理器。事实上，一个saga所做的事情就是和一个流程一样的事情。只不过传统的流程，都有一个流程定义，当用户发起一个流程时，就会产生一个流程实例，该流程实例会严格按照流程定义的流向来进行流转，驱动流程流转的往往是人的操作，比如审批操作。而process manager，也是一个流程，只不过这个流程是由消息驱动的。一个典型的process manager会响应事件，然后产生新的命令去执行下一步操作。用过NServiceBus的人应该知道，NServiceBus中就内置了saga的机制，我们可以很轻松的利用它来实现分布式的消息驱动的long-running process；
如何用enode框架来实现saga
为了说明问题，我就以经典的银行转账为例子来讲解吧，因为转账的核心业务大家都很清楚，所以我们就没有了业务上理解的不一致，我们就能专心思考如何实现的问题了。但是，为了便于下面的分析，我还是简单定义一下本例中的银行转账的核心业务流程。注意：实际的转账业务流程远比我定义的复杂，我这里重点是为了分析如何实现一个会涉及多个聚合修改的的业务场景。核心业务描述如下：
- 两个银行账号，一个是源账号，一个是目标账号；
- 用户点击确认转账按钮后，指定数目的钱会从源账号转入到目标账号；
- 整个转账过程有两个阶段：1）钱从源账号转出；2）钱转入到目标账号；如果一切顺利，那转账流程就结束了；
- 如果源账号的当前余额不足，则转出操作会失败，系统记录错误日志，转账流程结束；
- 如果钱转入到目标账号时出现异常，则需要回滚源账号已转出的钱，同时记录错误日志，回滚完成后，转账流程结束；
思路分析：
- 通过上面的需求，我们知道，应该有一个聚合根，表示银行账号，我设计为BankAccount；BankAccount有转出钱和转入钱的行为职责。另外，根据上面第5条需求，BankAccount可能会有回滚转出钱的行为职责；另外，当然一个银行账号还会有一个表示当前余额的状态属性；
- 由于我们是通过saga的思想来实现转账流程，那我们具体该如何设计此saga呢？saga在CQRS架构中的作用是响应事件，产生command，从而起到以事件消息驱动的原理来控制流程流转的作用；转账流程如何才能结束会由saga来决定。那么saga要响应的事件哪里来呢？很明显，就是从流程中涉及到的聚合根里来，本例就是响应BankAccount的事件；当BankAccount的转出事件或转入事件发生后，会被saga响应，然后saga会做出响应，决定下一步该怎么走。 saga是聚合根吗？或者说saga属于领域层的东西吗？这个问题很重要，我觉得没有明确的答案。而且我也没有从各种资料上明确看到saga是属于ddd中的应用层还是领域层还是其他层。以下是我个人的一些思考：
- 关于认为saga应该属于领域层的原因的一些思考：和经典的DDD做类比，经典DDD的书本上，会有一个银行转账的领域服务，该领域服务完成转账的核心业务逻辑；而一些外围的逻辑，如记录日志、发送邮件或短信通知等功能，则在外围的应用层服务中做掉；所以按照这个理解，假如我们设计一个saga，来实现转账的核心业务逻辑，那我觉得saga也是一个聚合根。因为saga是一个流程，职责是控制转账的过程，它有一个全局唯一的流程ID（一次转账就会产生一个转账流程，流程ID是流程的全局唯一标识），属于领域层，saga可以理解为是领域中对行为过程的建模。当然saga与普通的聚合根稍有区别，普通的聚合根我们通常是根据名词法则去识别，而saga则是从交互行为或者流程的角度去识别；这点就好比经典DDD的领域模型中有聚合根和领域服务一样，聚合根是数据的建模，领域服务是交互行为的建模；
- 关于认为saga不应该属于领域层的原因的一些思考：按照saga在CQRS架构下的定义，它会接受响应event，然后发送command。而command是应用层的东西，所以就会导致domain层依赖于应用层，显然不太合理；
- 关于认为saga不应该属于应用层的原因的一些思考：因为saga是流程，且有流程状态，有状态就需要保存，这样就变成应用层中的saga需要保存状态，这种做法合理吗？值得我们深思；另外，按照经典DDD的架构中对应用层的职责定义，应用层应该是很薄的，更加不会出现需要保存状态的属于应用层的对象；
- 通过上面第3点的一些讨论，我个人会把saga设计在领域层，设计为聚合根，但是，我会对saga的实现做一些调整：1）saga聚合根不会直接响应事件，而是经过一个中间command来过度；2）saga聚合根也不会直接发送command，取而代之的是像普通聚合根一样也产生事件，这种事件表达的意思是“发送某某command的意图已下达”，然后外层的event handler接受到这样的事件后，会发送对应的command给command service；这样一来，saga聚合根就和普通的聚合根无任何差别，听上去感觉很不可思议，稍后我们结合代码一起看一下具体实现吧。
- 如果saga也是一个聚合根，那不是和BankAccount平级了，那BankAccount产生的事件如何传递给saga呢？显然，我们还缺少一样东西，就是需要把流程中涉及到修改的聚合根产生的事件传递给saga聚合根的event handler。这种event handler本身无业务逻辑，他们的职责是监听聚合根产生的事件，然后将event转化为command，然后将command发送到command service，从而最后通知到对应的saga，然后saga就开始处理该事件，比如决定接下来该如何处理；
代码实现：
BankAccount聚合根的设计
/// <summary>银行账号聚合根 /// </summary> [Serializable] public class BankAccount : AggregateRoot<Guid>, IEventHandler<AccountOpened>, //银行账户已开 IEventHandler<Deposited>, //钱已存入 IEventHandler<TransferedOut>, //钱已转出 IEventHandler<TransferedIn>, //钱已转入 IEventHandler<TransferOutRolledback> //转出已回滚 { /// <summary>账号（卡号） /// </summary> public string AccountNumber { get; private set; } /// <summary>拥有者 /// </summary> public string Owner { get; private set; } /// <summary>当前余额 /// </summary> public double Balance { get; private set; } public BankAccount() : base() { } public BankAccount(Guid accountId, string accountNumber, string owner) : base(accountId) { RaiseEvent(new AccountOpened(Id, accountNumber, owner)); } /// <summary>存款 /// </summary> /// <param name="amount"></param> public void Deposit(double amount) { RaiseEvent(new Deposited(Id, amount, string.Format("向账户{0}存入金额{1}", AccountNumber, amount))); } /// <summary>转出 /// </summary> /// <param name="targetAccount"></param> /// <param name="processId"></param> /// <param name="transferInfo"></param> public void TransferOut(BankAccount targetAccount, Guid processId, TransferInfo transferInfo) { //这里判断当前余额是否足够 if (Balance < transferInfo.Amount) { throw new Exception(string.Format("账户{0}余额不足，不能转账！", AccountNumber)); } RaiseEvent(new TransferedOut(processId, transferInfo, string.Format("{0}向账户{1}转出金额{2}", AccountNumber, targetAccount.AccountNumber, transferInfo.Amount))); } /// <summary>转入 /// </summary> /// <param name="sourceAccount"></param> /// <param name="processId"></param> /// <param name="transferInfo"></param> public void TransferIn(BankAccount sourceAccount, Guid processId, TransferInfo transferInfo) { RaiseEvent(new TransferedIn(processId, transferInfo, string.Format("{0}从账户{1}转入金额{2}", AccountNumber, sourceAccount.AccountNumber, transferInfo.Amount))); } /// <summary>回滚转出 /// </summary> /// <param name="processId"></param> /// <param name="transferInfo"></param> public void RollbackTransferOut(Guid processId, TransferInfo transferInfo) { RaiseEvent(new TransferOutRolledback(processId, transferInfo, string.Format("账户{0}取消转出金额{1}", AccountNumber, transferInfo.Amount))); } void IEventHandler<AccountOpened>.Handle(AccountOpened evnt) { AccountNumber = evnt.AccountNumber; Owner = evnt.Owner; } void IEventHandler<Deposited>.Handle(Deposited evnt) { Balance += evnt.Amount; } void IEventHandler<TransferedOut>.Handle(TransferedOut evnt) { Balance -= evnt.TransferInfo.Amount; } void IEventHandler<TransferedIn>.Handle(TransferedIn evnt) { Balance += evnt.TransferInfo.Amount; } void IEventHandler<TransferOutRolledback>.Handle(TransferOutRolledback evnt) { Balance += evnt.TransferInfo.Amount; } }
转账流程TransferProcess聚合根的设计
/// <summary>转账流程状态 /// </summary> public enum ProcessState { NotStarted, Started, TransferOutRequested, TransferInRequested, RollbackTransferOutRequested, Completed, Aborted } /// <summary>转账信息值对象，包含了转账的基本信息 /// </summary> [Serializable] public class TransferInfo { public Guid SourceAccountId { get; private set; } public Guid TargetAccountId { get; private set; } public double Amount { get; private set; } public TransferInfo(Guid sourceAccountId, Guid targetAccountId, double amount) { SourceAccountId = sourceAccountId; TargetAccountId = targetAccountId; Amount = amount; } } /// <summary>银行转账流程聚合根，负责控制整个转账的过程，包括遇到异常时的回滚处理 /// </summary> [Serializable] public class TransferProcess : AggregateRoot<Guid>, IEventHandler<TransferProcessStarted>, //转账流程已开始 IEventHandler<TransferOutRequested>, //转出的请求已发起 IEventHandler<TransferInRequested>, //转入的请求已发起 IEventHandler<RollbackTransferOutRequested>, //回滚转出的请求已发起 IEventHandler<TransferProcessCompleted>, //转账流程已正常完成 IEventHandler<TransferProcessAborted> //转账流程已异常终止 { /// <summary>当前转账流程状态 /// </summary> public ProcessState State { get; private set; } public TransferProcess() : base() { } public TransferProcess(BankAccount sourceAccount, BankAccount targetAccount, TransferInfo transferInfo) : base(Guid.NewGuid()) { RaiseEvent(new TransferProcessStarted(Id, transferInfo, string.Format("转账流程启动，源账户：{0}，目标账户：{1}，转账金额：{2}", sourceAccount.AccountNumber, targetAccount.AccountNumber, transferInfo.Amount))); RaiseEvent(new TransferOutRequested(Id, transferInfo)); } /// <summary>处理已转出事件 /// </summary> /// <param name="transferInfo"></param> public void HandleTransferedOut(TransferInfo transferInfo) { RaiseEvent(new TransferInRequested(Id, transferInfo)); } /// <summary>处理已转入事件 /// </summary> /// <param name="transferInfo"></param> public void HandleTransferedIn(TransferInfo transferInfo) { RaiseEvent(new TransferProcessCompleted(Id, transferInfo)); } /// <summary>处理转出失败的情况 /// </summary> /// <param name="transferInfo"></param> public void HandleFailedTransferOut(TransferInfo transferInfo) { RaiseEvent(new TransferProcessAborted(Id, transferInfo)); } /// <summary>处理转入失败的情况 /// </summary> /// <param name="transferInfo"></param> public void HandleFailedTransferIn(TransferInfo transferInfo) { RaiseEvent(new RollbackTransferOutRequested(Id, transferInfo)); } /// <summary>处理转出已回滚事件 /// </summary> /// <param name="transferInfo"></param> public void HandleTransferOutRolledback(TransferInfo transferInfo) { RaiseEvent(new TransferProcessAborted(Id, transferInfo)); } void IEventHandler<TransferProcessStarted>.Handle(TransferProcessStarted evnt) { State = ProcessState.Started; } void IEventHandler<TransferOutRequested>.Handle(TransferOutRequested evnt) { State = ProcessState.TransferOutRequested; } void IEventHandler<TransferInRequested>.Handle(TransferInRequested evnt) { State = ProcessState.TransferInRequested; } void IEventHandler<RollbackTransferOutRequested>.Handle(RollbackTransferOutRequested evnt) { State = ProcessState.RollbackTransferOutRequested; } void IEventHandler<TransferProcessCompleted>.Handle(TransferProcessCompleted evnt) { State = ProcessState.Completed; } void IEventHandler<TransferProcessAborted>.Handle(TransferProcessAborted evnt) { State = ProcessState.Aborted; } }
响应BankAccount聚合根所发生的事件的event handler设计
/// <summary>事件订阅者，用于监听和响应银行账号聚合根产生的事件 /// </summary> public class BankAccountEventHandler : IEventHandler<AccountOpened>, //银行账户已开 IEventHandler<Deposited>, //钱已存入 IEventHandler<TransferedOut>, //钱已转出 IEventHandler<TransferedIn>, //钱已转入 IEventHandler<TransferOutRolledback> //转出已回滚 { private ICommandService _commandService; public BankAccountEventHandler(ICommandService commandService) { _commandService = commandService; } void IEventHandler<AccountOpened>.Handle(AccountOpened evnt) { Console.WriteLine(string.Format("创建银行账户{0}", evnt.AccountNumber)); } void IEventHandler<Deposited>.Handle(Deposited evnt) { Console.WriteLine(evnt.Description); } void IEventHandler<TransferedOut>.Handle(TransferedOut evnt) { Console.WriteLine(evnt.Description); //响应已转出事件，发送“处理已转出事件”的命令 _commandService.Send(new HandleTransferedOut { ProcessId = evnt.ProcessId, TransferInfo = evnt.TransferInfo }); } void IEventHandler<TransferedIn>.Handle(TransferedIn evnt) { Console.WriteLine(evnt.Description); //响应已转入事件，发送“处理已转入事件”的命令 _commandService.Send(new HandleTransferedIn { ProcessId = evnt.ProcessId, TransferInfo = evnt.TransferInfo }); } void IEventHandler<TransferOutRolledback>.Handle(TransferOutRolledback evnt) { Console.WriteLine(evnt.Description); //响应转出已回滚事件，发送“处理转出已回滚事件”的命令 _commandService.Send(new HandleTransferOutRolledback { ProcessId = evnt.ProcessId, TransferInfo = evnt.TransferInfo }); } }
响应TransferProcess聚合根所发生的事件的event handler设计
/// <summary>事件订阅者，用于监听和响应转账流程聚合根产生的事件 /// </summary> public class TransferProcessEventHandler : IEventHandler<TransferProcessStarted>, //转账流程已开始 IEventHandler<TransferOutRequested>, //转出的请求已发起 IEventHandler<TransferInRequested>, //转入的请求已发起 IEventHandler<RollbackTransferOutRequested>, //回滚转出的请求已发起 IEventHandler<TransferProcessCompleted>, //转账流程已完成 IEventHandler<TransferProcessAborted> //转账流程已终止 { private ICommandService _commandService; public TransferProcessEventHandler(ICommandService commandService) { _commandService = commandService; } void IEventHandler<TransferProcessStarted>.Handle(TransferProcessStarted evnt) { Console.WriteLine(evnt.Description); } void IEventHandler<TransferOutRequested>.Handle(TransferOutRequested evnt) { //响应“转出的命令请求已发起”这个事件，发送“转出”命令 _commandService.Send(new TransferOut { ProcessId = evnt.ProcessId, TransferInfo = evnt.TransferInfo }, (result) => { //这里是command的异步回调函数，如果有异常，则发送“处理转出失败”的命令 if (result.Exception != null) { Console.WriteLine(result.Exception.Message); _commandService.Send(new HandleFailedTransferOut { ProcessId = evnt.ProcessId, TransferInfo = evnt.TransferInfo }); } }); } void IEventHandler<TransferInRequested>.Handle(TransferInRequested evnt) { //响应“转入的命令请求已发起”这个事件，发送“转入”命令 _commandService.Send(new TransferIn { ProcessId = evnt.ProcessId, TransferInfo = evnt.TransferInfo }, (result) => { //这里是command的异步回调函数，如果有异常，则发送“处理转入失败”的命令 if (result.Exception != null) { Console.WriteLine(result.Exception.Message); _commandService.Send(new HandleFailedTransferIn { ProcessId = evnt.ProcessId, TransferInfo = evnt.TransferInfo }); } }); } void IEventHandler<RollbackTransferOutRequested>.Handle(RollbackTransferOutRequested evnt) { //响应“回滚转出的命令请求已发起”这个事件，发送“回滚转出”命令 _commandService.Send(new RollbackTransferOut { ProcessId = evnt.ProcessId, TransferInfo = evnt.TransferInfo }); } void IEventHandler<TransferProcessCompleted>.Handle(TransferProcessCompleted evnt) { Console.WriteLine("转账流程已正常完成！"); } void IEventHandler<TransferProcessAborted>.Handle(TransferProcessAborted evnt) { Console.WriteLine("转账流程已异常终止！"); } }
BankAccount聚合根相关的command handlers
/// <summary>银行账户相关命令处理 /// </summary> public class BankAccountCommandHandlers : ICommandHandler<OpenAccount>, //开户 ICommandHandler<Deposit>, //存钱 ICommandHandler<TransferOut>, //转出 ICommandHandler<TransferIn>, //转入 ICommandHandler<RollbackTransferOut> //回滚转出 { public void Handle(ICommandContext context, OpenAccount command) { context.Add(new BankAccount(command.AccountId, command.AccountNumber, command.Owner)); } public void Handle(ICommandContext context, Deposit command) { context.Get<BankAccount>(command.AccountId).Deposit(command.Amount); } public void Handle(ICommandContext context, TransferOut command) { var sourceAccount = context.Get<BankAccount>(command.TransferInfo.SourceAccountId); var targetAccount = context.Get<BankAccount>(command.TransferInfo.TargetAccountId); sourceAccount.TransferOut(targetAccount, command.ProcessId, command.TransferInfo); } public void Handle(ICommandContext context, TransferIn command) { var sourceAccount = context.Get<BankAccount>(command.TransferInfo.SourceAccountId); var targetAccount = context.Get<BankAccount>(command.TransferInfo.TargetAccountId); targetAccount.TransferIn(sourceAccount, command.ProcessId, command.TransferInfo); } public void Handle(ICommandContext context, RollbackTransferOut command) { context.Get<BankAccount>(command.TransferInfo.SourceAccountId).RollbackTransferOut(command.ProcessId, command.TransferInfo); } }
TransferProcess聚合根相关的command handlers
/// <summary>银行转账流程相关命令处理 /// </summary> public class TransferProcessCommandHandlers : ICommandHandler<StartTransfer>, //开始转账 ICommandHandler<HandleTransferedOut>, //处理已转出事件 ICommandHandler<HandleTransferedIn>, //处理已转入事件 ICommandHandler<HandleFailedTransferOut>, //处理转出失败 ICommandHandler<HandleFailedTransferIn>, //处理转入失败 ICommandHandler<HandleTransferOutRolledback> //处理转出已回滚事件 { public void Handle(ICommandContext context, StartTransfer command) { var sourceAccount = context.Get<BankAccount>(command.TransferInfo.SourceAccountId); var targetAccount = context.Get<BankAccount>(command.TransferInfo.TargetAccountId); context.Add(new TransferProcess(sourceAccount, targetAccount, command.TransferInfo)); } public void Handle(ICommandContext context, HandleTransferedOut command) { context.Get<TransferProcess>(command.ProcessId).HandleTransferedOut(command.TransferInfo); } public void Handle(ICommandContext context, HandleTransferedIn command) { context.Get<TransferProcess>(command.ProcessId).HandleTransferedIn(command.TransferInfo); } public void Handle(ICommandContext context, HandleFailedTransferOut command) { context.Get<TransferProcess>(command.ProcessId).HandleFailedTransferOut(command.TransferInfo); } public void Handle(ICommandContext context, HandleFailedTransferIn command) { context.Get<TransferProcess>(command.ProcessId).HandleFailedTransferIn(command.TransferInfo); } public void Handle(ICommandContext context, HandleTransferOutRolledback command) { context.Get<TransferProcess>(command.ProcessId).HandleTransferOutRolledback(command.TransferInfo); } }
上面的代码中都加了详细的注释，有不清楚的，直接回复问我吧，呵呵。