AspNetCoreMassTransit Courier实现分布式事务

在之前的一篇博文中,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 Courier

MassTransit Courier是一种用于创建和执行带有故障补偿的分布式事务的机制,它可以用于满足本地事务的需求,也可以在分布式系统中实现分布式事务。

Courier实现了Routing Slip模式,通过有序组合一系列的Activity,得到一个Routing slip。每个Activity都有 Execute 和 Compensate 两个方法(最后一个可以只有一个Execute方法)。Compensate即为补偿操作。

补偿服务

当开启一个事务前,需要做一些准备,准备一个事务Id,记录整个事务执行情况,各Tj事务执行情况,当前请求上下文参数,入参参数记录等,以方便执行补偿操作时需要用到。如当Tj事务执行失败时,需要对Cj-1到C1执行补偿操作,此时各补偿操作需要一些正向执行T1,Tj-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.AddRequestClient<createordercommand>();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.ConfigureConsumer<createorderrequestproxy>(context);ep.ConfigureConsumer<createorderresponseproxy>(context);});// 配置订阅队列及Handler处理cfg.ReceiveEndpoint("CreateOrder_execute", ep =>{ep.ExecuteActivityHost<createorderactivity, createordermodel="">(context);});});
});
services.AddMassTransitHostedService();

服务编排

构建Routing Slip,此处依据用例的需求,对需要协作的服务编排,组合一系列的Activity。

Task BuildRoutingSlip(RoutingSlipBuilder builder, ConsumeContext<createordercommand> 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 IRequestClient<createordercommand> _createOrderClient;public OrderController(IRequestClient<createordercommand> createOrderClient){_createOrderClient = createOrderClient;}[HttpGet("CreateOrder")]public async Task<commoncommandresponse<createorderresult>> CreateOrder(){var result = await _createOrderClient.GetResponse
<commoncommandresponse<createorderresult>>(new CreateOrderCommand(){// ...});return result.Message;}
}

各服务中对于Activity设置侦听队列以及请求信息,调用Execute执行逻辑,当出现异常时返回到MQ通知编排器类,在对之前执行的Activity执行Compensate。如在CreateOrderActivity中执行异常,由编排器类执行补偿,ReduceStockActivity调用Compensate,执行增加库存逻辑

public class ReduceStockActivity : IActivity<ReduceStockModel, ReduceStockLog>
{public async Task<ExecutionResult> Execute(ExecuteContext<ReduceStockModel> context){var argument = context.Arguments;// 扣减库存await Task.Delay(100);return context.Completed(new ReduceStockLog() { ProductId = argument.ProductId, Amount = 1 });}public async Task<CompensationResult> Compensate(CompensateContext<ReduceStockLog> context){// 增加库存await Task.Delay(100);return context.Compensated();}
}

执行成功

用例请求执行后,先由Controller发送请求,再由库存服务扣减库存,支付服务扣减余额,最后由订单服务创建订单,当创建失败时,执行补偿操作,库存服务增加库存,支付服务增加余额。

执行补偿

用例请求执行后,先由Controller发送请求,再由库存服务扣减库存,支付服务扣减余额,最后由订单服务创建订单,当创建失败时,执行补偿操作,库存服务增加库存,支付服务增加余额。

在整个事务失败后,先会返回异常,再由编排器执行补偿操作,实现最终的数据一致性。MassTransit也提供了重试机制以实现向前恢复,避免因数据库连接超时、网络波动等问题造成的失败。

Demo

参考

Masstransit中的 Request/Response 与 Courier 功能实现最终一致性 - 丁松松松

理解分布式事务 (juejin.cn)-陈彩华

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/300450.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

可能会紧急用到的Linux命令

查看当前网络连接数&#xff1a; netstat -na|grep ESTABLISHED|awk {print $5}|awk -F: {print $1}|sort|uniq -c|sort -r -n删除掉不以java和xml结尾7天没有使用的文件 : find . ! -name *.java ! -name *.xml -atime 7 -exec rm {} \;转载于:https://blog.51cto.com/guihai…

设置列表字段为主键

转贴:Sample event handler to set a field as a pr imary key (enforce no duplicates) Got this as a request from a reader- how to prevent users from adding items with same titles as ones that already exist in the list. Codeusing System;using System.Collectio…

被学校辞退、拒绝FB后:语音识别大牛Povey确认兼职北京初创公司,称主业还选中国...

全世界只有3.14 % 的人关注了青少年数学之旅由于5月的学生抗议事件&#xff0c;语音识别领域著名学者、原约翰霍普金斯大学教授Daniel Povey被学校辞退。随后&#xff0c;Daniel Povey准备进入Facebook从事语音识别系统的开发&#xff0c;但是由于Facebook要对其进行长达6周的审…

也谈子网划分和子网通信

子网划分和子网通信在小型网络中应用不多&#xff0c;但随着公司规模的扩大&#xff0c;这样的问题就会提上网管的工作计划。资料略加整理&#xff0c;希望对各位有帮助。也谈子网划分和子网通信1.网络IP地址和子网掩码的分类&#xff1a;A类&#xff1a;IP: 0.0.0.0 &#xff…

java io流学设置编码_Java学习日志(21-2-IO流-基本数据类型与字节数组对象与、编码解码)...

操作基本数据类型的流对象DataStream/*可以用于操作基本数据类型数据的流对象*/import java.io.*;class DataStreamDemo{public static void main(String[] args)throws IOException{// writeData();// readData();// writeUTFDemo();// OutputStreamWriter oswnew OutputStrea…

dotnet中的counters说明(二)

上篇说了System.Runtime&#xff0c;它负责应用运行的环境资源的收集&#xff0c;这篇要继续说AspNetCore的Hosting,Http.Connections和Server.Kestrel三个计数器。同时&#xff0c;下面指标各项()里的项目是--counters 参数[]里的项&#xff0c;用逗号分隔多项指标。Microsoft…

Asp组件中级入门与精通系列之五

我们学习来看一下Response对象。其实我们前面的教程中一直都在使用这个对象的Write方法。 这里我们用Response对象设置cookie。 ? 打开vb6,新建Activex Dll工程。工程名修改为fCom,类名修改为fZ5 引用“Microsoft Active Server Pages Object”对象库。 创建两个组件事件&…

有这些好习惯,可以让你悄悄变优秀

全世界只有3.14 %的人关注了青少年数学之旅这是一个普遍无趣的时代&#xff0c;很多人看似忙到起飞内在却空虚迷茫。今天我们为你诚意推荐几个公众号它们会成为你生活的一剂调味料&#xff0c;让你做一个学识丰富、灵魂有趣的人。快来关注&#xff0c;开启精彩的生活吧&#xf…

java设置access-allow_Java Web如何设置多个Access-Control-Allow-Origin

有没有办法让Access-Control-Allow-Origin header允许设置multiple cross-domains呢&#xff1f;如果设置response.addHeader("Access-Control-Allow-Origin","*");感觉这个接口太开放了&#xff0c;不太安全。 我想只设置自己指定的若干个域名或者端口可以…

paip.提升用户体验--提升java的热部署热更新能力

paip.提升用户体验--提升java的热部署热更新能力 想让java做到php那么好的热部署能力 "fix online"/在线修复吗&#xff1f;&#xff1f;直接在服务器上修改源码生效&#xff0c;无需重启应用。。作者Attilax 艾龙&#xff0c; EMAIL:1466519819qq.com来源&#x…

夏利车电动窗功能设定

夏利车电动窗功能设定 我车的前有门电动车窗坏了&#xff0c;所以去汽配城购买了电动升降器&#xff0c;准备自己进行更换&#xff0c;修车的要我100元RMB&#xff0c;自己购买配件花了我65省了35元呵呵&#xff0c; 但是自己花了很长时间更换完成之后发现车窗怎么也关不严&…

微软 MS Learn 上线 Blazor 入门教程

微软官方学习网站 MS Learn 上线了 Blazor 入门教程模块&#xff0c;希望通过这个课程&#xff0c;让开发人员了解如何设置开发环境&#xff0c;以及如何使用 Blazor、Visual Studio Code 和 C# 生成你的首个 Web 应用。Build a web app with Blazor - Learndocs.microsoft.com…

Type Casting

Type Casting C : Documents : C Language Tutorial : Type Casting Search: userpass[register] javascript and cookies required C Language TutorialIntroduction?Instructions for useBasics of C?Structure of a program?Variables. Data Types.?Constants?Oper…

世界最牛实验室,堪称诺贝尔奖孵化器!到底是个怎样神奇的存在?!

▲ 点击查看随着诺贝尔各个奖项陆陆续续的公布&#xff0c;卡文迪许实验室&#xff0c;又开始重回大众视野。在这个世界最牛实验室之一的实验室里&#xff0c;仅仅过去了一百多年&#xff0c;就不断涌现出一批又一批世界一流的科学家&#xff1a;把电与磁进行有机统一的麦克斯…

java接口那一节是哪的知识_Java中的接口知识汇总

Java中的接口知识汇总发布于 2020-4-29|复制链接本文给大家汇总介绍了在java中的接口知识&#xff0c;包括为什么要使用接口、什么是接口、抽象类和接口的区别、如何定义接口以及定义接口注意点&#xff0c;希望大家能够喜欢一.为什么要使用接口 假如有一个需求&#xff1a;要求…

用C语言实现解析简单配置文件的小工具

本文介绍作者写的一个小工具&#xff0c;简单的代码中包含了C语言对字符串的处理技巧&#xff0c;对文本文件的简单解析&#xff0c;二进制文件的数据复制的方法&#xff0c;以及格式化输出文本文件的示例。 工具的输入是如下内容的配置文件&#xff1a; [plain] view plaincop…

Delphi应用程序在命令行下带参数执行返回命令行提示的问题

在命令行模式&#xff08;CMD&#xff09;下执行时&#xff0c;想获得执行参数&#xff0c;用以下变量&#xff1a; ParamCount&#xff1a;参数个数 ParamStr&#xff1a;为参数数组 如果想在执行完一个操作后在命令行作出相应提示&#xff0c;就应该在相应位置放入…

开源的负载测试/压力测试工具 NBomber

负载测试和压力测试对于确保 web 应用的性能和可缩放性非常重要。尽管它们的某些测试是相同的&#xff0c;但目标不同。负载测试&#xff1a;测试应用是否可以在特定情况下处理指定的用户负载&#xff0c;同时仍满足响应目标。应用在正常状态下运行。压力测试&#xff1a;在极端…

人生失败的31种致命原因

人生失败的31种致命原因 一、不利的遗传背景。天生智力不足的人&#xff0c;是没什么办法可想的。唯一的补救办法就是“以勤补拙”。 二、缺乏明确的人生目标&#xff0c;凡是没有明确人生目标的人&#xff0c;便没有成功的希望&#xff0c;在我曾经分析过的100人中&#x…

男人都应该懂的一张图。。 | 今日趣图

全世界只有3.14 % 的人关注了青少年数学之旅美国人为了教民众如何辨别韩国人制作的韩国女性标准照左右军事成为一个有钱人的概率有多高&#xff1f;最新版男人都该懂的汽车品牌从属关系图twi:NOCO_1002肥胖和骨架没有必然联系800斤胖子的X射线照 科普君XueShu 雪树来猜一猜这是…