分布式事务及分布式框架Seata

分布式事务

==分布式事务是什么？

==》本地事务是一个单元的sql，分布式事务也是一个单元的sql，他们区别在于，分布式事务的sql分布在了不同服务上，这里的服务指微服务和数据库服务

==？为什么强调服务是微服务和数据库服务？

一个服务，多个数据库是分布式事务
多个服务，一个数据库是分布式事务
多个服务，多个数据库是分布式事务
总之，分布在不同服务上的不同的事务，它们各自rollback和commit，仅凭本地事务是无法保证它们一致性的

分布式事务解决方案

分布式事务的解决方案有三种，分别是全局事务、可靠消息服务、TCC事务

全局事务

全局事务基于DTP模型实现。DTP是由X/Open组织提出的一种分布式事务模型

全局事务中有三个重要角色

AP: Application 应用系统 (微服务)
TM: Transaction Manager 事务管理器 (全局事务管理)
RM: Resource Manager 资源管理器 (数据库)

整个事务分成两个阶段:

一阶段: 表决阶段，所有参与者都将本事务执行预提交，当且仅当参与者都成功无异常时commit，否则rollback
二阶段: 执行阶段，协调者根据所有参与者的反馈，通知所有参与者，步调一致地执行commit或者rollback。

优点

提高了数据一致性的概率，实现成本较低

缺点

单点问题: 事务协调者宕机
同步阻塞: 延迟了提交时间，加长了资源阻塞时间
数据不一致: 提交第二阶段，依然存在commit结果未知的情况，有可能导致数据不一致（概率较小，只有commit那一瞬间服务宕机才会出现这种情况）

可靠消息服务

全称叫做基于可靠消息的最终一致性事务方案

消息就是我们所用的消息中间件，也就是我们将通过消息中间件实现可靠消息服务解决分布式事务

这里以RocketMQ为消息中间件，订单服务调用积分服务为例

实现流程：

①订单服务发送中间状态（不可消费的消息）的消息到MQ

②MQ返回通知订单服务中间状态发送成功

③订单服务执行业务，若无异常则发送通知MQ已commit，有异常则通知MQ已rollback

④MQ接收订单服务的通知，为commit则标记信息为提交事务（即正常可消费消息），为rollback则删除半消息

⑤当消息变为全消息后，积分服务正常消费，如果积分服务不消费，且RocketMQ重试16次（消费方要做好幂等性处理）还是不成功就会认为消息消费不了，丢进死信队列（这就是所谓的可靠消息，也是MQ为了防止消息堆积的处理方案）。对于没有Commit/Rollback的事务消息，事务发起方将会到MQ中回查消息

事务消息共有三种状态，提交事务：它允许消费者消费此消息；回滚事务：它代表消息将被删除，不允许被消费；中间状态：它代表需要消息队列来确认状态

注意事项：

事务消息共有三种状态，提交事务：它允许消费者消费此消息， 回滚事务：它代表消息将被删除，不允许被消费， 中间状态：它代表需要消息队列来确认状态
可靠消息最终一致性就是保证消息从生产方经过消息中间件传递到消费方的一致性，如果消费方因为异常而rollback，生产方是无感知的，不会因为它进行rollback，因此可靠消息服务需要保证消费方（也即事务的下游）接收消息后一定要执行成功，否则全局事务一致性。

优点：

项目中使用MQ（或者其他消息中间件）时，可以直接配置解决分布式事务

缺点（局限性）：

不适用于要求实时性的事务（因为MQ就是延时性的操作，要求MQ实时性是无稽之谈）
必须保证消费方能够完成业务

TCC事务

TCC即为Try Confirm Cancel，它属于补偿型分布式事务。TCC实现分布式事务一共有三个步骤：

Try：尝试待执行的业务，这个过程并未执行业务，只是完成所有业务的一致性检查，并预留好执行所需的全部资源
Confirm ：确认执行业务，只要Try成功（没有异常），Confirm 一定成功。若Confirm 阶段真的出错了，需引入重试机制或人工处理。
Cancel：取消待执行的业务，释放Try预留的资源，只要Try失败（出现异常），Cancel阶段就一定成功。若Cancel阶段真的出错了，需引入重试机制或人工处理。

注意：这里说预留资源就是直接commit能成功的条件，如转账业务需要转出方有足够的钱，commit之后就等于预留了资源防止下一个线程消费。如果是收款方的话，即不需要预留资源操作，也就是说这种条件下try不需要也不应该进行资源预留操作的，因为如果commit之后其他线程就可能拿到足够的钱去执行其他操作了。

举几个例子

#积分下单业务订单服务                     |                 积分服务
try       update status "支付中"       |                 update intergral = intergral - #{intergral}|
confirm   update status "已支付"       |                 无|
cancel    update status "未支付"       |               update intergral = intergral + #{intergral}

#积分退款业务订单服务                     |                 积分服务
try       update status "退款中"       |                 无|
confirm   update status "已退款"       |                 update intergral = intergral + #{intergral}|
cancel    update status "已支付"       |               无

#转账业务A服务                                  |                 B服务
try       update money = money - #{money}       |                 无|
confirm   无                                     |                update money = money + #{money}   |
cancel    update money = money + #{money}       |               update money = money - #{money}

Try必须做资源预留（减操作）、一致性检查，如转账中，需要保证出款方的金额足够扣除

Confirm在Try成功后的执行，在转账中，出款方Try检查完毕后即可执行相同逻辑，因此为空

Cancel就是在Try检查失败后执行的操作，转账失败，预留的资源就要释放，也即回滚金额

左边为事务发起者，事务发起者本身的事务由本地事务保证即可，它会去引导其他事务参与者TCC

Seata分布式事务

Seata分布式事务框架支持有多种模式，如： AT、TCC、SAGA、XA。除了AT，后面三种分布式事务模式的大致流程相同，所以我们只要搞清楚AT和TCC流程是如何即可

Seata-AT（Automatic Transaction）模式

AT模式下，把每个数据库被当做是一个 Resource，Seata 里称为 DataSource Resource。

Seata的模块组成：

TC：Transaction Coordinator 事务协调器，管理全局的分支事务的状态，用于全局事务的提交和回滚。
TM：Transaction Manager 事务管理器，事务发起者用于开启全局事务。
RM：Resource Manager 资源管理器，分支事务上的资源管理，用于所有事务参与者（包括事务发起者）向TC注册分支事务，上报分支事务的状态，接受TC的命令来提交或者回滚分支事务。

Seata-AT的执行流程：

大致可以分为两个阶段：

一阶段：事务参与者异步完成commit事务并记录在undo_log日志中
二阶段：事务参与者同步完成commit或rollback操作，即删除undo_log日志或根据undo_log日志反向DML操作

①事务发起者（服务，也是事务参与者）的TM向TC发起全局事务，TC返回一个全局唯一标识XID（一个个全局事务组）

②事务发起者的RM向TC注册事务分支branch并被TC纳入XID，随后服务立即执行事务操作数据库

③事务发起者将XID传播到其他事务参与者（服务），事务参与者的RM向TC注册事务分支branch并被纳入XID，随后服务立即执行事务操作数据库

④所有事务在DML的时候会产生一个个undo_log日志文件（mysql事务的特点），操作完立即commit/rollback事务，每个分支事务commit成功则返回通知XID标记分支事务执行成功，rollback则标记为失败。真正执行服务各自commit/rollback操作的都是TM，说白了就是commit/rollback都要先通知TM，TM再帮忙完成操作

⑤待所有分支事务执行完毕，TM根据所有分支事务的情况，向TC发起全局事务的commit/rollback，即

有异常（说明存在事务分支rollback或者异常）则逆向操作所有undo_log文件，即回滚

无异常则将undo_log文件删除释放磁盘空间（此时事务分支都是已经commit的）

Seata-TCC（Try Confirm Cancel）模式

Seata 框架把每组 TCC 接口当做一个 Resource，称为 TCC Resource。

TC：Transaction Coordinator 事务协调器，管理全局的分支事务的状态，用于全局事务的提交和回滚。
TM：Transaction Manager 事务管理器，事务发起者用于开启全局事务。
RM：Resource Manager 资源管理器，分支事务上的资源管理，用于所有事务参与者（包括事务发起者）向TC注册分支事务，上报分支事务的状态，接受TC的命令来提交或者回滚分支事务。

Seata-TCC流程：

大致可以分为两个阶段：

一阶段：事务参与者异步完成Try
二阶段：事务参与者同步完成Confirm/Cancel

TCC流程与AT相似，区别在于TCC根据方法而不是文件决定全局事务的commit/rollback

①事务发起者（服务，也是事务参与者）的TM向TC发起全局事务，TC返回一个全局唯一标识XID（一个个全局事务组）

②事务发起者的RM向TC注册事务分支branch并被TC纳入XID同时事务发起者将XID传播到其他事务参与者（服务）

③事务参与者的RM向TC注册事务分支branch并被纳入XID

④TC驱动所有事务分支执行Try方法

⑤事务分支各自commit/rollback到数据库

⑥RM根据事务分支commit/rollback情况通知TC

⑦待所有分支事务执行完毕，TC根据所有事务分支情况决定执行Confirm/Cancel

注意：

真正执行服务commit/rollback操作的都是TM，说白了就是commit/rollback都要先通知TM，TM在提交分支事务的commit/rollback，TC是全局事务的Confirm和Cancel的，不要混淆
Seata的TCC就是通过@GlobalTransactionScanner提交事务，@LocalTCC表示事务参与者，通过TC协调调用，将资源预留、提交、回滚交由业务方编码控制写在Try()、Confirm()、Cancel()中

TCC功能固然强大，但是还需要注意三种异常处理，分别是空回滚、幂等、悬挂

空回滚、幂等、悬挂。

空回滚

产生原因：空回滚就是在没有调用Try() 方法的情况下，调用了二阶段的 Cancel() 方法。如前面说到，如果网络异常也会导致Try()失败，紧接着到第二阶段Cancel()先执行了。注意：网络波动造成的异常是不可避免的，我们只能通过技术手段去解决。

后果：Cancel会将原本失败才回滚的资源提前回滚，形成脏数据。

解决方案：要应对空回滚的问题，就需要让参与者在二阶段的Cancel方法中有办法识别到一阶段的Try是否已经执行。解决方案就是，使用分支事务记录表（事务状态控制表），让Try()方法执行后留下信息记录，Cancel通过查询信息记录的有无情况，判断Try()方法是否执行过。也即，Try()方法后，就在分支事务记录表中插入一条数据，数据包含全局事务XID、分支事务branchId、状态码等（该表还会用于其他异常情况）

悬挂

产生原因：悬挂产生的原因是基于空回滚的前提下，因为网络波动，待Cancel()执行完之后，Try()又可以执行了（注意，网络波动异常虽然会让Cancel()先执行，但Try()方法不会因此停止，TM会在一定时间内尝试调用Try()），前面解决了空回滚问题，这时候会因为Cancel()无法执行，导致Try()执行完之后，Cancel()也不再执行的情况

后果：资源预留扣除，因为不再有二阶段操作，所以资源预留成了脏数据

解决方案：还是使用分支事务记录表，当Try()执行成功同时像分支事务记录表中插入数据。当Cancel()执行的时候就要做判断：用XID和branchId查找是否有对应的数据存在，如果存在说明Try()执行成功，Cancel()执行完也应该留下对应的状态；如果该数据不存在，说明Try()从未执行，注意，这个时候Cancel()就要留下数据，状态对应用自己的，目的就是为了Try()也要做防悬挂，Try()中需要新增判断：用XID和branchId查找是否有对应的数据存在，如果存在说明Cancel()执行了；如果不存在就正常插入新数据（空回滚所做的事）。总而言之：Try()和Cancel()方法都要根据XID和branchId查询分支事务记录表中是否存在对方留下的数据，没有数据代表对方没有执行，自己就应该执行并插入数据。

==？有没有空提交这种异常？

==》理论上或者说几乎不会出现这种情况，因为只有Try()成功才会执行Confirm()方法。如果需要做应对策略的话，可以在Confirm()方法中做对应的判断，判断根据就是XID和branchId中找到的数据中的状态码（对应Try()和Cancel()方法中插入数据时就要留下状态区分是哪个方法留下的数据，也即谁先执行的），根据状态码判断：如果状态码是Try()的，正常执行Confirm()，否则返回空或者不执行操作

幂等

产生原因：前面说到，因为网络抖动等原因，分布式事务框架可能会重复调用同一个分布式事务中的一个分支事务的二阶段接口

结果：重复调用方法，多次执行相同操作

解决方案：幂等性检查，不论执行多少次，结果保证相同。还是分支事务记录表，前面到的预防几乎不可能出现的空提交的方法其实就可以解决做幂等性处理。根据状态码判断：如果状态码是Try()的，正常执行Confirm()，否则返回空或者不执行操作（二阶段执行完之后就会通知TM帮忙commit/rollback，接着就等待TC的全局事务结果通知啦）

Seata的AT模式和TCC模式的区别比较

At模式

优点：

使用简单
无侵入性（即不影响原业务代码）

缺点：

并发能力较TCC弱（两个阶段加锁，没有预留资源操作）
基于connection进行代理，所有微服务的业务都是使用关系型数据库的情况下才能使用Seata-AT （es、mdb、redis都是没有connection这个概念的）

TCC模式

优点：

并发能力较AT强（一阶段加锁，Try结束释放锁）
通过业务层面进行补偿，即使是非关系型数据库也能包装业务数据一致性（因为TCC的RM管理的资源对象不再只是关系型数据库）

缺点：

侵入性太大
实现难度大
开发成本高

注意事项：

高并发体现在：账户 A 上有 100 元，事务 T1 要扣除其中的 30 元，事务 T2 也要扣除 30 元，出现并发。两阶段锁这样的并发访问控制技术要求一直持有数据库资源锁直到整个事务执行结束，特别是在分布式事务架构下，要求持有锁到分布式事务第二阶段执行结束，也就是说，分布式事务会加长资源锁的持有时间，导致并发性能进一步下降。TCC 模型的隔离性思想就是通过业务的改造，在Try结束之后，从底层数据库资源层面的加锁过渡为上层业务层面的加锁，从而释放底层数据库锁资源，放宽分布式事务锁协议，将锁的粒度降到最低，以最大限度提高业务并发性能。虽然数据库层面资源锁被释放了，但这笔资金被业务隔离，不允许除本事务之外的其它并发事务动用。
从高并发中我们还会发现一个小问题（可以不处理，具体看业务），如果事务执行较长，用户转账后发现自己的钱被扣除了，但是对方的钱还没到账。为了用户体验，此时我们就要多做一个“资金冻结”业务，让用户知道”转账总会有结果，但是这部分的钱你不能用，成功了我就给你转过去，冻结自然也消失，如果失败，我就把你给钱从冻结中返回到你可用账户“
TCC虽然优秀，但是侵入性大（需要在原有业务代码上修改，但是相信不久后它的开发团队会解决这一问题），而且会影响业务效率，因此在实际开发中不一定非要使用它。它能解决0.1％概率的事务问题，但却会影响0.1％问题所涉及的全部业务效率。而对于个别案例，我们更倾向于人工处理解决。

SpringCloud集成Seata

无论使用哪种模式都要导入依赖

<!--springcloud集成seata-->
<dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-starter-alibaba-seata</artifactId><version>2.2.2.RELEASE</version><exclusions><exclusion><groupId>io.seata</groupId><artifactId>seata-spring-boot-starter</artifactId></exclusion></exclusions>
</dependency>
<dependency><groupId>io.seata</groupId><artifactId>seata-spring-boot-starter</artifactId><version>1.3.0</version>
</dependency>

SpringCloud集成Seata-AT模式

事务发起者

/*** 积分退款，事务发起者本身@GlobalTransactional就是事务，因此其具备自己执行TCC方法* @param orderInfo*/
@Override
@GlobalTransactional
public void refundIntergral(OrderInfo orderInfo) {int count = orderInfoMapper.changeRefundStatus(orderInfo.getOrderNo(), OrderInfo.STATUS_REFUND);if (count == 0){  //退款失败throw new BusinessException(SeckillCodeMsg.REFUND_ERROR);}//远程调用积分服务，返回积分给用户OperateIntergralVo vo = new OperateIntergralVo();vo.setUserId(orderInfo.getUserId());vo.setValue(orderInfo.getIntergral());intergralFeignApi.incrIntergral(vo);  //RPG业务
}

事务参与者

@Override
@GlobalTransactional
public void incrIntergral(Long userId, Long amount) {usableIntegralMapper.incrIntergral(userId, amount);
}

SpringCloud集成Seata-TCC模式

事务发起者

/*** 积分退款，事务发起者本身@GlobalTransactional就是事务，因此其具备自己执行TCC方法* @param orderInfo*/
@Override
@GlobalTransactional
public void refundIntergral(OrderInfo orderInfo) {int count = orderInfoMapper.changeRefundStatus(orderInfo.getOrderNo(), OrderInfo.STATUS_REFUND);if (count == 0){  //退款失败throw new BusinessException(SeckillCodeMsg.REFUND_ERROR);}//远程调用积分服务，返回积分给用户OperateIntergralVo vo = new OperateIntergralVo();vo.setUserId(orderInfo.getUserId());vo.setValue(orderInfo.getIntergral());intergralFeignApi.incrIntergral(vo);  //RPG业务
}

事务参与者

/*** 积分退款业务* @param vo* @return*/
@RequestMapping("/refund")
Result incrIntergral(@RequestBody OperateIntergralVo vo){usableIntegralService.incrIntergralTry(vo,null);return Result.success();
}

@LocalTCC
public interface IUsableIntegralService {/*** 增加积分*/@TwoPhaseBusinessAction(name = "incrIntergralTry", commitMethod = "incrIntergralConfirm", rollbackMethod = "incrIntergralCancel")void incrIntergralTry(@BusinessActionContextParameter(paramName = "operateIntergralVo") OperateIntergralVo operateIntergralVo, BusinessActionContext context);void incrIntergralConfirm(BusinessActionContext context);void incrIntergralCancel(BusinessActionContext context);
}

/*** 增加积分Try，空操作，新增表数据* @param vo* @param context*/
@Transactional
@Override
public void incrIntergralTry(OperateIntergralVo vo, BusinessActionContext context) {AccountTransaction accountTransaction = new AccountTransaction();AccountTransaction getAccountTransaction = accountTransactionMapper.get(context.getXid(), context.getBranchId());if (getAccountTransaction != null){  //如果在此之前有该xid和branchId的数据则不执行trythrow new BusinessException(IntergralCodeMsg.OP_INTERGRAL_ERROR);}accountTransaction.setState(AccountTransaction.STATE_TRY);accountTransaction.setTxId(context.getXid());  //全局事务xidaccountTransaction.setActionId(context.getBranchId());  //分支事务idaccountTransaction.setUserId(vo.getUserId());  //用户idDate date = new Date();accountTransaction.setGmtCreated(date);  //事务日志记录创建时间accountTransaction.setGmtModified(date);  //事务日志记录修改时间accountTransaction.setAmount(vo.getValue());  //金额accountTransactionMapper.insert(accountTransaction);  //持久化到数据库//空操作
}/*** 增加积分Confirm，增加积分，更改表状态* @param context*/
@Transactional
@Override
public void incrIntergralConfirm(BusinessActionContext context) {AccountTransaction getAccountTransaction = accountTransactionMapper.get(context.getXid(), context.getBranchId());if ((getAccountTransaction == null) && (getAccountTransaction.getState()==AccountTransaction.STATE_TRY)){  //如果在此之前没有该xid和branchId的数据，并且状态不为try，说明try未成功throw new BusinessException(IntergralCodeMsg.OP_INTERGRAL_ERROR);}//增加积分AccountTransaction accountTransaction = accountTransactionMapper.get(context.getXid(), context.getBranchId());usableIntegralMapper.incrIntergral(accountTransaction.getUserId(),accountTransaction.getAmount());//更新表状态accountTransactionMapper.updateAccountTransactionState(context.getXid(),context.getBranchId(),AccountTransaction.STATE_COMMIT,AccountTransaction.STATE_TRY);
}/*** 增加积分Cancel，更改表状态，空操作* @param context*/
@Transactional
@Override
public void incrIntergralCancel(BusinessActionContext context) {AccountTransaction getAccountTransaction = accountTransactionMapper.get(context.getXid(), context.getBranchId());if ((getAccountTransaction == null) && (getAccountTransaction.getState()==AccountTransaction.STATE_TRY)){  //如果在此之前没有该xid和branchId的数据，并且状态不为try，说明try未成功throw new BusinessException(IntergralCodeMsg.OP_INTERGRAL_ERROR);}//更新表状态accountTransactionMapper.updateAccountTransactionState(context.getXid(),context.getBranchId(),AccountTransaction.STATE_CANCEL,AccountTransaction.STATE_COMMIT);//空操作
}

原理：

①贴有@GlobalTransactional的业务方法即为事务发起者，seata的GlobalTransactionScanner扫描器扫描随即向TC开启全局分布式事务

②在事务参与者的接口上贴@LocalTCC注解，Try()方法上贴@TwoPhaseBusinessAction注解标记这是个TCC接口，同时指定commitMethod，rollbackMethod的名称，名称对应的是同Try()方法中的接口的另外两个提交和回滚方法

③调用各个微服务的Try()方法，执行资源检查及预留操作（每调用一次 Try ()方法，切面会先向 TC 注册一个分支事务branch，然后才去执行原来的 RPC （远程过程调用）调用）

④当所有Try()方法均执行成功时，对全局事物进行提交，即由TC调用每个微服务的Confirm()方法；当任意一个方法Try()失败(预留资源不足，异或网络异常，代码异常等任何异常)，由TC调用每个微服务的Cancel()方法对全局事务进行回滚。

事务的起点就是事务的发起者，其下所有方法业务如果扫描到@GlobalTransactional注解就是事务参与者。一个事务发起者所涵盖的所有事务参与者就是一个事务组。