服装网站建设的需求,wordpress 水煮鱼小程序,国家精品课程网官网,网络营销与策划实践在之前的一篇博文中#xff0c;CAP框架可以方便我们实现非实时、异步场景下的最终一致性#xff0c;而有些用例总是无法避免的需要在实时、同步场景下进行#xff0c;可以借助Saga事务来解决这一困扰。在一些博文和仓库中也搜寻到了.Net下实现Saga模式的解决方案MassTransit… 在之前的一篇博文中CAP框架可以方便我们实现非实时、异步场景下的最终一致性而有些用例总是无法避免的需要在实时、同步场景下进行可以借助Saga事务来解决这一困扰。在一些博文和仓库中也搜寻到了.Net下实现Saga模式的解决方案MassTransit这就省得自己再造轮子了。分布式事务分布式系统中分布式事务是一个不能避免的问题如何保证不同节点间的数据一致性。举个常见的例子下订单、减库存、扣余额三者在单个节点时可以借助本地事务实现要么成功要么失败。而当三者处于不同节点时又参杂了如网络环境、节点自身环境、服务环境等各种因素使得三个节点想要实现要么成功、要么失败就增加了许多困难。CAP理论和BASE理论很好的诠释了这一问题也有了许多的解决分布式事务的方案如2PC、3PC、TCC、本地消息表、Saga等一系列解决方案面对不同场景、不同要求等可选择不同的解决方案。数据一致性容错性复杂性性能维护成本2PC强低中低低3PC强低高低中TCC弱高高中高本地消息表弱高低中中MQ事务弱高低高中Saga事务弱高高中高在之前提到过一个基于本地消息表的CAP框架借助最终一致性很方便的解决了异步非实时请求下的分布式事务而对于大部分场景虽然可以直接或者妥协方式使用着异步非实时如同步实时场景的下订单且减库存变更到异步非实时场景的下订单后发事件减库存但是总有那么一些场景不得不去考虑同步实时请求下的分布式事务。Saga模式Saga模式又叫做长时间运行事务Long-running-transaction, 由普林斯顿大学的 Hector Garcia-Molina和Kenneth Salem 1987年发表的论文《Sagas》。核心思想是将长事务拆分为多个本地短事务通过保证所有短事务的成功或失败来决定整体的成功或失败由Saga事务协调器协调管理所有节点执行成功则成功如有节点失败则反向执行前置节点的补偿操作。每个Saga事务由一系列幂等的有序子事务(sub-transaction) Ti 组成。每个Ti 都有对应的幂等补偿动作Ci补偿动作用于撤销Ti造成的结果。执行过程当正常执行时依照T1、T2、T3三个短事务正常执行下去直到最后一个Tn事务执行完毕宣告整个事务的成功。而当执行到某个Tj出现故障时则反向补偿之前的Tj-1..T1,每个对应的补偿操作Cj-1...C1其中Tj事务由于在执行阶段就已失败所以Tj对应的补偿动作Cj不需要执行即也确定了最后一个Tn事务可以不设置补偿动作Cn。恢复策略向前恢复(forward recovery)对于Ti事务的执行部分场景下可能因为数据库的连接、网络的波动等导致短暂的失败对Ti事务重试执行以确保整个事务的执行如执行T1, T2, T3当执行T3失败时不直接宣告失败对T3执行重试以排除部分不稳定因素如在若干次重试无效后再考虑向后恢复。向后恢复(backward recovery)按照执行顺序方式作为向前的指向则向后为反向补偿对已执行过的节点顺序倒退执行各Ti的补偿动作Ci也就是把走过的路往回走对执行过的操作执行业务上的反操作如正向流程执行减库存则补偿操作时执行加库存。协作方式对于服务与服务间的协作我们通常有两种模式Orchestration(编排式) 和 Choreography(协同式)在Saga模式中也有着这两种的实现。编排式Orchestrator把 Saga 的决策和执行顺序逻辑集中在一个 Saga 编排器类中。Saga 编排器发出命令式消息给各个 Saga 参与方指示这些参与方服务完成具体操作本地事务。协同式Choreography把 Saga 的决策和执行顺序逻辑分布在 Saga 的每个参与方中它们通过交换事件的方式来进行沟通。编排式与协同式的差异仅在于服务之间的协作方式每个参与服务的接口定义却没有任何区别。编排式Orchestrator编排式的 Saga 需要开发人员定义一个编排器类用于编排一个Saga中多个参与服务执行的流程。如果整个业务流程正常结束业务就成功完成一旦这个过程的任何环节出现失败Saga编排器类就会以相反的顺序调用补偿操作重新进行业务回滚。对于每个参与的服务而言需要做的事情是订阅并处理命令消息执行命令后返回响应消息设计执行逻辑和补偿逻辑以提交订单为例假设场景是分布式系统下进程间以消息传递进行通信1、事务发起方的主业务逻辑请求预先定义好的Saga编排器类(内部编排了执行顺序)。2、Saga编排器类向MQ发送减库存事件库存服务订阅事件、执行处理并返回MQ处理结果。3、Saga编排器类向MQ发送减余额事件支付服务订阅事件、执行处理并返回MQ处理结果。4、Saga编排器类向MQ发送创建订单命令订单服务订阅事件并按照命令创建订单。5、主业务逻辑接收并处理Saga编排器类处理结果。6、整个过程由Saga 编排器类对接收到的回复进行判决来决定是继续执行还是悬崖勒马。协同式Choreography没有集中式的编排类而是各参与方间相互订阅一个服务订阅另一个服务的事件。先由事务发起方执行逻辑并发布一个事件该事件被一个或多个服务进行订阅这些服务执行本地数据库操作并发布或不发布新的事件该部分需要保证本地数据库的操作成功且写入MQ的消息也成功可考虑使用本地消息表或是基于MQ事务。当最后一个服务执行本地事务并且不发布任何事件或者发布的事件没有被任何Saga参与者订阅意味着事务结束则整个业务流程的分布式事务完成。如果某一服务出现故障那么则反向发布事件执行补偿操作以此回滚。以提交订单为例假设场景是分布式系统下进程间以消息传递进行通信1、事务发起方执行主业务逻辑发送提交订单命令。2、库存服务订阅事件、扣减库存并发布已扣减事件。3、订单服务订阅库存已扣减事件创建订单并发布订单已创建事件。4、支付服务订阅订单已创建事件执行支付并发布订单已支付事件。5、主业务逻辑订阅订单已支付事件并处理。当某服务内执行时如存在异常则反向发布事件如订单创建失败则发布OrderCreatedFailed事件库存服务订阅该事件并执行补偿操作。相比而言编排式中参与服务无需向协同式中订阅上游服务的事件减少了服务间对事件协议的依赖而只需要关心集权的编排器类发送的消息。MassTransit CourierMassTransit Courier是一种用于创建和执行带有故障补偿的分布式事务的机制它可以用于满足本地事务的需求也可以在分布式系统中实现分布式事务。Courier实现了Routing Slip模式通过有序组合一系列的Activity得到一个Routing slip。每个Activity都有 Execute 和 Compensate 两个方法(最后一个可以只有一个Execute方法)。Compensate即为补偿操作。补偿服务当开启一个事务前需要做一些准备准备一个事务Id记录整个事务执行情况各Tj事务执行情况当前请求上下文参数入参参数记录等以方便执行补偿操作时需要用到。如当Tj事务执行失败时需要对Cj-1到C1执行补偿操作此时各补偿操作需要一些正向执行T1Tj-1的请求参数或执行结果因此都需要记录下来。在Courier中,通过Routing Slip来完成这些记录创建一个Guid记录请求上下文参数信息可以绑定几个内置事件在各阶段到来时会发送事件如有需要可以订阅。var builder new RoutingSlipBuilder(NewId.NextGuid());
builder.AddSubscription(context.ReceiveContext.InputAddress, RoutingSlipEvents.Completed | RoutingSlipEvents.Faulted | RoutingSlipEvents.CompensationFailed);
builder.AddVariable(RequestId, context.RequestId);
builder.AddVariable(ResponseAddress, context.ResponseAddress);
builder.AddVariable(FaultAddress, context.FaultAddress);
builder.AddVariable(Request, context.Message);
//组合一系列Activity
var routingSlip builder.Build();
await context.Execute(routingSlip).ConfigureAwait(false);
服务建立弄了个Demo建立了三个服务此处我使用编排式来完成但无论是选用编排式还是协同式都借助RabbitMQ实现消息传递。每个服务都安装了MassTransit相关的包MassTransit.AspNetCore
MassTransit.RabbitMQ
将Saga编排器类放置在OrderService中了对于编排器类的放置个人认为是应该看用例的主服务是谁而放置想过放在BFF去协调三个服务但是总是感觉不是BFF的职责范围。服务配置在各服务中对MassTransit配置如下在OrderService中对MassTransit需要使用到的RabbitMQ配置对需要进行多个服务协作的用例配置Routing Slip对消息队列侦听订阅需要的事件并配置相应的Activity处理。services.AddMassTransit(x
{var currentAssembly Assembly.GetExecutingAssembly();x.AddActivities(currentAssembly);x.AddConsumers(currentAssembly);x.AddRequestClientcreateordercommand();x.UsingRabbitMq((context, cfg) {// 配置RabbitMQcfg.Host(Configuration[RabbitmqConfig:HostIP], ushort.Parse(Configuration[RabbitmqConfig:HostPort]), Configuration[RabbitmqConfig:VirtualHost], h {h.Username(Configuration[RabbitmqConfig:Username]);h.Password(Configuration[RabbitmqConfig:Password]);});//配置Routing Slipcfg.ReceiveEndpoint(CreateOrderCommand, ep {ep.ConfigureConsumercreateorderrequestproxy(context);ep.ConfigureConsumercreateorderresponseproxy(context);});// 配置订阅队列及Handler处理cfg.ReceiveEndpoint(CreateOrder_execute, ep {ep.ExecuteActivityHostcreateorderactivity, createordermodel(context);});});
});
services.AddMassTransitHostedService();
服务编排构建Routing Slip此处依据用例的需求对需要协作的服务编排组合一系列的Activity。Task BuildRoutingSlip(RoutingSlipBuilder builder, ConsumeContextcreateordercommand request)
{builder.AddActivity(ReduceStock, new Uri(...), new {});builder.AddActivity(DeductBalance, new Uri(...), new {});builder.AddActivity(CreateOrder, new Uri(...), new { });return Task.CompletedTask;
}
执行请求当请求进入后通过RequestClient发送CreateOrderCommand同步等待执行结果再由编排器类负责协调预设好的Activity发送事件到消息队列经各Activity订阅处理最终返回结果。[Route([controller])]
public class OrderController : ControllerBase
{private readonly IRequestClientcreateordercommand _createOrderClient;public OrderController(IRequestClientcreateordercommand createOrderClient){_createOrderClient createOrderClient;}[HttpGet(CreateOrder)]public async Taskcommoncommandresponsecreateorderresult CreateOrder(){var result await _createOrderClient.GetResponse
commoncommandresponsecreateorderresult(new CreateOrderCommand(){// ...});return result.Message;}
}
各服务中对于Activity设置侦听队列以及请求信息调用Execute执行逻辑当出现异常时返回到MQ通知编排器类在对之前执行的Activity执行Compensate。如在CreateOrderActivity中执行异常由编排器类执行补偿ReduceStockActivity调用Compensate执行增加库存逻辑public class ReduceStockActivity : IActivityReduceStockModel, ReduceStockLog
{public async TaskExecutionResult Execute(ExecuteContextReduceStockModel context){var argument context.Arguments;// 扣减库存await Task.Delay(100);return context.Completed(new ReduceStockLog() { ProductId argument.ProductId, Amount 1 });}public async TaskCompensationResult Compensate(CompensateContextReduceStockLog context){// 增加库存await Task.Delay(100);return context.Compensated();}
}
执行成功用例请求执行后先由Controller发送请求再由库存服务扣减库存支付服务扣减余额最后由订单服务创建订单当创建失败时执行补偿操作库存服务增加库存支付服务增加余额。执行补偿用例请求执行后先由Controller发送请求再由库存服务扣减库存支付服务扣减余额最后由订单服务创建订单当创建失败时执行补偿操作库存服务增加库存支付服务增加余额。在整个事务失败后先会返回异常再由编排器执行补偿操作实现最终的数据一致性。MassTransit也提供了重试机制以实现向前恢复避免因数据库连接超时、网络波动等问题造成的失败。Demo参考Masstransit中的 Request/Response 与 Courier 功能实现最终一致性 - 丁松松松理解分布式事务 (juejin.cn)-陈彩华
