公司网站怎么备案,电动车行业网站建设,WordPress小工具吗美化,零食网站源码目录 分布式事务基础事务本地事务分布式事务分布式事务的场景 分布式事务解决方案全局事务/两阶段提交可靠消息服务最大努力通知TCC事务 Seata介绍Seata实现分布式事务控制案例基本代码修改order微服务OrderSeataControllerOrderServiceImpl5注释容错相关代码ProductClient 修改… 目录 分布式事务基础事务本地事务分布式事务分布式事务的场景 分布式事务解决方案全局事务/两阶段提交可靠消息服务最大努力通知TCC事务 Seata介绍Seata实现分布式事务控制案例基本代码修改order微服务OrderSeataControllerOrderServiceImpl5注释容错相关代码ProductClient 修改Product微服务ProductControllerProductServiceImpl 测试正常下订单和扣库存异常模拟 启动Seata下载seata修改配置文件初始化seata在nacos的配置启动seata服务 使用Seata实现事务控制初始化数据表添加配置添加依赖DataSourceProxyConfigregistry.confbootstrap.yaml 在order微服务开启全局事务 seata运行流程分析 分布式事务基础
事务
事务指的就是一个操作单元在这个操作单元中的所有操作最终要保持一致的行为要么所有操作都成功要么所有的操作都被撤销。简单地说事务提供一种“要么什么都不做要么做全套”机制。
本地事务
本地事物其实可以认为是数据库提供的事务机制。说到数据库事务就不得不说数据库事务中的四大特性:
A原子性(Atomicity)一个事务中的所有操作要么全部完成要么全部不完成 C一致性(Consistency)在一个事务执行之前和执行之后数据库都必须处于一致性状态 I隔离性(Isolation)在并发环境中当不同的事务同时操作相同的数据时事务之间互不影响 D持久性(Durability)指的是只要事务成功结束它对数据库所做的更新就必须永久的保存下来
数据库事务在实现时会将一次事务涉及的所有操作全部纳入到一个不可分割的执行单元该执行单元中的所有操作要么都成功要么都失败只要其中任一操作执行失败都将导致整个事务的回滚
分布式事务
分布式事务指事务的参与者、支持事务的服务器、资源服务器以及事务管理器分别位于不同的分布式系统的不同节点之上。
简单的说就是一次大的操作由不同的小操作组成这些小的操作分布在不同的服务器上且属于不同的应用分布式事务需要保证这些小操作要么全部成功要么全部失败。
本质上来说分布式事务就是为了保证不同数据库的数据一致性。
分布式事务的场景
1.单体系统访问多个数据库 一个服务需要调用多个数据库实例完成数据的增删改操作 2.多个微服务访问同一个数据库 多个服务需要调用一个数据库实例完成数据的增删改操作 3.多个微服务访问多个数据库 多个服务需要调用一个数据库实例完成数据的增删改操作
分布式事务解决方案
全局事务/两阶段提交 全局事务基于DTP模型实现。DTP是由X/Open组织提出的一种分布式事务模型——X/Open Distributed Transaction Processing Reference Model。它规定了要实现分布式事务需要三种角色
AP: Application 应用系统 (微服务) TM: Transaction Manager 事务管理器 (全局事务管理) RM: Resource Manager 资源管理器 (数据库)
整个事务分成两个阶段:
阶段一: 表决阶段所有参与者都将本事务执行预提交并将能否成功的信息反馈发给协调者。 阶段二: 执行阶段协调者根据所有参与者的反馈通知所有参与者步调一致地执行提交或者回滚。 优点 提高了数据一致性的概率实现成本较低
缺点 单点问题: 事务协调者宕机 同步阻塞: 延迟了提交时间加长了资源阻塞时间 数据不一致: 提交第二阶段依然存在commit结果未知的情况有可能导致数据不一致
可靠消息服务
基于可靠消息服务的方案是通过消息中间件保证上、下游应用数据操作的一致性。假设有A和B两个系统分别可以处理任务A和任务B。此时存在一个业务流程需要将任务A和任务B在同一个事务中处理。就可以使用消息中间件来实现这种分布式事务。 第一步: 消息由系统A投递到中间件
1.在系统A处理任务A前首先向消息中间件发送一条消息 2.消息中间件收到后将该条消息持久化但并不投递。持久化成功后向A回复一个确认应答 3.系统A收到确认应答后则可以开始处理任务A 4.任务A处理完成后向消息中间件发送Commit或者Rollback请求。该请求发送完成后对系统A而言该事务的处理过程就结束了 5.如果消息中间件收到Commit则向B系统投递消息如果收到Rollback则直接丢弃消息。但是如果消息中间件收不到Commit和Rollback指令那么就要依靠超时询问机制。
超时询问机制系统A除了实现正常的业务流程外还需提供一个事务询问的接口供消息中间件调用。当消息中间件收到发布消息便开始计时如果到了超时没收到确认指令就会主动调用系统A提供的事务询问接口询问该系统目前的状态。该接口会返回三种结果中间件根据三种结果做出不同反应提交:将该消息投递给系统B回滚:直接将条消息丢弃处理中:继续等待第二步: 消息由中间件投递到系统B
消息中间件向下游系统投递完消息后便进入阻塞等待状态下游系统便立即进行任务的处理任务处理完成后便向消息中间件返回应答。
如果消息中间件收到确认应答后便认为该事务处理完毕 如果消息中间件在等待确认应答超时之后就会重新投递直到下游消费者返回消费成功响应为止。一般消息中间件可以设置消息重试的次数和时间间隔如果最终还是不能成功投递则需要手工干预。这里之所以使用人工干预而不是使用让系统回滚主要是考虑到整个系统设计的复杂度问题。
基于可靠消息服务的分布式事务前半部分使用异步注重性能后半部分使用同步注重开发成本。
最大努力通知
最大努力通知也被称为定期校对其实是对第二种解决方案的进一步优化。它引入了本地消息表来记录错误消息然后加入失败消息的定期校对功能来进一步保证消息会被下游系统消费。 第一步: 消息由系统A投递到中间件 1.处理业务的同一事务中向本地消息表中写入一条记录 2.准备专门的消息发送者不断地发送本地消息表中的消息到消息中间件如果发送失败则重试
第二步: 消息由中间件投递到系统B 1.消息中间件收到消息后负责将该消息同步投递给相应的下游系统并触发下游系统的任务执行 2.当下游系统处理成功后向消息中间件反馈确认应答消息中间件便可以将该条消息删除从而该事务完成 3.对于投递失败的消息利用重试机制进行重试对于重试失败的写入错误消息表 4.消息中间件需要提供失败消息的查询接口下游系统会定期查询失败消息并将其消费
这种方式的优缺点 优点 一种非常经典的实现实现了最终一致性。 缺点 消息表会耦合到业务系统中如果没有封装好的解决方案会有很多杂活需要处理。
TCC事务
TCC即为Try Confirm Cancel它属于补偿型分布式事务。TCC实现分布式事务一共有三个步骤
Try尝试待执行的业务 这个过程并未执行业务只是完成所有业务的一致性检查并预留好执行所需的全部资源
Confirm确认执行业务 确认执行业务操作不做任何业务检查 只使用Try阶段预留的业务资源。通常情况下采用TCC则认为 Confirm阶段是不会出错的。即只要Try成功Confirm一定成功。若Confirm阶段真的出错了需引入重试机制或人工处理。
Cancel取消待执行的业务 取消Try阶段预留的业务资源。通常情况下采用TCC则认为Cancel阶段也是一定成功的。若Cancel阶段真的出错了需引入重试机制或人工处理。 TCC两阶段提交与XA两阶段提交的区别是 XA是资源层面的分布式事务强一致性在两阶段提交的整个过程中一直会持有资源的锁。 TCC是业务层面的分布式事务最终一致性不会一直持有资源的锁。
TCC事务的优缺点 优点把数据库层的二阶段提交上提到了应用层来实现规避了数据库层的2PC性能低下问题。 缺点TCC的Try、Confirm和Cancel操作功能需业务提供开发成本高。
Seata介绍
2019 年 1 月阿里巴巴中间件团队发起了开源项目 FescarFast EaSy Commit And Rollback其愿景是让分布式事务的使用像本地事务的使用一样简单和高效并逐步解决开发者们遇到的分布式事务方面的所有难题。后来更名为 Seata意为Simple Extensible Autonomous Transaction Architecture是一套分布式事务解决方案。
Seata的设计目标是对业务无侵入因此从业务无侵入的2PC方案着手在传统2PC的基础上演进。它把一个分布式事务理解成一个包含了若干分支事务的全局事务。全局事务的职责是协调其下管辖的分支事务达成一致要么一起成功提交要么一起失败回滚。此外通常分支事务本身就是一个关系数据库的本地事务。 Seata主要由三个重要组件组成 TCTransaction Coordinator 事务协调器管理全局的分支事务的状态用于全局性事务的提交和回滚。 TMTransaction Manager 事务管理器用于开启、提交或者回滚全局事务。 RMResource Manager 资源管理器用于分支事务上的资源管理向TC注册分支事务上报分支事务的状态接受TC的命令来提交或者回滚分支事务。 Seata的执行流程如下: 1.A服务的TM向TC申请开启一个全局事务TC就会创建一个全局事务并返回一个唯一的XID 2.A服务的RM向TC注册分支事务并及其纳入XID对应全局事务的管辖 3.A服务执行分支事务向数据库做操作 4.A服务开始远程调用B服务此时XID会在微服务的调用链上传播 5.B服务的RM向TC注册分支事务并将其纳入XID对应的全局事务的管辖 6.B服务执行分支事务向数据库做操作 7.全局事务调用链处理完毕TM根据有无异常向TC发起全局事务的提交或者回滚 8.TC协调其管辖之下的所有分支事务 决定是否回滚回滚即根据undolog日志把数据都改回去
Seata实现2PC与传统2PC的差别 1.架构层次方面传统2PC方案的 RM 实际上是在数据库层RM本质上就是数据库自身通过XA协议实现而 Seata的RM是以jar包的形式作为中间件层部署在应用程序这一侧的。 2.两阶段提交方面传统2PC无论第二阶段的决议是commit还是rollback事务性资源的锁都要保持到Phase2完成才释放。而Seata的做法是在Phase1 就将本地事务提交这样就可以省去Phase2持锁的时间整体提高效率。
Seata实现分布式事务控制
本示例通过Seata中间件实现分布式事务模拟电商中的下单和扣库存的过程 我们通过订单微服务执行下单操作然后由订单微服务调用商品微服务扣除库存
案例基本代码
修改order微服务
OrderSeataController /*** 模拟分布式事务*/RequestMapping(/order/product7/{pid})public Order createOrder7(PathVariable(pid) Integer pid) {Order order orderServiceImpl5.createOrder(pid);return order;}OrderServiceImpl5 public Order createOrder(Integer pid) {log.info(接收到{}号商品的下单请求接下来调用商品微服务查询此商品信息, pid);// 1.调用商品微服务查询商品信息Product product productClient.getProductByPid(pid);log.info(查询到{}号商品的信息内容是{}, pid, JSON.toJSONString(product));// 2.下单/创建订单Order order new Order();order.setUid(1);order.setUsername(测试用户);order.setPid(product.getPid());order.setPname(product.getPname());order.setPprice(product.getPprice());order.setNumber(1);orderDao.save(order);log.info(创建订单成功订单信息为{}, JSON.toJSONString(order));// 3.扣库存新增代码productClient.reduceInventory(pid, order.getNumber());//4.下单成功之后,将消息放到mq中 向mq中投递一个下单成功的消息// 参数一指定topic 参数二指定消息体
// rocketMQTemplate.convertAndSend(order-topic, order);SendResult sendResult rocketMQTemplate.syncSend(order-topic, JSON.toJSONString(order), 10000);System.out.println(sendResult);
// orderServiceImpl4.createOrderBefore(order);return order;}注释容错相关代码
fallbackfallbackFactory
ProductClient //扣减库存//参数一商品标识 参数二扣减数量RequestMapping(/product/reduceInventory)void reduceInventory(RequestParam(pid) Integer pid, RequestParam(number)Integer number);修改Product微服务
ProductController // 扣减库存RequestMapping(/product/reduceInventory)public void reduceInventory(Integer pid, Integer number) {productService.reduceInventory(pid, number);}ProductServiceImpl Overridepublic void reduceInventory(Integer pid, Integer number) {//查询Product product productDao.findById(pid).get();//省略校验//内存中扣减product.setStock(product.getStock()-number);//保存productDao.save(product);}测试正常下订单和扣库存
浏览器访问http://localhost:8091/order/product7/1然后查看订单表和库存表
异常模拟
在ProductServiceImpl的代码中模拟一个异常, 然后调用下单接口 TransactionalOverridepublic void reduceInventory(Integer pid, Integer number) {//查询Product product productDao.findById(pid).get();//省略校验//内存中扣减product.setStock(product.getStock()-number);//模拟异常int i 1 / 0;//保存productDao.save(product);}可发现订单已下但是库存并没有减这就是一个典型的分布式事务问题
启动Seata
即seata serverTC单独跑的一个应用
下载seata
下载地址https://github.com/seata/seata/releases/v0.9.0
修改配置文件
将下载得到的压缩包进行解压进入conf目录调整下面的配置文件
registry.conf
registry {type nacos nacos {serverAddr localhost namespace public cluster default}
}config {type nacos nacos {serverAddr localhost namespace public cluster default}
}nacos-config.txt 替换service.vgroup_mapping.my_test_tx_groupdefault为以下
service.vgroup_mapping.service-productdefault
service.vgroup_mapping.service-orderdefault这里的语法为 service.vgroup_mapping.${your-service-gruop}default 中间的 ${your-service-gruop} 为自己定义的服务组名称 这里需要我们在程序的配置文件中配置。
初始化seata在nacos的配置
# 初始化seata 的nacos配置一批数据
# 注意: 这里要保证nacos是已经正常运行的
cd conf
nacos-config.sh 127.0.0.1启动seata服务
cd bin
seata-server.bat -p 9000 -m file启动后会注册到nacos在 Nacos 的服务列表下面可以看到一个名为 serverAddr 的服务。
至此TC已准备好接下来去应用中添加TM和RM以依赖包的形式加入服务中
使用Seata实现事务控制
初始化数据表
在我们的数据库中加入一张undo_log表,这是Seata记录事务日志要用到的表
CREATE TABLE undo_log (id BIGINT ( 20 ) NOT NULL AUTO_INCREMENT,branch_id BIGINT ( 20 ) NOT NULL,xid VARCHAR ( 100 ) NOT NULL,context VARCHAR ( 128 ) NOT NULL,rollback_info LONGBLOB NOT NULL,log_status INT ( 11 ) NOT NULL,log_created DATETIME NOT NULL,log_modified DATETIME NOT NULL,ext VARCHAR ( 100 ) DEFAULT NULL,PRIMARY KEY ( id ),
UNIQUE KEY ux_undo_log ( xid, branch_id )
) ENGINE INNODB AUTO_INCREMENT 1 DEFAULT CHARSET utf8;添加配置
在需要进行分布式控制的微服务中进行下面几项配置
添加依赖
dependencygroupIdcom.alibaba.cloud/groupIdartifactIdspring-cloud-starter-alibaba-seata/artifactId
/dependencyDataSourceProxyConfig
Seata 是通过代理数据源实现事务分支的所以需要配置 io.seata.rm.datasource.DataSourceProxy 的 Bean且是 Primary默认的数据源否则事务不会回滚无法实现分布式事务
Configuration
public class DataSourceProxyConfig {BeanConfigurationProperties(prefix spring.datasource)public DruidDataSource druidDataSource() {return new DruidDataSource();}PrimaryBeanpublic DataSourceProxy dataSource(DruidDataSource druidDataSource) {return new DataSourceProxy(druidDataSource);}
}registry.conf
在resources下添加Seata的配置文件 registry.conf
bootstrap.yaml
添加namespacegrouptx-service-group
spring:application:name: service-product/ordercloud:nacos:config:server-addr: 127.0.0.1:8848 #nacos中心地址namespace: publicgroup: SEATA_GROUPalibaba:seata:tx-service-group: service-product/order # 和nacos-config.txt里的命名对应在order微服务开启全局事务
OrderServiceImpl5
GlobalTransactional//全局事务控制
public Order createOrder(Integer pid) {}浏览器下单和扣库存测试http://localhost:8091/order/product7/2
seata运行流程分析 要点说明
1、每个RM使用DataSourceProxy连接数据库其目的是使用ConnectionProxy使用数据源和数据连接代理的目的就是在第一阶段将undo_log和业务数据放在一个本地事务提交这样就保存了只要有业务操作就一定有undo_log。
2、在第一阶段undo_log中存放了数据修改前和修改后的值为事务回滚作好准备所以第一阶段完成就已经将分支事务提交也就释放了锁资源。
3、TM开启全局事务开始将XID全局事务id放在事务上下文中通过feign调用也将XID传入下游分支事务每个分支事务将自己的Branch ID分支事务ID与XID关联。
4、第二阶段全局事务提交TC会通知各各分支参与者提交分支事务在第一阶段就已经提交了分支事务这里各各参与者只需要删除undo_log即可并且可以异步执行第二阶段很快可以完成。
5、第二阶段全局事务回滚TC会通知各各分支参与者回滚分支事务通过 XID 和 Branch ID 找到相应的回滚日志通过回滚日志生成反向的 SQL 并执行以完成分支事务回滚到之前的状态如果回滚失败则会重试回滚操作。
