文章目录前言一、分布式事务基础1. 什么是事务2. 本地事务3. 分布式事务4. 分布式事务的场景二、分布式事务解决方案1. 全局事务2. 可靠消息服务3. 最大努力通知4. TCC事务三、Seata工作原理1. Seata介绍2. Seata架构3. Seata实现2PC与传统2PC的差别4. Seata流程实例分析四、Seata实现分布式事务控制1. 案例基本代码2. 启动Seata3. 使用Seata实现事务控制前言事务指的就是一个操作单元在这个操作单元中的所有操作最终要保持一致的行为要么所有操作都成功要么所有的操作都被撤销。简单地说事务提供一种“要么什么都不做要么做全套”机制。一、分布式事务基础1. 什么是事务事务指的就是一个操作单元在这个操作单元中的所有操作最终要保持一致的行为要么所有操作都成功要么所有的操作都被撤销。简单地说事务提供一种“要么什么都不做要么做全套”机制。2. 本地事务其实可以认为是数据库提供的事务机制。说到数据库事务就不得不说数据库事务中的四大特性ACID:A原子性(Atomicity)一个事务中的所有操作要么全部完成要么全部不完成。C一致性(Consistency)在一个事务执行之前和执行之后数据库都必须处于一致性状态。I隔离性(Isolation)在并发环境中当不同的事务同时操作相同的数据时事务之间互不影响。D持久性(Durability)指的是只要事务成功结束它对数据库所做的更新就必须永久的保存下来。数据库事务在实现时会将一次事务涉及的所有操作全部纳入到一个不可分割的执行单元该执行单元中的所有操作要么都成功要么都失败只要其中任一操作执行失败都将导致整个事务的回滚。3. 分布式事务分布式事务指事务的参与者、支持事务的服务器、资源服务器以及事务管理器分别位于不同的分布式系统的不同节点之上。简单的说就是一次大的操作由不同的小操作组成这些小的操作分布在不同的服务器上且属于不同的应用分布式事务需要保证这些小操作要么全部成功要么全部失败。本质上来说分布式事务就是为了保证不同数据库的数据一致性。4. 分布式事务的场景单体系统访问多个数据库一个服务需要调用多个数据库实例完成数据的增删改操作。多个微服务访问同一个数据库多个服务需要调用同一个数据库实例完成数据的增删改操作。多个微服务访问多个不同数据库多个服务需要调用不同数据库实例完成数据的增删改操作。二、分布式事务解决方案1. 全局事务全局事务基于DTP模型实现。DTP是由X/Open组织提出的一种分布式事务模型——X/Open** Distributed Transaction Processing Reference Model**。它规定了要实现分布式事务需要三种角色AP:Application 应用系统 (微服务)TM:Transaction Manager 事务管理器 (全局事务管理)RM: Resource Manager 资源管理器 (数据库)整个事务分成两个阶段阶段一:表决阶段所有参与者都将本事务执行预提交并将能否成功的信息反馈发给协调者。阶段二:执行阶段协调者根据所有参与者的反馈通知所有参与者步调一致地执行提交或者回滚。优点提高了数据一致性的概率实现成本较低。缺点单点问题: 事务协调者宕机。同步阻塞: 延迟了提交时间加长了资源阻塞时间。数据不一致: 提交第二阶段依然存在commit结果未知的情况有可能导致数据不一致。2. 可靠消息服务基于可靠消息服务的方案是通过消息中间件保证上、下游应用数据操作的一致性。假设有A和B两个系统分别可以处理任务A和任务B。此时存在一个业务流程需要将任务A和任务B在同一个事务中处理。就可以使用消息中间件来实现这种分布式事务。第一步: 消息由系统A投递到中间件在系统A处理任务A前首先向消息中间件发送一条消息。消息中间件收到后将该条消息持久化但并不投递。持久化成功后向A回复一个确认应答。系统A收到确认应答后则可以开始处理任务A。任务A处理完成后向消息中间件发送Commit或者Rollback请求。该请求发送完成后对系统A而言该事务的处理过程就结束了。如果消息中间件收到Commit则向B系统投递消息如果收到Rollback则直接丢弃消息。但是如果消息中间件收不到Commit和Rollback指令那么就要依靠超时询问机制。超时询问机制系统A除了实现正常的业务流程外还需提供一个事务询问的接口供消息中间件调用。当消息中间件收到发布消息便开始计时如果到了超时没收到确认指令就会主动调用系统A提供的事务询问接口询问该系统目前的状态。该接口会返回三种结果中间件根据三种结果做出不同反应提交:将该消息投递给系统B回滚:直接将条消息丢弃处理中:继续等待第二步: 消息由中间件投递到系统B消息中间件向下游系统投递完消息后便进入阻塞等待状态下游系统便立即进行任务的处理任务处理完成后便向消息中间件返回应答。如果消息中间件收到确认应答后便认为该事务处理完毕。如果消息中间件在等待确认应答超时之后就会重新投递直到下游消费者返回消费成功响应为止。一般消息中间件可以设置消息重试的次数和时间间隔如果最终还是不能成功投递则需要人工干预手动补偿。这里之所以使用人工干预而不是使用让系统回滚主要是考虑到整个系统设计的复杂度问题。基于可靠消息服务的分布式事务前半部分使用异步注重性能后半部分使用同步注重开发成本。3. 最大努力通知最大努力通知也被称为定期校对其实是对第二种解决方案可靠消息服务的进一步优化。它引入了本地消息表来记录错误消息然后加入失败消息的定期校对功能来进一步保证消息会被下游系统消费。第一步: 消息由系统A投递到中间件处理业务的同一事务中向本地消息表中写入一条记录。准备专门的消息发送者不断地发送本地消息表中的消息到消息中间件如果发送失败则重试。第二步: 消息由中间件投递到系统B消息中间件收到消息后负责将该消息同步投递给相应的下游系统并触发下游系统的任务执行。当下游系统处理成功后向消息中间件反馈确认应答消息中间件便可以将该条消息删除从而该事务完成。对于投递失败的消息利用重试机制进行重试对于重试失败的写入失败消息表。消息中间件需要提供失败消息的查询接口下游系统会定期查询失败消息并将其消费。这种方式的优缺点优点 一种非常经典的实现实现了最终一致性。缺点 消息表会耦合到业务系统中如果没有封装好的解决方案会提升开发成本。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工作原理1. Seata介绍2019 年 1 月阿里巴巴中间件团队发起了开源项目 FescarFast EaSy Commit And Rollback其愿景是让分布式事务的使用像本地事务的使用一样简单和高效并逐步解决开发者们遇到的分布式事务方面的所有难题。后来更名为 Seata意为Simple Extensible AutonomousTransaction Architecture是一套分布式事务解决方案。Seata的设计目标是对业务无侵入因此从业务无侵入的2PC方案着手在传统2PC的基础上演进。它把一个分布式事务理解成一个包含了若干分支事务的全局事务。全局事务的职责是协调其管辖下的分支事务达成一致要么一起成功提交要么一起失败回滚。此外通常分支事务本身就是一个关系数据库的本地事务。2. Seata架构Seata主要由三个重要组件组成TCTransaction Coordinator 事务协调器管理全局的分支事务的状态用于全局性事务的提交和回滚。TMTransaction Manager 事务管理器用于开启、提交或者回滚全局事务。RMResource Manager 资源管理器用于分支事务上的资源管理向TC注册分支事务上报分支事务的状态接受TC的命令来提交或者回滚分支事务。Seata的执行流程如下:A服务的TM向TC申请开启一个全局事务TC就会创建一个全局事务并返回一个唯一的XID。A服务的RM向TC注册分支事务并及其纳入XID对应全局事务的管辖。A服务执行分支事务向数据库做操作。A服务开始远程调用B服务此时XID会在微服务的调用链上传播。B服务的RM向TC注册分支事务并将其纳入XID对应的全局事务的管辖。B服务执行分支事务向数据库做操作。全局事务调用链处理完毕TM根据有无异常向TC发起全局事务的提交或者回滚。TC协调其管辖之下的所有分支事务 决定是否回滚。3. Seata实现2PC与传统2PC的差别架构层次方面传统2PC方案的 RM 实际上是在数据库层RM本质上就是数据库自身通过XA协议实现而Seata的RM是以jar包的形式作为中间件层部署在应用程序这一侧的。两阶段提交方面传统2PC无论第二阶段的决议是commit还是rollback事务性资源的锁都要保持到Phase2完成才释放。而Seata的做法是在Phase1 就将本地事务提交这样就可以省去Phase2持锁的时间整体提高效率。4. Seata流程实例分析要点说明每个RM使用DataSourceProxy连接数据库其目的是使用ConnectionProxy使用数据源和数据连接代理的目的就是在第一阶段将undo_log和业务数据放在一个本地事务提交这样就保存了只要有业务操作就一定有undo_log。在第一阶段undo_log中存放了数据修改前和修改后的值为事务回滚作好准备所以第一阶段完成就已经将分支事务提交也就释放了锁资源。TM开启全局事务开始将XID全局事务id放在事务上下文中通过feign调用也将XID传入下游分支事务每个分支事务将自己的Branch ID分支事务ID与XID关联。第二阶段全局事务提交TC会通知各各分支参与者提交分支事务在第一阶段就已经提交了分支事务这里各各参与者只需要删除undo_log即可并且可以异步执行第二阶段很快可以完成。第二阶段全局事务回滚TC会通知各各分支参与者回滚分支事务通过 XID 和 Branch ID 找到相应的回滚日志通过回滚日志生成反向的 SQL 并执行以完成分支事务回滚到之前的状态如果回滚失败则会重试回滚操作。四、Seata实现分布式事务控制本示例通过Seata中间件实现分布式事务模拟电商中的下单和扣库存的过程。我们通过订单微服务执行下单操作然后由订单微服务调用商品微服务扣除库存。1. 案例基本代码order微服务ControllerRestControllerSlf4jpublicclassOrderController{AutowiredprivateOrderServiceImplorderService;//下单RequestMapping(/order/prod/{pid})publicOrderorder(PathVariable(pid)Integerpid){log.info(接收到{}号商品的下单请求,接下来调用商品微服务查询此商品信息,pid);returnorderService.createOrder(pid);}}OrderServiceServiceSlf4jpublicclassOrderServiceImpl5{AutowiredprivateOrderDaoorderDao;AutowiredprivateProductServiceproductService;AutowiredprivateRocketMQTemplaterocketMQTemplate;GlobalTransactionalpublicOrdercreateOrder(Integerpid){//1 调用商品微服务,查询商品信息ProductproductproductService.findByPid(pid);log.info(查询到{}号商品的信息,内容是:{},pid,JSON.toJSONString(product));//2 下单(创建订单)OrderordernewOrder();order.setUid(1);order.setUsername(测试用户);order.setPid(pid);order.setPname(product.getPname());order.setPprice(product.getPprice());order.setNumber(1);orderDao.save(order);log.info(创建订单成功,订单信息为{},JSON.toJSONString(order));//3 扣库存productService.reduceInventory(pid,order.getNumber());//4 向mq中投递一个下单成功的消息rocketMQTemplate.convertAndSend(order-topic,order);returnorder;}}ProductServiceFeignClient(valueservice-product)publicinterfaceProductService{//减库存RequestMapping(/product/reduceInventory)voidreduceInventory(RequestParam(pid)Integerpid,RequestParam(num)intnum);}Product微服务Controller//减少库存RequestMapping(/product/reduceInventory)publicvoidreduceInventory(Integerpid,intnum){productService.reduceInventory(pid,num);}ServiceOverridepublicvoidreduceInventory(Integerpid,intnum){ProductproductproductDao.findById(pid).get();product.setStock(product.getStock()-num);//减库存productDao.save(product);}异常模拟正常系-超卖异常系-代码异常在ProductService实现类的方法代码中模拟一个异常, 然后调用下单接口OverridepublicvoidreduceInventory(Integerpid,Integernumber){ProductproductproductDao.findById(pid).get();if(product.getStock()number){thrownewRuntimeException(库存不足);}inti1/0;product.setStock(product.getStock()-number);productDao.save(product);}2. 启动Seata下载seata下载地址: Seata修改配置文件将下载得到的压缩包进行解压进入conf目录调整下面的配置文件registry.confregistry { type nacos nacos { serverAddr localhost namespace public cluster default } } config { type nacos nacos { serverAddr localhost namespace public cluster default } }nacos-config.txtservice.vgroup_mapping.service-productdefault service.vgroup_mapping.service-orderdefault这里的语法为 service.vgroup_mapping.${your-service-gruop}default 中间的${your-service-gruop} 为自己定义的服务组名称 这里需要我们在应用程序的配置文件中配置。初始化seata在nacos的配置# 初始化seata 的nacos配置# 注意: 这里要保证nacos是已经正常运行的cdconf nacos-config.sh127.0.0.1执行成功后可以打开Nacos的控制台在配置列表中可以看到初始化了很多Group为SEATA_GROUP的配置。启动seata服务# 初始化seata 的nacos配置# 注意: 这里要保证nacos是已经正常运行的cdbin seata-server.bat-p9000-mfile启动后在 Nacos 的服务列表下面可以看到一个名为 serverAddr 的服务。3. 使用Seata实现事务控制初始化数据表在我们的数据库中加入一张undo_log表,这是Seata记录事务日志要用到的表CREATETABLEundo_log(idBIGINT(20)NOTNULLAUTO_INCREMENT,branch_idBIGINT(20)NOTNULL,xidVARCHAR(100)NOTNULL,contextVARCHAR(128)NOTNULL,rollback_infoLONGBLOBNOTNULL,log_statusINT(11)NOTNULL,log_createdDATETIMENOTNULL,log_modifiedDATETIMENOTNULL,extVARCHAR(100)DEFAULTNULL,PRIMARYKEY(id),UNIQUEKEYux_undo_log(xid,branch_id))ENGINEINNODBAUTO_INCREMENT1DEFAULTCHARSETutf8;添加配置在需要进行分布式控制的微服务中进行下面几项配置:添加依赖dependencygroupIdcom.alibaba.cloud/groupIdartifactIdspring-cloud-starter-alibaba-seata/artifactId/dependencydependencygroupIdcom.alibaba.cloud/groupIdartifactIdspring-cloud-starter-alibaba-nacos-config/artifactId/dependencyDataSourceProxyConfigSeata 是通过代理数据源实现事务分支的所以需要配置io.seata.rm.datasource.DataSourceProxy 的 Bean且是Primary默认的数据源否则事务不会回滚无法实现分布式事务。ConfigurationpublicclassDataSourceProxyConfig{BeanConfigurationProperties(prefixspring.datasource)publicDruidDataSourcedruidDataSource(){returnnewDruidDataSource();}PrimaryBeanpublicDataSourceProxydataSource(DruidDataSourcedruidDataSource){returnnewDataSourceProxy(druidDataSource);}}registry.conf在resources下添加Seata的配置文件 registry.confregistry { type nacos nacos { serverAddr localhost namespace public cluster default } } config { type nacos nacos { serverAddr localhost namespace public cluster default } }bootstrap.yamlspring:application:name:service-productcloud:nacos:config:server-addr:localhost:8848# nacos的服务端地址namespace:publicgroup:SEATA_GROUPalibaba:seata:tx-service-group:${spring.application.name}在order微服务开启全局事务GlobalTransactional//全局事务控制publicOrdercreateOrder(Integerpid){}测试再次下单测试本文的引用仅限自我学习如有侵权请联系作者删除。参考知识传智教育·黑马程序员
