分布式事务与Seata落地

一、事务基础

1.1 本地事务

事务指的就是一个操作单元, 在这个操作单元中的所有操作最终要保持一致的行为, 要么所有操作都成功, 要么所有的操作都被撤销。

1.2 本地事务特性

本地事务四大特性: ACID

• A: 原子性(Atomicity), 一个事务中的所有操作, 要么全部完成, 要么全部不完成

• C: 一致性(Consistency), 在一个事务执行之前和执行之后数据库都必须处于一致性状态

• I: 隔离性(Isolation), 在并发环境中, 当不同的事务同时操作相同的数据时, 事务之间互不影响

• D: 持久性(Durability), 指的是只要事务成功结束, 它对数据库所做的更新就必须永久的保存下来

数据库事务在实现时会将一次事务涉及的所有操作全部纳入到一个不可分割的执行单元, 该执行单元中的所有操作要么都成功, 要么都失败, 只要其中任一操作执行失败, 都将导致整个事务的回滚。

1.3 事务实现原理

以MySQL 的InNODB (InNODB 是 MySQL 的一个存储引擎)为例, 介绍一下单一数据库的事务实现原理。

InNODB 是通过 日志和锁 来保证的事务的 ACID特性, 具体如下:

1. 通过数据库锁的机制, 保障事务的隔离性；

2. 通过 Redo Log(重做日志)来, 保障事务的持久性；

3. 通过 Undo Log (撤销日志)来, 保障事务的原子性；

4. 通过 Undo Log (撤销日志)来, 保障事务的一致性；

Undo Log 如何保障事务的原子性呢？

具体的方式为: 在操作任何数据之前, 首先将数据备份到一个地方(这个存储数据备份的地方称为 Undo Log), 然后进行数据的修改。如果出现了错误或者用户执行了 Rollback 语句, 系统可以利用 Undo Log 中的备份将数据恢复到事务开始之前的状态。

Redo Log如何保障事务的持久性呢？

具体的方式为: Redo Log 记录的是新数据的备份(和 Undo Log 相反)。在事务提交前, 只要将 Redo Log 持久化即可, 不需要将数据持久化(异步)。当系统崩溃时, 虽然数据没有持久化, 但是 Redo Log 已经持久化。系统可以根据 Redo Log 的内容, 将所有数据恢复到崩溃之前的状态

1.4 分布式事务

分布式事务: 在分布式系统中一次操作需要由多个(微)服务协同完成, 这种由不同的服务之间通过网络协同完成的事务称为分布式事务

通俗讲, 就是一次大的操作由不同的小操作组成, 这些小的操作分布在不同的服务器上, 且属于不同的应用, 分布式事务需要保证这些小操作要么全部成功, 要么全部失败。

大白话: 分布式环境下保证各个事务的ACID

1.5 分布式事务场景

1.5.1 跨库事务

跨库事务指的是, 一个相对复杂的逻辑功能需要操作多个库数据, 不同的库中存储不同的业务数据。

此时操作需要考虑DB1事务, 也需要考虑DB2的事务, 妥妥的分布式事务

1.5.2 分库分表

当一个库/或者一张表数据量比较大或者预期未来的数据量会有比较大增长时, 再设计表是考虑水平拆分, 即分库分表。

如果分库了, 那么就是跨库操作, 那么就要使用分布式事务

DB2分库分表之后, 执行DML操作时, 需要考虑分区1, 分区2操作一起成功, 一起失败。此时service面临着分布式事务的问题。

1.5.3 微服务化

一个逻辑多个微服务参与, 涉及多个数据库数据改动, 又是一个分布式事务场景。

• 如果说都是在同一个库中操作的, 那么就没有所谓的分布式事务, 都是本地事务

1.6 分布式事务特性

1.6.1 CAP概念

CAP定理是分布式系统中的重要理论, 在一个分布式系统中最多只能同时满足一致性(Consistency)、可用性(Availability)和分区容错性(Partition tolerance)这三项中的两项, 不能同时满足这三项。

C: Consistency(一致性)

一致性指"all NOdes see the same data at the same time", 即更新操作成功后, 所有节点在同一时间的数据完全一致。

A: Availability(可用性)

可用性指"reads and writes always succeed", 即服务一直可用且能够正常响应(不保证返回的是最新写入的数据)。

P: Partition tolerance(分区容错性)

分区容错性指"the system continues to operate despite arbitrary message loss or failure of part of the system", 即分布式系统在遇到某节点或网络分区故障的时候, 仍然能够对外提供满足一致性和可用性的服务。

1.6.2 CAP选择

无数实践证明, CAP只存在理想的状态, 真实的的分布式系统必须在CAP 3者中选其2:

CA

放弃P, 放弃分区容错性的话, 则放弃了分布式, 放弃了系统的可扩展性

CP

放弃A, 放弃可用性的话, 则在遇到网络分区或其他故障时, 受影响的服务需要等待一定的时间, 再此期间无法对外提供政策的服务, 即不可用

AP

放弃C, 放弃一致性的话, 则系统无法保证数据保持实时的一致性, 在数据达到最终一致性时, 有个时间窗口, 在时间窗口内, 数据是不一致的。

注意: 对于分布式系统来说, P是不能放弃的, 因此架构师通常是在可用性和一致性之间权衡。

16.3 CAP无法共存证明

**理想操作: **

用户第一次访问, 通过Nginx后, 走网段1, 执行数据添加, 数据同步到网段2的数据库中

用户第二次访问, 通过Nginx后, 走网段2, 执行数据查询, 查询到第一次添加的啊data数据

非理想状态

用户第一次访问, 通过Nginx后, 走网段1, 执行数据添加, 数据无法同步到网段2的数据库

用户第二次访问, 通过Nginx后, 走网段2, 执行数据查询

分布式系统存在意义, 最低要求必须保证分区容错性成立, 那么就意味着, 上面操作存在2种选择。

选择可用性(A), 牺牲一致性©: 返回空数据,

选择一致性©, 牺牲可用性(A): 阻塞等待, 直到网段1, 网段2连通, 数据同步完成之后, 再返回data数据。

1.6.4 BASE

BASE是Basically Available(基本可用)、**Soft state(软状态)和Eventually consistent(最终一致性)**三个短语的简写。

BASE是对CAP中一致性和可用性权衡的结果, 其来源于对大规模互联网系统分布式实践的总结, 是基于CAP定理逐步演化而来的, 其核心思想是即使无法做到强一致性, 但每个应用都可以根据自身的业务特点, 采用适当的方法来使系统达到最终一致性。

Basically Available(基本可用)

基本可用指分布式系统在出现故障时, 系统允许损失部分可用性, 即保证核心功能或者当前最重要功能可用。对于用户来说, 他们当前最关注的功能或者最常用的功能的可用性将会获得保证, 但是其他功能会被削弱

Soft state(软状态)

软状态指的是: 允许系统中的数据存在中间状态, 并认为该状态不影响系统的整体可用性, 即允许系统在多个不同节点的数据副本存在数据延时。

Eventually consistent(最终一致性)

上面说软状态, 然后不可能一直是软状态, 必须有个时间期限。在期限过后, 应当保证所有副本保持数据一致性, 从而达到数据的最终一致性。这个时间期限取决于网络延时、系统负载、数据复制方案设计等等因素。

总结

总的来说, BASE 理论面向的是大型高可用可扩展的分布式系统, 和传统事务的 ACID 是相反的, 它完全不同于 ACID 的强一致性模型, 而是通过牺牲强一致性来获得可用性, 并允许数据在一段时间是不一致的。

二、分布式事务问题

2.1 需求

需求: 用户下单, 扣款, 扣库存。

2.2 项目演示

根据上面分析, 项目设计出3个微服务

业务服务: business-service

订单服务: order-service

账户服务: account-service

库存服务: stock-service

代码如下

2.2.1 数据库准备

创建3个数据库与3张表(跨库了, 所以存在分布式事务)

seata-account

CREATE TABLE `t_account`
(`id`      int(11) NOT NULL AUTO_INCREMENT,`user_id` varchar(255) DEFAULT NULL,`money`   int(11) DEFAULT 0,PRIMARY KEY (`id`)
) ENGINE=InNODB DEFAULT CHARSET=utf8;INSERT INTO `t_account` VALUES (1, 'U100000', 900);

seata-order

CREATE TABLE `t_order`
(`id`             int(11) NOT NULL AUTO_INCREMENT,`user_id`        varchar(255) DEFAULT NULL,`commodity_code` varchar(255) DEFAULT NULL,`count`          int(11) DEFAULT 0,`money`          int(11) DEFAULT 0,PRIMARY KEY (`id`)
) ENGINE=InNODB DEFAULT CHARSET=utf8;

seata-stock

CREATE TABLE `t_stock`
(`id`             int(11) NOT NULL AUTO_INCREMENT,`commodity_code` varchar(255) DEFAULT NULL,`count`          int(11) DEFAULT 0,PRIMARY KEY (`id`),UNIQUE KEY (`commodity_code`)
) ENGINE=InNODB DEFAULT CHARSET=utf8;INSERT INTO `t_stock` VALUES (1, 'C100000', 100);

2.2.2 库存服务

stock-service

依赖

<dependencies><!-- bootstrap 启动器 --><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-bootstrap</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.56</version></dependency><dependency><groupId>com.baomidou</groupId><artifactId>mybatis-plus-boot-starter</artifactId><version>3.4.2</version></dependency><!--nacos客户端--><dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId></dependency><!--fegin组件--><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-openfeign</artifactId></dependency><!--sentinel组件--><dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-starter-alibaba-sentinel</artifactId></dependency><dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></dependency><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-loadbalancer</artifactId></dependency>
</dependencies>

配置文件

# Tomcat
server:port: 8083
# Spring
spring:application:# 应用名称name: stock-serviceprofiles:# 环境配置active: devcloud:nacos:discovery:# 服务注册地址server-addr: 127.0.0.1:8848datasource:driver-class-name: com.mysql.cj.jdbc.Driverurl: jdbc:mysql:///seata-stock?serverTimezone=UTC&useUnicode=true&characterEncoding=utf-8&useSSL=trueusername: rootpassword: admin
mybatis-plus:configuration:log-impl: org.apache.ibatis.logging.stdout.StdOutImpl

domain

import com.baomidou.mybatisplus.anNOtation.IdType;
import com.baomidou.mybatisplus.anNOtation.TableId;
import com.baomidou.mybatisplus.anNOtation.TableName;
import lombok.Data;@Data
@TableName("t_stock")
public class Stock {@TableId(type = IdType.AUTO)private Integer id;private String commodityCode;private Integer count;
}

mapper

import cn.wolfcode.tx.stock.domain.Stock;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;public interface StockMapper  extends BaseMapper<Stock> {
}

service

import com.baomidou.mybatisplus.extension.service.IService;public interface IStockService  extends IService<Stock> {/*** 扣库存* @param commodityCode* @param count*/void deduct(String commodityCode, int count);
}

service.impl

import cn.wolfcode.tx.stock.domain.Stock;
import cn.wolfcode.tx.stock.mapper.StockMapper;
import cn.wolfcode.tx.stock.service.IStockService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import io.seata.core.context.RootContext;
import org.springframework.stereotype.Service;
import org.springframework.transaction.anNOtation.Transactional;@Service
public class StockServiceImpl extends ServiceImpl<StockMapper, Stock> implements IStockService {@Override@Transactionalpublic void deduct(String commodityCode, int count) {Stock one = lambdaQuery().eq(Stock::getCommodityCode, commodityCode).one();if(one != null && one.getCount() < count){throw new RuntimeException("NOt ENOugh Count ...");}lambdaUpdate().setSql("count = count-" + count).eq(Stock::getCommodityCode, commodityCode).update();}
}

controller

import cn.wolfcode.tx.stock.service.IStockService;
import org.springframework.beans.factory.anNOtation.Autowired;
import org.springframework.web.bind.anNOtation.*;@RestController
@RequestMapping("stocks")
public class StockController {@Autowiredprivate IStockService StockService;@Autowiredprivate IStockService stockService;@GetMapping(value = "/deduct")public String deduct(String commodityCode, int count) {try {stockService.deduct(commodityCode, count);} catch (Exception exx) {exx.printStackTrace();return "FAIL";}return "SUCCESS";}
}

启动类

import org.mybatis.spring.anNOtation.MapperScan;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
import org.springframework.cloud.openfeign.EnableFeignClients;@SpringBootApplication
@EnableDiscoveryClient
@EnableFeignClients
@MapperScan("cn.wolfcode.tx.stock.mapper")
public class StockApplication {public static void main(String[] args) {SpringApplication.run(StockApplication.class, args);}
}

2.2.3 账户服务

account-service

依赖

<dependencies><!-- bootstrap 启动器 --><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-bootstrap</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.56</version></dependency><dependency><groupId>com.baomidou</groupId><artifactId>mybatis-plus-boot-starter</artifactId><version>3.4.2</version></dependency><!--nacos客户端--><dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId></dependency><!--fegin组件--><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-openfeign</artifactId></dependency><!--sentinel组件--><dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-starter-alibaba-sentinel</artifactId></dependency><dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></dependency>
</dependencies>

配置文件

# Tomcat
server:port: 8081
# Spring
spring:application:# 应用名称name: account-serviceprofiles:# 环境配置active: devcloud:nacos:discovery:# 服务注册地址server-addr: 127.0.0.1:8848datasource:driver-class-name: com.mysql.cj.jdbc.Driverurl: jdbc:mysql:///seata-account?serverTimezone=UTC&useUnicode=true&characterEncoding=utf-8&useSSL=trueusername: rootpassword: admin
mybatis-plus:configuration:log-impl: org.apache.ibatis.logging.stdout.StdOutImpl

domain

import com.baomidou.mybatisplus.anNOtation.IdType;
import com.baomidou.mybatisplus.anNOtation.TableId;
import com.baomidou.mybatisplus.anNOtation.TableName;
import lombok.Data;@Data
@TableName("t_account")
public class Account {@TableId(type = IdType.AUTO)private Integer id;private String userId;private int money;
}

mapper

import com.baomidou.mybatisplus.core.mapper.BaseMapper;public interface AccountMapper extends BaseMapper<Account> {
}

service

import com.baomidou.mybatisplus.extension.service.IService;public interface IAccountService  extends IService<Account> {/*** 账户扣款* @param userId* @param money* @return*/void reduce(String userId, int money);
}

service.impl

package cn.wolfcode.tx.account.service.impl;import cn.wolfcode.tx.account.domain.Account;
import cn.wolfcode.tx.account.mapper.AccountMapper;
import cn.wolfcode.tx.account.service.IAccountService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import io.seata.core.context.RootContext;
import org.springframework.stereotype.Service;
import org.springframework.transaction.anNOtation.Transactional;@Service
public class AccountServiceImpl extends ServiceImpl<AccountMapper, Account> implements IAccountService {@Override@Transactionalpublic void reduce(String userId, int money) {Account one = lambdaQuery().eq(Account::getUserId, userId).one();if(one != null && one.getMoney() < money){throw new RuntimeException("NOt ENOugh Money ...");}lambdaUpdate().setSql("money = money - " + money).eq(Account::getUserId, userId).update();}
}

controller

import org.springframework.beans.factory.anNOtation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.anNOtation.*;@RestController
@RequestMapping("accounts")
public class AccountController {@Autowiredprivate IAccountService accountService;@GetMapping(value = "/reduce")public String reduce(String userId, int money) {try {accountService.reduce(userId, money);} catch (Exception exx) {exx.printStackTrace();return "FAIL";}return "SUCCESS";}}

启动类

import org.mybatis.spring.anNOtation.MapperScan;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
import org.springframework.cloud.openfeign.EnableFeignClients;@SpringBootApplication
@MapperScan("com.whitebrocade.tx.account.mapper") // 包扫描换成你的mapper目录
@EnableDiscoveryClient
@EnableFeignClients
public class AccountApplication {public static void main(String[] args) {SpringApplication.run(AccountApplication.class, args);}
}

2.2.4 订单服务

order-service

依赖

<dependencies><!-- bootstrap 启动器 --><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-bootstrap</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.56</version></dependency><dependency><groupId>com.baomidou</groupId><artifactId>mybatis-plus-boot-starter</artifactId><version>3.4.2</version></dependency><!--nacos客户端--><dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId></dependency><!--fegin组件--><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-openfeign</artifactId></dependency><!--sentinel组件--><dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-starter-alibaba-sentinel</artifactId></dependency><dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></dependency><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-loadbalancer</artifactId></dependency>
</dependencies>

配置文件

# Tomcat
server:port: 8082
# Spring
spring:application:# 应用名称name: order-serviceprofiles:# 环境配置active: devcloud:nacos:discovery:# 服务注册地址server-addr: 127.0.0.1:8848datasource:driver-class-name: com.mysql.cj.jdbc.Driverurl: jdbc:mysql:///seata-order?serverTimezone=UTC&useUnicode=true&characterEncoding=utf-8&useSSL=trueusername: rootpassword: adminmybatis-plus:configuration:log-impl: org.apache.ibatis.logging.stdout.StdOutImpl

domain

import com.baomidou.mybatisplus.anNOtation.IdType;
import com.baomidou.mybatisplus.anNOtation.TableId;
import com.baomidou.mybatisplus.anNOtation.TableName;
import lombok.Data;@Data
@TableName("t_order")
public class Order {// id自增长@TableId(type = IdType.AUTO)private Integer id;private String userId;private String commodityCode;private Integer count;private Integer money;
}

mapper

import com.baomidou.mybatisplus.core.mapper.BaseMapper;public interface OrderMapper  extends BaseMapper<Order> {
}

service

import com.baomidou.mybatisplus.extension.service.IService;public interface IOrderService  extends IService<Order> {/*** 创建订单*/void create(String userId, String commodityCode, int orderCount);
}

service.impl

import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import io.seata.core.context.RootContext;
import org.springframework.beans.factory.anNOtation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.anNOtation.Transactional;@Service
@Transactional
public class OrderServiceImpl extends ServiceImpl<OrderMapper, Order> implements IOrderService {@Autowiredprivate AccountFeignClient accountFeignClient;@Override@Transactionalpublic void create(String userId, String commodityCode, int count) {// 定单总价 = 订购数量(count) * 商品单价(100)int orderMoney = count * 100;// 生成订单Order order = new Order();order.setCount(count);order.setCommodityCode(commodityCode);order.setUserId(userId);order.setMoney(orderMoney);super.save(order);// 调用账户余额扣减String result = accountFeignClient.reduce(userId, orderMoney);if (!"SUCCESS".equals(result)) {throw new RuntimeException("Failed to call Account Service. ");}}
}

controller

import org.springframework.beans.factory.anNOtation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.anNOtation.*;@RestController
@RequestMapping("orders")
public class OrderController {@Autowiredprivate IOrderService orderService;@GetMapping(value = "/create")public String create(String userId, String commodityCode, int orderCount) {try {orderService.create(userId, commodityCode, orderCount);} catch (Exception exx) {exx.printStackTrace();return "FAIL";}return "SUCCESS";}
}

Feign接口

import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.anNOtation.GetMapping;
import org.springframework.web.bind.anNOtation.RequestParam;@FeignClient(name = "account-service")
public interface AccountFeignClient {@GetMapping("/accounts/reduce")String reduce(@RequestParam("userId") String userId, @RequestParam("money") int money);
}

启动类

import org.mybatis.spring.anNOtation.MapperScan;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
import org.springframework.cloud.openfeign.EnableFeignClients;@SpringBootApplication
@MapperScan("com.whitebrocade.tx.order.mapper")
@EnableDiscoveryClient
@EnableFeignClients
public class OrderApplication {public static void main(String[] args) {SpringApplication.run(OrderApplication.class, args);}
}

2.2.6 业务服务

business-service

依赖

<dependencies><!-- bootstrap 启动器 --><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-bootstrap</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.56</version></dependency><!--nacos客户端--><dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId></dependency><!--fegin组件--><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-openfeign</artifactId></dependency><!--sentinel组件--><dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-starter-alibaba-sentinel</artifactId></dependency><dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></dependency><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-loadbalancer</artifactId></dependency>
</dependencies>

配置文件

# Tomcat
server:port: 8088
# Spring
spring:application:# 应用名称name: business-serviceprofiles:# 环境配置active: devcloud:nacos:discovery:# 服务注册地址server-addr: 127.0.0.1:8848

测试数据

public class TestDatas {public static final String USER_ID = "U100000";public static final String COMMODITY_CODE = "C100000";
}

service

public interface IBusinessService{void purchase(String userId, String commodityCode, int orderCount, boolean rollback);
}

service.impl

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.anNOtation.Autowired;
import org.springframework.stereotype.Service;@Slfj
@Service
public class BusinessServiceImpl implements IBusinessService {@Autowiredprivate StockFeignClient stockFeignClient;@Autowiredprivate OrderFeignClient orderFeignClient;@Overridepublic void purchase(String userId, String commodityCode, int orderCount, boolean rollback) {String result = stockFeignClient.deduct(commodityCode, orderCount);if (!"SUCCESS".equals(result)) {throw new RuntimeException("库存服务调用失败,事务回滚!");}result = orderFeignClient.create(userId, commodityCode, orderCount);if (!"SUCCESS".equals(result)) {throw new RuntimeException("订单服务调用失败,事务回滚!");}if (rollback) {throw new RuntimeException("Force rollback ... ");}}
}

feign接口

import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.anNOtation.GetMapping;
import org.springframework.web.bind.anNOtation.RequestParam;@FeignClient(name = "order-service")
public interface OrderFeignClient {@GetMapping("/orders/create")String create(@RequestParam("userId") String userId, @RequestParam("commodityCode") String commodityCode, @RequestParam("orderCount") int orderCount);}

import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.anNOtation.GetMapping;
import org.springframework.web.bind.anNOtation.RequestParam;@FeignClient(name = "stock-service")
public interface StockFeignClient {@GetMapping("/stocks/deduct")String deduct(@RequestParam("commodityCode") String commodityCode, @RequestParam("count") int count);}

controller

import org.springframework.beans.factory.anNOtation.Autowired;
import org.springframework.web.bind.anNOtation.GetMapping;
import org.springframework.web.bind.anNOtation.RequestMapping;
import org.springframework.web.bind.anNOtation.RestController;@RestController
@RequestMapping("businesses")
public class BusinessController {@Autowiredprivate IBusinessService businessService;@GetMapping(value = "/purchase")public String purchase(Boolean rollback, Integer count) {int orderCount = 10;if (count != null) {orderCount = count;}try {businessService.purchase(TestDatas.USER_ID, TestDatas.COMMODITY_CODE, orderCount, rollback == null ? false : rollback.booleanValue());} catch (Exception exx) {return "Purchase Failed:" + exx.getMessage();}return "SUCCESS";}
}

启动类

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
import org.springframework.cloud.openfeign.EnableFeignClients;@SpringBootApplication
@EnableDiscoveryClient
@EnableFeignClients
public class BusinessApplication {public static void main(String[] args) {SpringApplication.run(BusinessApplication.class, args);}
}

2.2.5 测试

启动nacos, 启动4个服务,

访问: http://localhost:8088/businesses/purchase?rollback=false&count=10

2.3 提出问题

关闭order-service 再发起请求会怎样？

此时该怎么办? 此时就需要用到分布式事务啦~

三、分布式事务常用方案

方案汇总

先上一幅图: 尼恩大神画的

原地址: https://blog.csdn.net/crazymakercircle/article/details/109459593

很明显可以看出分布式事务后续演变成2条路径

CP(一致性 + 分区)

放弃可用性A, 保证数据强一致性.

经典方案:

• 2PC(二阶段提交)
• 3PC(三阶段提交)

AP(可用性 + 分区)

暂时放弃一致性(注意咯, 这里说的是暂时),保证可用,后续通过某种手段(比如: MQ/程序补偿)打到最终一致性性.

经典方案

1. 本地消息表
2. MQ消息事务
3. TCC
4. SAGA

CP解决方案

方案1: 2PC

XA协议

要讲清楚2PC模式, 需要科普一个协议: XA协议

XA协议是X/Open的组织定义的分布式事务处理标准规范(DTP)。它定义了全局的事务管理器(Transaction Manager, 用于协调全局事务)和局部的资源管理器(Resource Manager, 用于驱动本地事务)之间的通讯接口。在TM与多个RM之间形成一个双向通信桥梁, 从而在多个数据库资源下保证ACID四个特性。目前几乎所有的主流数据库都对XA规范提供了支持。**

XA协议包含有几个角色:

• AP(Application Program) : 既应用程序, 可以理解为使用DTP分布式事务的程序。
• RM(Resource Manager) : 即资源管理器, 可以理解为事务的参与者, 一般情况下是指一个数据库实例, 通过资源管理器对该数据库进行控制, 资源管理器控制着分支事务。
• TM(Transaction Manager) : 事务管理器, 负责协调和管理事务, 事务管理器控制着全局事务, 管理事务生命周期, 并协调各个RM。全局事务是指分布式事务处理环境中, 需要操作多个数据库共同完成一个工作, 这个TM即是一个全局事务

2PC(Two-phase commit protocol), 中文叫二阶段提交。 二阶段提交是一种强一致性设计, 2PC 引入一个事务协调者(TM)的角色来协调管理各参与者(参与者也可以认为是各本地资源RM)的提交和回滚, 二阶段分别指的是准备和提交两个阶段

第一阶段: 准备阶段

准备阶段, 事务协调者™会给各事务参与者(RM)发送准备命令(prepare), 参与者准备成功后返回(ready)

• 协调者向所有参与者发送事务操作指令, 参与者执行除了事务提交外所有操作
• 如参与者执行成功, 给协调者反馈执行成功, 否则反馈中止, 表示事务失败

第二阶段: 提交阶段

协调者收到各个参与者的准备消息后, 根据反馈情况通知各个参与者commit提交或者rollback回滚

commit提交

当第一阶段所有参与者都反馈成功时, 协调者发起正式提交事务的请求, 当所有参与者都回复提交成功时, 则意味着完成事务

• 协调者节点向所有参与者发出正式提交的 commit 请求。
• 收到协调者的 commit 请求后, 参与者正式执行事务提交操作, 并释放在整个事务期间内占用的资源
• 参与者完成事务提交后, 向协调者节点发送已提交消息
• 协调者节点收到所有参与者节点反馈的已提交消息后, 完成事务

rollback回滚

如果任意一个参与者节点在第一阶段返回的消息为中止(或者异常), 或者协调者节点在第一阶段的询问超时, 无法获取到全部参数者反馈, 那么这个事务将会被回滚。

• 协调者向所有参与者发出 rollback 回滚操作的请求
• 参与者执行事务回滚, 并释放在整个事务期间内占用的资源
• 参与者在完成事务回滚之后, 向协调者发送回滚完成的反馈消息
• 协调者收所有参与者反馈的消息后, 取消事务
  • 发现没有, 无论是准备命令, ommit, rollback, 都是要搜集到所有的RM的反馈, 才做决定的

优缺点

缺点

• 性能问题: 执行过程中, 所有参与节点都是事务阻塞性的, 当参与者占有公共资源时, 其他第三方节点访问公共资源就不得不处于阻塞状态, 为了数据的一致性而牺牲了可用性, 对性能影响较大, 不适合高并发高性能场景

• 可靠性问题: 2PC非常依赖协调者, 当协调者发生故障时, 尤其是第二阶段, 那么所有的参与者就会都处于锁定事务资源的状态中, 而无法继续完成事务操作

• 数据一致性问题: 在阶段二中, 当协调者向参与者发送commit请求之后, 发生了局部网络异常或者在发送commit请求过程中协调者发生了故障, 这回导致只有一部分参与者接受到了commit请求。而在这部分参与者接到commit请求之后就会执行commit操作。但是其他部分未接到commit请求的机器则无法执行事务提交。于是整个分布式系统便出现了数据不一致性的现象

• 二阶段无法解决的问题: 协调者在发出 commit 消息之后宕机, 而唯一接收到这条消息的参与者同时也宕机了, 那么即使协调者通过选举协议产生了新的协调者, 这条事务的状态也是不确定的, 没人知道事务是否被已经提交

优点

• 尽量保证了数据的强一致, 适合对数据强一致要求很高的关键领域

方案2: 3PC

概念

3PC, 三阶段提交协议, 是二阶段提交协议的改进版本, 以解决2PC存在的缺陷问题, 具体改进如下:

• 在协调者和参与者中都引入超时机制

• 引入确认机制, 当所有参与者能正常工作才执行事务

所以3PC分为3个阶段: CanCommit 准备阶段、PreCommit 预提交阶段、DoCommit 提交阶段

第一阶段: CanCommit 准备阶段

协调者向参与者发送 canCommit 请求, 参与者如果可以提交就返回YES响应, 否则返回NO响应, 具体流程如下:

• 事务询问: 协调者™向所有参与者(RM)发出包含事务内容的canCommit 请求, 询问是否可以提交事务, 并等待所有参与者(RM)答复
• 响应反馈: 参与者(RM)收到canCommit请求后, 如果认为可以执行事务操作, 则反馈 YES 并进入预备状态, 否则反馈NO

第二阶段: PreCommit 阶段

协调者根据参与者的反应情况来决定是否可以进行事务的PreCommit 操作。根据响应情况, 有以下两种可能:

执行事务: 返回都是YES

所有参与者™向协调者(RM)发送了YES响应, RM将会执行本地事务但是不提交, 如下

• 协调者™向参与者(RM)发送 PreCommit 请求, 并进入准备阶段
• 参与者(RM)接收到PreCommit请求后, 会执行本地事务操作, 但不提交事务
• 参与者成功的执行了事务操作后, 返回ACK响应, 同时开始等待最终指令。

中断事务: 返回存在NO

如果存在一个参与者(RM)向协调者™发送了NO响应, 或者等待超时之后, 协调者™都没有接到参与者(RM)的响应, 那么就执行事务的中断

• 协调者™向所有参与者(RM)发送abort请求(即发送中断事务请求)
• 参与者(RM)收到来自协调者™的abort请求之后(或超时之后, 仍未收到协调者的请求), 执行事务的中断

第三阶段: doCommit阶段

该阶段进行真正的事务提交, 也存在2种情况:

提交事务: 返回都是YES

第二阶段的preCommit 请求, 所有参与者向协调者发送了YES响应, 将会提交事务

• 协调接收到所有参与者发送的ACK响应, 那么他将从预提交状态进入到提交状态, 并向所有参与者发送doCommit请求(真正执行commit操作的阶段)
• 参与者接收到doCommit请求之后, 执行正式的事务提交, 并在完成事务提交之后释放所有事务资源
• 事务提交完之后, 参与者向协调者发送ack响应
• 协调者接收到所有参与者的ack响应之后, 完成事务。

中断事务: 返回存在NO

如果存在一个参与者向协调者发送了NO响应, 或者等待超时之后, 协调者都没有接到参与者的响应, 那么就执行事务的中断

• 协调者处向所有参与者发出abort请求
• 参与者接收到abort请求之后, 马上回滚事务, 释放所有的事务资源
• 参与者完成事务回滚之后, 向协调者反馈ACK消息
• 协调者接收到参与者反馈的ACK消息之后, 执行事务的中。

优缺点

缺点

• 数据不一致问题依然存在, 当在参与者(RM)收到 preCommit 请求后等待 doCommit 指令时, 此时如果协调者™请求中断事务, 协调者无法与参与者正常通信, 会导致参与者继续提交事务, 造成数据不一致。

优点

• 相比二阶段提交, 三阶段提交降低了阻塞范围, 在等待超时后协调者或参与者会中断事务。避免了协调者™单点问题, 阶段3中协调者出现问题时, 参与者会继续提交事务

AP解决方案

方案3: TCC

概念

TCC(Try Confirm Cancel)方案是一种应用层面侵入业务的两阶段提交。是目前最火的一种分布式事务方案, 其核心思想是: 针对每个操作, 都要注册一个与其对应的确认和补偿(撤销)操作

怎么理解？看下图

一个完整的TCC业务由一个主业务服务和若干个从业务服务组成, 主业务服务发起并完成整个业务活动, TCC 模式要求从服务提供三个接口: Try、Confirm、Cancel

Try(尝试)

• 这个过程并未执行业务, 只是完成所有业务的一致性检查, 并预留好执行所需的全部资源

核心点: 一致性检查和预留资源

Confirm(确认)

• 确认执行业务操作, 不做任何业务检查, 只使用Try阶段预留的业务资源。通常情况下, 采用TCC则认为 Confirm阶段是不会出错的。即: 只要Try成功, Confirm一定成功。若Confirm阶段真的出错了, 需引入重试机制或人工处理

Cancel(取消)

• 取消Try阶段预留的业务资源。通常情况下, 采用TCC则认为Cancel阶段也是一定成功的。若Cancel阶段真的出错了, 需引入重试机制或人工处理

**注意要点: **

并且由于confirm 或者cancel有可能会重试, 因此对应的部分需要支持幂等(防重表或逐渐insert保持幂等)

经典案例

需求: 用户下单, 扣库存, 扣款

假设: 一个商品单价为100, 库存总数10, 现在要买购买2件, 当前账户余额1000, 购买应该扣款200, 余额为800

优缺点

优点

• 性能提升: 具体业务来实现, 控制资源锁的粒度变小, 不会锁定整个资源。
• 数据最终一致性: 基于Confirm和Cancel 的幂等性, 保证事务最终完成确认或者取消, 保证数据的一致性。
• 可靠性: 解决了XA协议的协调者单点故障问题, 由主业务方发起并控制整个业务活动, 业务活动管理器也变成多点, 引入集群

缺点

• TCC 的Try、Confirm 和 Cancel操作功能要按具体业务来实现, 业务耦合度较高, 提高了开发成本

方案4: SAGA

概念

Saga是分布式事务领域最有名气的解决方案之一, 最初出现在1987年Hector Garcaa-Molrna & Kenneth Salem发表的论文SAGAS里。

Saga是由一系列的本地事务构成。每一个本地事务在更新完数据库之后, 会发布一条消息或者一个事件来触发Saga中的下一个本地事务的执行。如果一个本地事务因为某些业务规则无法满足而失败, Saga会执行在这个失败的事务之前成功提交的所有事务的补偿操作(重点是补偿)。

Saga的实现有很多种方式, 其中最流行的两种方式是:

• 命令协调(Order Orchestrator): 这种方式的工作形式就像一只乐队, 由一个指挥家(协调中心)来协调大家的工作。协调中心来告诉Saga的参与方应该执行哪一个本地事务
• 事件编排(Event Choreographyo): 这种方式没有协调中心, 整个模式的工作方式就像舞蹈一样, 各个舞蹈演员按照预先编排的动作和走位各自表演, 最终形成一只舞蹈。处于当前Saga下的各个服务, 会产生某类事件, 或者监听其它服务产生的事件并决定是否需要针对监听到的事件做出响应

命令协调

中央协调器(Orchestrator, 简称 OSO)以命令/回复的方式与每项服务进行通信, 全权负责告诉每个参与者该做什么以及什么时候该做什么

• 有点类似于事务管理器™的感觉

• 主业务接口发起事务业务, 开启订单事务
• Saga协调器库存服务请求扣减库存, 库存服务操作后, 回复处理结果
• Saga协调器账户服务请求扣减余额, 账户服务操作后, 回复处理结果
• Saga协调器订单服务请求创建订单, 订单服务操作后, 回复处理结果
• 主业务逻辑接收并处理Saga协调器事务处理结果回复

中央协调器OSO必须事先知道执行整个事务所需的流程, 如果有任何失败, 它还负责通过向每个参与者发送命令来撤销之前的操作来协调分布式的回滚, 基于中央协调器协调一切时, 回滚要容易得多, 因为协调器默认是执行正向流程, 回滚时只要执行反向流程即可

执行顺序: A–>B–>C 回滚顺序: C–>B—>A

事件编排

在基于事件的方式中, 第一个服务执行完本地事务之后, 会产生一个事件。其它服务会监听这个事件, 触发该服务本地事务的执行, 并产生新的事件。当最后一个服务执行本地事务并且不发布任何事件时, 或者发布的事件没有被任何Saga参与者监听到, 都意味着事务结束

• 主业务接口发布下单事件
• 库存服务监听下单事件, 扣减库存, 并发布库存已扣减事件
• 账户服务监听已扣减库存事件, 扣减余额, 并发已扣减余额事件
• 订单服务监听已扣减余额事件, 创建订单, 并发布下单成功事件
• 主业务逻辑监听下单成功事件后, 执行后续处理

异常恢复

前面讲到saga模式, 在本地事务因为某些业务规则无法满足而失败, Saga会执行在这个失败的事务之前成功提交的所有事务的补偿操作

上面意思可以理解为, saga模式下, 每个事务参与者提供一对接口, 一个做正常事务操作, 一个做异常事务回滚操作。比如: 支付与退款, 扣款与回补等

saga支持事务恢复策略

向后恢复(backward recovery)

当执行事务失败时, 补偿所有已完成的事务, 是"一退到底"的方式, 这种做法的效果是撤销掉之前所有成功的子事务, 使得整个 Saga 的执行结果撤销。

从上图可知事务执行到了支付事务T3, 但是失败了, 因此事务回滚需要从C3 -> C2 -> C1依次进行回滚补偿, 对应的执行顺序为: T1 -> T2 -> T3 -> C3 -> C2 -> C1。

向前恢复(forward recovery)

对于执行不通过的事务, 会尝试重试事务, 这里有一个假设就是每个子事务最终都会成功, 这种方式适用于必须要成功的场景, 事务失败了重试, 不需要补偿

优缺点

命令协调设计

优点

• 服务之间关系简单, 避免服务间循环依赖, 因为 Saga 协调器会调用Saga参与者, 但参与者不会调用协调器
• 程序开发简单, 只需要执行命令/回复(其实回复消息也是一种事件消息), 降低参与者的复杂性
• 易维护扩展, 在添加新步骤时, 事务复杂性保持线性, 回滚更容易管理, 更容易实施和测试

缺点

• 中央协调器处理逻辑容易变得庞大复杂, 导致难以维护
• 存在协调器单点故障风险

事件编排设计

优点

• 避免中央协调器单点故障风险。
• 当涉及的步骤较少服务开发简单, 容易实现。

缺点

• 服务之间存在循环依赖的风险。
• 当涉及的步骤较多, 服务间关系混乱, 难以追踪调测。

命令/事件对比和选择

命令协调方式与事件编排方式怎么选择？

• 系统复杂性: 如果系统的业务逻辑复杂, 事务需要严格控制和编排, 命令方式可以提供更好的可见性和可控性。
• 系统扩展性: 如果系统需要频繁扩展和修改, 需要一定的灵活性, 事件方式可以提供解耦和扩展性更好的架构。
• 性能需求: 如果需要更好的性能和可伸缩性, 并行执行事务的各个步骤, 事件方式更适合。
• 异步需求: 如果系统需要异步处理和解耦, 事件方式提供了更好的可行性。

方案5: 本地消息表

操作流程

**本地消息表的核心思路就是将分布式事务拆分成本地事务进行处理(到底怎么拆就精髓了!), 当消息表状态变更了, **

在该方案中主要有两种角色: 事务主动方和事务被动方

• 事务主动方: 需要额外新建事务消息表, 并在本地事务中完成业务处理和记录事务消息, 并轮询事务消息表的数据, 将事务消息发送到MQ
• 事务被动方: 基于MQ消费事务消息表中的事务

操作步骤:

1. 发生分布式事务操作时, 事务主动方在DB1中的操作业务表, 记录事务信息在消息表中, 状态为未处理

2. 事务主动方: 往MQ中推送一个事务操作消息, 并通知事务被动方处理事务消息。

3. 事务被动方: 监控MQ, 读取事务消息, 完成DB2中业务操作, 往MQ返回ack

4. 事务主动方: 监控MQ, 读取事务消息, 更新事务消息表状态为已处理

异常情况处理:

1. 当1处理出错, 事务主动方在本地事务中, 直接回滚就行

2. 当2处理出错, 由于DB1中还是保存事务消息, 可以设置轮询逻辑, 将消息重新推送给消息中间件, 在通知事务被动方。

3. 当3处理出错, 重复获取消息, 重复执行即可

4. 如果是业务上处理失败, 事务被动方可以发消息给事务主动方回滚事务

5. 如果事务被动方已经消费了消息, 事务主动方需要回滚事务的话, 需要发消息通知事务主动方进行回滚事务。

优缺点

优点

• 从应用设计开发的角度实现了消息数据的可靠性, 消息数据的可靠性不依赖于消息中间件, 弱化了对MQ中间件特性的依赖。
• 方案轻量, 容易实现

缺点

• 与具体的业务场景绑定, 耦合性强, 不可公用
• 消息数据与业务数据同库, 占用业务系统资源
• 业务系统在使用关系型数据库的情况下, 消息服务性能会受到关系型数据库并发性能的局限

方案6: MQ消息事务(本地消息表进阶版)

操作流程

基于MQ的分布式事务方案本质上是对本地消息表的封装, 整体流程与本地消息表一致, 唯一不同的就是将本地消息表存在了MQ内部, 而不是业务数据库中

以RocketMQ为例子

事务消息发送及提交

1. 程序调用MQ发送方发送消息(这个消息是半消息, 因为还不能立马消费, 需要二次确认)

2. MQ服务端进行确认, 将响应消息写入MQ

3. 程序根据发送结果执行本地事务(如果写入失败,此时半消息对业务不可见, 本地逻辑不执行)

4. 根据本地事务状态执行Commit或者Rollback(Commit操作生产消息索引, 消息对消费者可见)

事务补偿

(1) 对没有Commit/Rollback的事务消息(pending状态的消息), 从服务端发起一次"回查"

(2) Producer收到回查消息, 检查回查消息对于的本地事务的状态

(3) 根据本地事务状态, 重新Commit或者Rollback

其中, 补偿阶段用户解决消息Commit或者Rollback发生超时或者失效的情况

事务消息状态

事务消息共有三种状态: 提交状态, 回查状态, 中间状态:

• TransactionStatus.CommitTransaction: 提交事务, 它允许消费者消费此消息
• TransactionStatus.RollbackTransaction: 回滚事务, 它代表消息将被删除, 不允许被消费
• TransactionStatus.UnkNOwn: 中间状态, 它代表需要消息队列来确认状态

优缺点

优点

• 消息数据独立存储 , 降低业务系统与消息系统之间的耦合
• 吞吐量大于使用本地消息表方案

缺点

• 一次消息发送需要两次网络请求(半消息 + commit/rollback 消息) 。
• 业务处理服务需要实现消息状态回查接口

方案7: 最大努力通知(MQ消息事务进阶版)

最大努力通知也称为定期校对, 是对MQ事务方案的进一步优化。它在事务主动方增加了消息校对的接口, 如果事务被动方没有接收到主动方发送的消息, 此时可以调用事务主动方提供的消息校对的接口主动获取

在可靠消息事务中, 事务主动方需要将消息发送出去, 并且让接收方成功接收消息, 这种可靠性发送是由事务主动方保证的；但是最大努力通知, 事务主动方仅仅是尽最大努力(重试, 轮询…)将事务发送给事务接收方, 所以存在事务被动方接收不到消息的情况, 此时需要事务被动方主动调用事务主动方的消息校对接口查询业务消息并消费, 这种通知的可靠性是由事务被动方保证的

分布式事务方案选择

属性	2PC/3PC	TCC	Saga	本地消息表	尽最大努力通知(MQ)
事务一致性	强	弱	弱	弱	弱
复杂性	中	高	中	低	低
业务侵入性	小	大	小	中	中
使用局限性	大	大	中	小	中
性能	低	中	高	高	高
维护成本	低	高	中	低	中

• 2PC/3PC: 依赖于数据库, 能够很好的提供强一致性和强事务性, 但延迟比较高, 比较适合传统的单体应用, 在同一个方法中存在跨库操作的情况, 不适合高并发和高性能要求的场景。
• TCC: 适用于执行时间确定且较短, 实时性要求高, 对数据一致性要求高, 比如互联网金融企业最核心的三个服务: 交易、支付、账务。
• 本地消息表/MQ 事务: 适用于事务中参与方支持操作幂等, 对一致性要求不高, 业务上能容忍数据不一致到一个人工检查周期, 事务涉及的参与方、参与环节较少, 业务上有对账/校验系统兜底
• Saga 事务: 由于Saga事务不能保证隔离性, 需要在业务层控制并发, 适合于业务场景事务并发操作同一资源较少的情况。Saga 由于缺少预提交动作, 导致补偿动作的实现比较麻烦, 例如业务是发送短信, 补偿动作则得再发送一次短信说明撤销, 用户体验比较差。所以, Saga事务较适用于补偿动作容易处理的场景

四、分布式事务最佳实践-Seata

简介

官网: http://seata.io/zh-cn/

源码: https://github.com/seata/seata

Seata 是一款开源的分布式事务解决方案, 致力于提供高性能和简单易用的分布式事务服务。Seata 将为用户提供了AT、TCC、SAGA 和 XA 事务模式, 为用户打造一站式的分布式解决方案。

常见术语

TC (Transaction Coordinator) - 事务协调者: 维护全局和分支事务的状态, 驱动全局事务提交或回滚

TM (Transaction Manager) - 事务管理器: 定义全局事务的范围: 开始全局事务、提交或回滚全局事务

RM (Resource Manager) - 资源管理器: 管理分支事务处理的资源, 与TC交谈以注册分支事务和报告分支事务的状态, 并驱动分支事务提交或回滚

Seata配置

下载

官网: https://github.com/seata/seata/releases/tag/v1.7.0

修改配置文件

1.5.0之前的版本配置文件是有多个的, 都位于conf文件夹下, 如file.conf,registry,conf等。在1.5.0版本之后都整合到一个配置文件里了, 即application.yml。以下配置项请按照自己版本查找修改。

以seata-1.7.0为例, 打开conf/application.yml进行修改, 重点修改nacos部分配置。

其中的: application.example.yml 各种注册中心, 配置中心配置方式, 默认是配置本地, 这里以配置在nacos为例子。

server:port: 7091spring:application:name: seata-server# 日志相关配置
logging:config: classpath:logback-spring.xmlfile:# 日志存放的本地地址, 更改成你自己的地址, 这里path: ${user.home}/logs/seata# 日志推送        extend:# 推送到logstash中logstash-appender:destination: 127.0.0.1:4560# 推送到kafka中kafka-appender:bootstrap-servers: 127.0.0.1:9092# MQ的主题topic: logback_to_logstashconsole:user:username: seatapassword: seata
seata:# 配置中心config:# 支持的配置中心: nacos, consul, apollo, zk, etcd3# 这里将seata的配置存储到nacos中type: nacosnacos:server-addr: 127.0.0.1:8848namespace:group: SEATA_GROUPusername:password:context-path:##if use MSE Nacos with auth, mutex with username/password attribute#access-key:#secret-key:data-id: seataServer.yaml# 注册中心registry:# 支持的注册中心: nacos, eureka, redis, zk, consul, etcd3, sofatype: nacos# nacos配置nacos:application: seata-serverserver-addr: 127.0.0.1:8848group: SEATA_GROUPnamespace:cluster: defaultusername:password:context-path:##if use MSE Nacos with auth, mutex with username/password attribute#access-key:#secret-key:# 存储模式,指的是Seata服务器用于记录全局事务状态和分支事务状态的后端存储系统# file: 默认的存储模式，它将事务状态信息存储在本地文件系统中。这种模式适用于单节点测试或小规模部署# db: 将事务状态信息存储在数据库中# redis: 将事务状态信息存储在Redis中store:# support: file 、 db 、 redismode: file#  server:#    service-port: 8091 # 如果你没有配置service-port, 那么默认就是 ${server.port} + 1000security:# Seata安全密钥secretKey: SeataSecretKey0c382ef121d778043159209298fd40bf3850a017# token有效期: 60 mintokenValidityInMilliseconds: 1800000igNOre:# 不需要认证的URL路径     urls: /,/**/*.css,/**/*.js,/**/*.html,/**/*.map,/**/*.svg,/**/*.png,/**/*.jpeg,/**/*.ico,/api/v1/auth/login

修改成功后, 意味着seata将从nacos获取配置信息, 同时注册自身服务到nacos中心

nacos配置

上面配置项中有一项: seata.config.data-id=seataServer.yaml, 意思为要读nacos上的seataServer.yaml配置文件, 接下来去Nacos创建该配置文件, 注意Group与第2步中的保持一致, 这里是SEATA_GROUP。

配置内容从seata-server-1.7.0/seata/script/config-center/config.txt粘贴修改而来, 这里只使用对我们有用的配置, 主要是数据库配置信息。

store: # 事务存储配置，仅适用于服务器mode: dblock: mode: dbsession: mode: db# db存储模式相关的配置	db: # 数据源配置datasource: druiddbType: mysqldriverClassName: com.mysql.cj.jdbc.Driverurl: jdbc: mysql: //127.0.0.1:3306/seata?useSSL=false&useUnicode=true&rewriteBatchedStatements=trueuser: rootpassword: adminminConn: 5maxConn: 30# 存储全局事务、分支事务、分布式锁和锁表的名称globalTable: global_tablebranchTable: branch_tabledistributedLockTable: distributed_locklockTable: lock_table# 每次查询获取的事务数量queryLimit: 100# 分布式锁的最大等待时间, 单位是毫秒maxWait: 5000

数据库建表

在seata数据库内, 执行seata-server-1.7.0/seata/script/server/db目录下的sql脚本(根据数据库类型), 创建服务端所需的表。此处选择: mysql

-- -------------------------------- The script used when storeMode is 'db' --------------------------------
-- the table to store GlobalSession data
CREATE TABLE IF NOT EXISTS `global_table`
(`xid`                       VARCHAR(128) NOT NULL,`transaction_id`            BIGINT,`status`                    TINYINT      NOT NULL,`application_id`            VARCHAR(32),`transaction_service_group` VARCHAR(32),`transaction_name`          VARCHAR(128),`timeout`                   INT,`begin_time`                BIGINT,`application_data`          VARCHAR(2000),`gmt_create`                DATETIME,`gmt_modified`              DATETIME,PRIMARY KEY (`xid`),KEY `idx_status_gmt_modified` (`status` , `gmt_modified`),KEY `idx_transaction_id` (`transaction_id`)
) ENGINE = InNODBDEFAULT CHARSET = utf8mb4;-- the table to store BranchSession data
CREATE TABLE IF NOT EXISTS `branch_table`
(`branch_id`         BIGINT       NOT NULL,`xid`               VARCHAR(128) NOT NULL,`transaction_id`    BIGINT,`resource_group_id` VARCHAR(32),`resource_id`       VARCHAR(256),`branch_type`       VARCHAR(8),`status`            TINYINT,`client_id`         VARCHAR(64),`application_data`  VARCHAR(2000),`gmt_create`        DATETIME(6),`gmt_modified`      DATETIME(6),PRIMARY KEY (`branch_id`),KEY `idx_xid` (`xid`)
) ENGINE = InNODBDEFAULT CHARSET = utf8mb4;-- the table to store lock data
CREATE TABLE IF NOT EXISTS `lock_table`
(`row_key`        VARCHAR(128) NOT NULL,`xid`            VARCHAR(128),`transaction_id` BIGINT,`branch_id`      BIGINT       NOT NULL,`resource_id`    VARCHAR(256),`table_name`     VARCHAR(32),`pk`             VARCHAR(36),`status`         TINYINT      NOT NULL DEFAULT '0' COMMENT '0:locked ,1:rollbacking',`gmt_create`     DATETIME,`gmt_modified`   DATETIME,PRIMARY KEY (`row_key`),KEY `idx_status` (`status`),KEY `idx_branch_id` (`branch_id`),KEY `idx_xid` (`xid`)
) ENGINE = InNODBDEFAULT CHARSET = utf8mb4;CREATE TABLE IF NOT EXISTS `distributed_lock`
(`lock_key`       CHAR(20) NOT NULL,`lock_value`     VARCHAR(20) NOT NULL,`expire`         BIGINT,primary key (`lock_key`)
) ENGINE = InNODBDEFAULT CHARSET = utf8mb4;INSERT INTO `distributed_lock` (lock_key, lock_value, expire) VALUES ('AsyncCommitting', ' ', 0);
INSERT INTO `distributed_lock` (lock_key, lock_value, expire) VALUES ('RetryCommitting', ' ', 0);
INSERT INTO `distributed_lock` (lock_key, lock_value, expire) VALUES ('RetryRollbacking', ' ', 0);
INSERT INTO `distributed_lock` (lock_key, lock_value, expire) VALUES ('TxTimeoutCheck', ' ', 0);

启动seata-server

运行bin下的bat脚本启动服务。

访问: http://127.0.0.1:7091

默认账号与密码都是seata

Seata模式使用

XA模式

概念

XA规范是X/Open组织定义的分布式事务处理(DTP)标准, XA规范描述了全局的TM与局部RM之间的接口, 几乎所有主流的数据库都对XA规范提供了支持。

原理

XA是一种分布式事务协议, 用于在分布式环境下实现事务的一致性。它通过将本地事务(Local Transaction)与全局事务(Global Transaction)进行协调来实现分布式事务。

XA模式使用了一组标准的API(如Java中的javax.transaction包), 通常包括两个阶段: 事务准备(Prepare)和事务提交(Commit)。

思考: XA与2PC有啥区别？

注意点: XA模式可以使用2PC协议来实现全局事务的一致性。也就是说, XA模式是2PC的一种具体实现方式, 并且在XA模式中可以利用其他机制(如XA事务日志)来增强事务的持久性和可靠性。因此, 可以将XA视为2PC的一个变种或扩展。

seata的XA模式做了一些调整

操作流程:

第一阶段:

1. TM去TC中注册全局事务

2. TM调用RM事务接口, 注册分支事务

3. 执行RM事务操作, 不提交

4. RM往TC报告事务状态

第二阶段:

1. 所有RM执行完本地事务, TM发起全局事务提交/回滚(注意咯, 这里全局事务不是我们实际业务的事务, 是一个虚拟操作, 主要就是通过TC进行检查)

2. TC检查所有RM事务状态是YES还是NO?

   • 全部YES: 通知所有RM提交事务
   • 其中有一个为NO: 通知所有RM回滚事务

​

代码

项目集成seata

依赖

所有微服务导入seata依赖, 这里引入1.7.0的为例子(将默认的Spring Cloud中的seata排除掉), 实际中, 用对应的版本就行

<!-- 注意一定要引入对版本, 要引入spring-cloud版本seata, 而不是springboot版本的seata-->
<!-- 排除Spring Cloud的seata版本 -->
<dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-starter-alibaba-seata</artifactId><!-- 排除掉spring cloud默认的seata版本, 以免版本不一致出现问题--><exclusions><exclusion><groupId>io.seata</groupId><artifactId>seata-spring-boot-starter</artifactId></exclusion><exclusion><groupId>io.seata</groupId><artifactId>seata-all</artifactId></exclusion></exclusions>
</dependency>
<!-- 引入1.7.0 -->
<dependency><groupId>io.seata</groupId><artifactId>seata-spring-boot-starter</artifactId><version>1.7.0</version>
</dependency>

配置文件

在application.yml文件中配置, 每个微服务都要, 订单, 库存, 钱包都需添加上

# seata客户端配置
seata:# 开启seata客户端, 开启后将会与Seata服务端进行通信enabled: true# 当前应用的ID。在同一个Seata环境中，每个应用的ID必须是唯一的application-id: seata_tx# 事务分组的名称。在同一个分组下的应用可以共享同一个全局事务IDtx-service-group: seata_tx_group# 虚拟组和映射关系service:# default映射到哪个数据源代理，则该虚拟组下的所有事务都会使用这个数据源代理vgroup-mapping:seata_tx_group: default# 注册中心的配置registry:type: nacosnacos:application: seata-serverserver-addr: 127.0.0.1:8848namespace:group: SEATA_GROUP# 数据源代理模型, 支持的模式有: AT、XA、TCC、SAGAdata-source-proxy-mode: XA

其中seata_tx_group为我们自定义的事务组, 名字随便起, 但是下面service.vgroup-mapping下一定要有一个对应这个名字的映射, 映射到default(seata默认的集群名称)。 nacos方面, 我们仅配置注册项, 即registry下的配置, 配置内容与服务端保持一致

配置全局事务

在business-service服务的purchase 方法中加上全局事务注解: @GlobalTransactional

@Override
@GlobalTransactional // 核心就是这个全局事务注解
public void purchase(String userId, String commodityCode, int orderCount, boolean rollback) {// 扣库存String result = stockFeignClient.deduct(commodityCode, orderCount);// 扣库存失败, 回滚库存if (!"SUCCESS".equals(result)) {throw new RuntimeException("库存服务调用失败,事务回滚!");}// 创建订单result = orderFeignClient.create(userId, commodityCode, orderCount);// 创建订单失败, 回滚库存if (!"SUCCESS".equals(result)) {throw new RuntimeException("订单服务调用失败,事务回滚!");}if (rollback) {throw new RuntimeException("强制回滚... ");}
}

启动seata

测试

正常: http://localhost:8088/businesses/purchase?rollback=false&count=2

超库存: http://localhost:8088/businesses/purchase?rollback=false&count=12

超余额: http://localhost:8088/businesses/purchase?rollback=false&count=8

优缺点

优点

• 事务强一致性, 满足ACID原则
• 常用的数据库都支持, 实现简单, 并且没有代码侵入

缺点

• 第一阶段锁定数据库资源, 等待二阶段结束才释放, 锁定资源过长, 性能较差
• 依赖关系型数据库的实现事务

AT模式

概念

AT模式同样是分阶段提交事务模式, 操作起来算是XA模式的优化版

XA模式在第一阶段存在锁定资源的操作, 时间长之后会影响性能

AT模式在第一阶段直接提交事务, 弥补了XA模式中资源锁定周期过长缺陷

原理

操作流程:

第一阶段:

1. 注册全局事务

2. 调用RM事务接口, 注册分支事务

3. 执行RM事务操作, 并提交, 记录undo log日志快照

4. 往TC报告事务状态

第二阶段:

1. 所有RM执行完本地事务, TM发起全局事务提交/回滚

2. TC检查所有RM事务状态, YES or NO?

   • 全部YES: 通知所有RM提交事务, 删除undo log日志快照
   • 其中一个为NO: 通知所有RM回滚事务, 恢复undo log日志快照

XA vs AT

• XA模式一阶段不提交事务, 锁定资源； AT模式一阶段直接提交, 不锁定资源
• XA模式依赖数据库实现回滚； AT利用undo log实现数据回滚
• XA模式强一致性；AT模式最终一致(一阶段提交, 此时有事务查询, 就存在不一致)

AT存在隔离问题(更新丢失问题)

AT模式因为在全局事务中第一阶段就提交了事务, 释放资源。如果这个时, 另外RM/外部事务(非RM)操作相同资源, 可能存在读写隔离问题(更新丢失问题)。

问题出现原理

读写隔离问题-2个seata事务解决方案

加全局锁

读写隔离问题-1个seata事务 + 非seata事务解决方案

全局锁+多级快照

代码

配置seata-AT相关快照/全局锁/快照表数据库

配置数据库

TM数据库-seata

源sql位置: eata-server-1.7.0/script/server/db 中

各个RM数据库

源sql: https://seata.io/zh-cn/docs/dev/mode/at-mode.html

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;

配置文件

在application.yml文件中配置, 每个微服务都要

# seata客户端配置
seata:enabled: trueapplication-id: seata_txtx-service-group: seata_tx_groupservice:vgroup-mapping:seata_tx_group: defaultregistry:type: nacosnacos:application: seata-serverserver-addr: 127.0.0.1:8848namespace:group: SEATA_GROUP# 需要注意的就是这里的代理模式修改成了ATdata-source-proxy-mode: AT

测试

正常: http://localhost:8088/businesses/purchase?rollback=false&count=2

超库存: http://localhost:8088/businesses/purchase?rollback=false&count=12

超余额: http://localhost:8088/businesses/purchase?rollback=false&count=8

优缺点

优点

• 一阶段完成直接提交事务, 释放资源, 性能较好
• 利用全局锁实现读写隔离
• 没有代码侵入, 框架自动完成回滚与提交

缺点

• 两阶段之间存在数据不一致情况, 只能保证最终一致
• 框架的快照功能影响性能, 但比XA模式要好很多

TCC模式

概念

TCC模式的seata版实现。TCC模式与AT模式非常相似, 每阶段都是独立事务, 不同的TCC通过人工编码来实现数据恢复

• Try: 资源的检测和预留
• Confirm: 完成资源操作业务；要求Try 成功, Confirm一定要能成功
• Cancel: 预留资源释放, 可以理解为try的反向操作

原理

操作流程:

1. 注册全局事务

2. 调用RM事务接口, 注册分支事务

3. 执行RM事务try接口, 检查资源, 预留资源

4. 往TC报告事务状态

5. 所有RM执行完本地事务, TM发起全局事务提交/回滚

6. TC检查所有RM事务状态, YES or NO?

   • 全部YES: 通知所有RM执行confirm接口, 提交事务
   • 其中一个为NO: 通知所有RM执行cancel接口, 回滚事务

案例演示

存在问题

TCC模式中, 在执行Try, 执行Confirm, 执行Cancel 过程中会出现意外情况, 导致TCC模式经典问题: 空回滚, 业务悬挂, 重试幂等问题

空回滚

当某个分支事务try阶段阻塞时, 可能导致全局事务超时而触发二阶段的cancel操作, RM在没有执行try操作就执行cancel操作, 此时cancel无数据回滚, 这就是空回滚

业务悬挂

当发生的空回滚之后, 当阻塞的Try正常了, RM先执行空回滚(cancel)后, 又收到Try操作指令, 执行业务操作, 并冻结资源。但是事务已经结束, 不会再有confirm 或cancel了, 那刚执行try操作冻结资源, 就被悬挂起来了。这就是业务悬挂

重试幂等

因为网络抖动等原因, TC下达的Confirm/Cancel指令可能出现延时, 发送失败等问题, 此时TC会启用重试机制, 到时, RM可能收到多个confirm或cancel指令, 这就要求confirm接口或者cancel接口, 需要能够保证幂等性。

幂等性: 多次执行, 结果都一样

上面空回滚/业务悬挂问题解决, 一般都一起实现: 引入事务状态控制表

表字段: xid->事务, uid->用户id,freeze_money->锁定的资源, 即冻结的数据, state->事务状态(try、confirm/cancel)

以RM: account-service 中用户账户余额为例子。

try:

1. 在状态表中记录冻结金额以及事务状态为try
2. 扣减账户余额

confirm: 根据xid删除状态表中冻结记录

cancel:

1. 修改状态表冻结金额为0, 事务状态改为cancel
2. 恢复账户扣减

如何判断是否为空回滚: 在cancel中, 根据xid查询状态表, 如果查不到, 说明没有执行try阶段(因为只有执行了try阶段, 表中才会有记录), 拒绝执行cancel业务

如果避免业务悬挂: try业务中, 根据xid查询状态表(查state为try的), 如果查到了, 说明已经执行过cancel已经执行过, 拒绝执行try业务

重试幂等: 需要引入唯一标识, 比如第一次操作成功留下, 唯一标识, 下次来识别这个标识

代码

在AT模式基础上做代码TCC改造就行

TCC模式中核心注解解析

@LocalTCC

在 Seata TCC 模式中, @LocalTCC注解的作用是: 标注一个本地TCC事务方法。@LocalTCC包含 Try、Confirm、Cancel 三个操作

@LocalTCC 注解修饰的类有以下方法:

• confirmMethod: Confirm方法名称
• cancelMethod: Cancel方法名称
• tryMethod: Try方法名称

一个典型的@LocalTCC示例如下:

@LocalTCC
public class LocalTCCTransaction {@Transactional public void tryMethod() {//do something}public void confirmMethod() {//confirm }public void cancelMethod() {//cancel}}

这里三个方法使用同一个事务,他们分别对应 TCC 的 Try、Confirm 和 Cancel 操作。

这样通过**@LocalTCC + @Transactional 注解,就可以实现一个本地TCC事务**

@TwoPhaseBusinessAction 和 @BusinessActionContextParameter

@TwoPhaseBusinessAction: 定义TCC事务try阶段的方法, 作用在方法上

• name: 事务的名称
• commitMethod: 事务confirm方法名
• rollbackMethod: 事务cancel方法名

@BusinessActionContextParameter: 指定try方法的参数作为事务上下文参数, 作用在犯法的参数上

• paramName: 参数名称
• 可以在confirm和cancel方法中通过context获取

例如:

注意咯, 这是在同一个类中的

@TwoPhaseBusinessAction(name="try",commitMethod="confirm",rollbackMethod="cancel")
public void try(Business business) {// try logic
} public void confirm(BusinessActionContext context) {// get param by context  
}public void cancel(BusinessActionContext context) {// get param by context
}

通过这两个注解,可以更方便地实现TCC分布式事务,使代码和业务耦合度更低

account-service服务

新增: IAccountTCCService接口与实现

import io.seata.rm.tcc.api.BusinessActionContext;
import io.seata.rm.tcc.api.BusinessActionContextParameter;
import io.seata.rm.tcc.api.LocalTCC;
import io.seata.rm.tcc.api.TwoPhaseBusinessAction;/*** 钱包账户接口*/
@LocalTCC
public interface IAccountTCCService {/*** try阶段: 预扣款*/@TwoPhaseBusinessAction(name="tryReduce", commitMethod = "confirm", rollbackMethod = "cancel")void tryReduce(@BusinessActionContextParameter(paramName = "userId") String userId, @BusinessActionContextParameter(paramName = "money") int money);/*** confirm阶段: 提交* @param ctx BusinessActionContext* @return*/boolean confirm(BusinessActionContext ctx);/*** cancel阶段: 回滚* @param ctx BusinessActionContext* @return*/boolean cancel(BusinessActionContext ctx);
}

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import io.seata.rm.tcc.api.BusinessActionContext;
import org.springframework.beans.factory.anNOtation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.anNOtation.Transactional;import java.awt.*;/*
* 钱包账户接口
*/
@Service
public class AccountTCCServiceImpl implements IAccountTCCService {@Autowiredprivate AccountMapper accountMapper;@Overridepublic void tryReduce(String userId, int money) {System.err.println("-----------tryReduce-------------");// 校验余额Account one = accountMapper.selectOne(new LambdaQueryWrapper<Account>().eq(Account::getUserId, userId));if(one != null && one.getMoney() < money){throw new RuntimeException("mot enough money ...");}// 扣减余额LambdaUpdateWrapper<Account> wrapper = new LambdaUpdateWrapper<>();wrapper.setSql("money = money - " + money);wrapper.eq(Account::getUserId, userId);accountMapper.update(null, wrapper);}@Overridepublic boolean confirm(BusinessActionContext ctx) {System.err.println("-----------confirm-------------");return true;}@Overridepublic boolean cancel(BusinessActionContext ctx) {// 这里可以获取ctx中存储的userId和money参数, 然后进行补偿System.err.println("-----------cancel-------------");return true;}
}

controller改动

import org.springframework.beans.factory.anNOtation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.anNOtation.*;@Slfj
@RestController
@RequestMapping("accounts")
public class AccountController {//@Autowired//private IAccountService accountService;@Autowiredprivate IAccountTCCService accountTCCService;@GetMapping(value = "/reduce")public String reduce(String userId, int money) {try {accountTCCService.tryReduce(userId, money);} catch (Exception ex) {log.error(ex);return "FAIL";}return "SUCCESS";}
}

order-service服务

新增: IOrderTCCService接口与实现

import com.baomidou.mybatisplus.extension.service.IService;
import io.seata.rm.tcc.api.BusinessActionContext;
import io.seata.rm.tcc.api.BusinessActionContextParameter;
import io.seata.rm.tcc.api.LocalTCC;
import io.seata.rm.tcc.api.TwoPhaseBusinessAction;/*** 订单接口*/
@LocalTCC
public interface IOrderTCCService {/*** try阶段: 预扣款*/@TwoPhaseBusinessAction(name="tryCreate", commitMethod = "confirm", rollbackMethod = "cancel")void tryCreate(@BusinessActionContextParameter(paramName = "userId") String userId, @BusinessActionContextParameter(paramName = "commodityCode") String commodityCode, @BusinessActionContextParameter(paramName = "orderCount") int orderCount);/*** confirm阶段: 提交* @param ctx BusinessActionContext* @return*/boolean confirm(BusinessActionContext ctx);/*** cancel阶段: 回滚* @param ctx BusinessActionContext* @return*/boolean cancel(BusinessActionContext ctx);}

import com.alibaba.nacos.shaded.org.checkerframework.checker.units.qual.A;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import io.seata.rm.tcc.api.BusinessActionContext;
import org.springframework.beans.factory.anNOtation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.anNOtation.Transactional;/*
* 订单接口
*/
@Slfj
@Service
public class OrderTCCServiceImpl implements IOrderTCCService {@Autowiredprivate AccountFeignClient accountFeignClient;@Autowiredprivate OrderMapper orderMapper;@Overridepublic void tryCreate(String userId, String commodityCode, int count) {System.err.println("---------tryCreate-----------");// 定单总价 = 订购数量(count) * 商品单价(100)int orderMoney = count * 100;// 生成订单Order order = new Order();order.setCount(count);order.setCommodityCode(commodityCode);order.setUserId(userId);order.setMoney(orderMoney);orderMapper.insert(order);// 调用账户余额扣减String result = accountFeignClient.reduce(userId, orderMoney);if (!"SUCCESS".equals(result)) {throw new RuntimeException("Account服务调用失败");}}@Overridepublic boolean confirm(BusinessActionContext ctx) {System.err.println("---------confirm-----------");return true;}@Overridepublic boolean cancel(BusinessActionContext ctx) {System.err.println("---------cancel-----------");return true;}
}

controller改动

import org.springframework.beans.factory.anNOtation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.anNOtation.*;@Slfj
@RestController
@RequestMapping("orders")
public class OrderController {@Autowiredprivate IOrderTCCService orderTCCService;@GetMapping(value = "/create")public String create(String userId, String commodityCode, int orderCount) {try {orderTCCService.tryCreate(userId, commodityCode, orderCount);} catch (Exception ex) {log.error(ex);return "FAIL";}return "SUCCESS";}
}

stock-service服务

新增: IStockTCCService接口与实现

import com.baomidou.mybatisplus.extension.service.IService;
import io.seata.rm.tcc.api.BusinessActionContext;
import io.seata.rm.tcc.api.BusinessActionContextParameter;
import io.seata.rm.tcc.api.LocalTCC;
import io.seata.rm.tcc.api.TwoPhaseBusinessAction;/*** 库存接口*/
@LocalTCC
public interface IStockTCCService  {/*** try阶段: 预扣款*/@TwoPhaseBusinessAction(name="tryDeduct", commitMethod = "confirm", rollbackMethod = "cancel")void tryDeduct(@BusinessActionContextParameter(paramName = "commodityCode") String commodityCode, @BusinessActionContextParameter(paramName = "count") int count);/*** confirm阶段: 提交* @param ctx BusinessActionContext* @return*/boolean confirm(BusinessActionContext ctx);/*** cancel阶段: 回滚* @param ctx BusinessActionContext* @return*/boolean cancel(BusinessActionContext ctx);
}

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import io.seata.rm.tcc.api.BusinessActionContext;
import org.springframework.beans.factory.anNOtation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.anNOtation.Transactional;/*** 库存接口*/
@Service
public class StockTCCServiceImpl implements IStockTCCService {@Autowiredprivate StockMapper stockMapper;@Overridepublic void tryDeduct(String commodityCode, int count) {System.err.println("---------tryDeduct-----------");Stock one = stockMapper.selectOne(new LambdaQueryWrapper<Stock>().eq(Stock::getCommodityCode, commodityCode));if(one != null && one.getCount() < count){throw new RuntimeException("not enough count ...");}stockMapper.update(null, new LambdaUpdateWrapper<Stock>().setSql("count = count-" + count).eq(Stock::getCommodityCode, commodityCode));}@Overridepublic boolean confirm(BusinessActionContext ctx) {System.err.println("---------confirm-----------");return true;}@Overridepublic boolean cancel(BusinessActionContext ctx) {System.err.println("---------cancel-----------");return true;}
}

controller改动

import org.springframework.beans.factory.anNOtation.Autowired;
import org.springframework.web.bind.anNOtation.*;@Slfj
@RestController
@RequestMapping("stocks")
public class StockController {@Autowiredprivate IStockTCCService stockTCCService;@GetMapping(value = "/deduct")public String deduct(String commodityCode, int count) {try {stockTCCService.tryDeduct(commodityCode, count);} catch (Exception ex) {log.error(ex);return "FAIL";}return "SUCCESS";}
}

上面操作, 在理想情况下是没有问题的, 但是一旦出现需要回滚操作, 就出问题了, 无法进行数据回补。此时就需要使用到事务状态表实现数据回补, 同时实现空回滚, 避免业务悬挂。

account-service

在seata-account 新增事务状态表

CREATE TABLE `t_account_tx` (`id` bigint NOT NULL AUTO_INCREMENT COMMENT '主键',`tx_id` varchar(100) NOT NULL  COMMENT '事务id',`freeze_money` int DEFAULT NULL COMMENT '冻结金额',`state` int DEFAULT NULL COMMENT '状态 0->try 1->confirm 2->cancel',PRIMARY KEY (`id`)
) ENGINE=InNODB DEFAULT CHARSET=utf8;

新增domain: AccountTX

@Data
@TableName("t_account_tx")
public class AccountTX {public static final int STATE_TRY = 0;public static final int STATE_CONFIRM = 1;public static final int STATE_CANCEL = 2;@TableId(type = IdType.AUTO)private Integer id;private String txId;private int freezeMoney;private int state = STATE_TRY;
}

新增mapper: AccountTXMapper

import com.baomidou.mybatisplus.core.mapper.BaseMapper;public interface AccountTXMapper extends BaseMapper<AccountTX> {}

修改: AccountTCCServiceImpl

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import io.seata.core.context.RootContext;
import io.seata.rm.tcc.api.BusinessActionContext;
import org.checkerframework.checker.units.qual.A;
import org.springframework.beans.factory.anNOtation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.anNOtation.Transactional;import java.awt.*;/*
* 账户接口
*/
@Service
public class AccountTCCServiceImpl implements IAccountTCCService {@Autowiredprivate AccountMapper accountMapper;@Autowiredprivate AccountTXMapper accountTXMapper;@Overridepublic void tryReduce(String userId, int money) {System.err.println("-----------tryReduce-------------" + RootContext.getXID());// 增加业务悬挂判断// RootContext.getXID()获取的是事务idAccountTX accountTX = accountTXMapper.selectOne(new LambdaQueryWrapper<AccountTX>().eq(AccountTX::getTxId, RootContext.getXID()));if (accountTX != null){//存在, 说明已经canel执行过类, 拒绝服务return;}// 余额校验Account one = accountMapper.selectOne(new LambdaQueryWrapper<Account>().eq(Account::getUserId, userId));if(one != null && one.getMoney() < money){throw new RuntimeException("NOt ENOugh Money ...");}// 扣减余额LambdaUpdateWrapper<Account> wrapper = new LambdaUpdateWrapper<>();wrapper.setSql("money = money - " + money);wrapper.eq(Account::getUserId, userId);accountMapper.update(null, wrapper);// 新增账户事务状态AccountTX tx = new AccountTX();tx.setFreezeMoney(money);tx.setTxId(RootContext.getXID());tx.setState(AccountTX.STATE_TRY);accountTXMapper.insert(tx);}@Overridepublic boolean confirm(BusinessActionContext ctx) {System.err.println("-----------confirm-------------");// 事务执行成功, 删除事务状态记录int ret = accountTXMapper.delete(new LambdaQueryWrapper<AccountTX>().eq(AccountTX::getTxId, ctx.getXid()));return ret == 1;}@Overridepublic boolean cancel(BusinessActionContext ctx) {System.err.println("-----------cancel-------------");String userId = ctx.getActionContext("userId").toString();String money = ctx.getActionContext("money").toString();// 增加空回滚校验AccountTX accountTX = accountTXMapper.selectOne(new LambdaQueryWrapper<AccountTX>().eq(AccountTX::getTxId, ctx.getXid()));if (accountTX == null){ // 为空,  空回滚// 新增账户事务状态记录accountTX = new AccountTX();accountTX.setTxId(ctx.getXid());accountTX.setState(AccountTX.STATE_CANCEL);if(money != null){accountTX.setFreezeMoney(Integer.parseInt(money));}accountTXMapper.insert(accountTX);return true;}// 新增幂等处理if(accountTX.getState() == AccountTX.STATE_CANCEL){return true;}// 补偿操作: 恢复余额accountMapper.update(null, new LambdaUpdateWrapper<Account>().setSql("money = money + " + money).eq(Account::getUserId, userId));// 修改账户事务状态表, 状态改成cancelaccountTX.setFreezeMoney(0);accountTX.setState(AccountTX.STATE_CANCEL);int ret = accountTXMapper.updateById(accountTX);return ret == 1;}
}

order-service

在seata-order 新增事务状态表, 因为订单服务不需要做资源锁定(而库存服务需要锁库存, 账户服务需要锁商品的余额), 所以这里没有锁定资源相关的字段

CREATE TABLE `t_order_tx` (`id` bigint NOT NULL AUTO_INCREMENT COMMENT '主键',`tx_id` varchar(100) NOT NULL  COMMENT '事务id',`state` int DEFAULT NULL COMMENT '状态 0->try 1->confirm 2->cancel',PRIMARY KEY (`id`)
) ENGINE=InNODB DEFAULT CHARSET=utf8;

新增domain: OrderTX

import com.baomidou.mybatisplus.anNOtation.IdType;
import com.baomidou.mybatisplus.anNOtation.TableId;
import com.baomidou.mybatisplus.anNOtation.TableName;
import lombok.Data;@Data
@TableName("t_order_tx")
public class OrderTX {public static final int STATE_TRY = 0;public static final int STATE_CONFIRM = 1;public static final int STATE_CANCEL = 2;@TableId(type = IdType.AUTO)private Integer id;private String txId;private int state = STATE_TRY;
}

新增mapper: OrderTXMapper

import com.baomidou.mybatisplus.core.mapper.BaseMapper;public interface OrderTXMapper extends BaseMapper<OrderTX> {
}

修改: OrderTCCServiceImpl

import com.alibaba.nacos.shaded.org.checkerframework.checker.units.qual.A;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import io.seata.core.context.RootContext;
import io.seata.rm.tcc.api.BusinessActionContext;
import org.springframework.beans.factory.anNOtation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.anNOtation.Transactional;@Service
public class OrderTCCServiceImpl implements IOrderTCCService {@Autowiredprivate AccountFeignClient accountFeignClient;@Autowiredprivate OrderMapper orderMapper;@Autowiredprivate OrderTXMapper orderTXMapper;@Overridepublic void tryCreate(String userId, String commodityCode, int count) {System.err.println("---------tryCreate-----------");// 新增业务悬挂判断OrderTX orderTX = orderTXMapper.selectOne(new LambdaQueryWrapper<OrderTX>().eq(OrderTX::getTxId, RootContext.getXID()));if (orderTX != null){//存在, 说明已经canel执行过类, 拒绝服务return;}// 定单总价 = 订购数量(count) * 商品单价(100)int orderMoney = count * 100;// 生成订单Order order = new Order();order.setCount(count);order.setCommodityCode(commodityCode);order.setUserId(userId);order.setMoney(orderMoney);orderMapper.insert(order);// 新增订单事务状态OrderTX tx = new OrderTX();tx.setTxId(RootContext.getXID());tx.setState(OrderTX.STATE_TRY);orderTXMapper.insert(tx);// 调用账户余额扣减String result = accountFeignClient.reduce(userId, orderMoney);if (!"SUCCESS".equals(result)) {throw new RuntimeException("Account服务调用失败");}}@Overridepublic boolean confirm(BusinessActionContext ctx) {System.err.println("---------confirm-----------");// 删除订单事务状态记录int ret = orderTXMapper.delete(new LambdaQueryWrapper<OrderTX>().eq(OrderTX::getTxId, ctx.getXid()));return ret == 1;}@Overridepublic boolean cancel(BusinessActionContext ctx) {System.err.println("---------cancel-----------" );// 新增空回滚判断OrderTX orderTX = orderTXMapper.selectOne(new LambdaQueryWrapper<OrderTX>().eq(OrderTX::getTxId, ctx.getXid()));if (orderTX == null){ // 为空,  空回滚// 新增订单事务状态记录orderTX = new OrderTX();orderTX.setTxId(ctx.getXid());orderTX.setState(OrderTX.STATE_CANCEL);orderTXMapper.insert(orderTX);return true;}// 幂等处理if(orderTX.getState() == OrderTX.STATE_CANCEL){return true;}// 补偿操作: 恢复余额String userId = ctx.getActionContext("userId").toString();String commodityCode = ctx.getActionContext("commodityCode").toString();orderMapper.delete(new LambdaQueryWrapper<Order>().eq(Order::getUserId, userId).eq(Order::getCommodityCode, commodityCode));// 修改订单事务状态orderTX.setState(OrderTX.STATE_CANCEL);int ret = orderTXMapper.updateById(orderTX);return ret == 1;}
}

stock-service

在seata-stock 新增事务状态表

CREATE TABLE `t_order_tx` (`id` bigint NOT NULL AUTO_INCREMENT COMMENT '主键',`tx_id` varchar(100) NOT NULL  COMMENT '事务id',`count` int DEFAULT NULL COMMENT '冻结库存',`state` int DEFAULT NULL COMMENT '状态 0try 1confirm 2cancel',PRIMARY KEY (`id`)
) ENGINE=InNODB DEFAULT CHARSET=utf8;

新增domain: StockTX

import com.baomidou.mybatisplus.anNOtation.IdType;
import com.baomidou.mybatisplus.anNOtation.TableId;
import com.baomidou.mybatisplus.anNOtation.TableName;
import lombok.Data;@Data
@TableName("t_stock_tx")
public class StockTX {public static final int STATE_TRY = 0;public static final int STATE_CONFIRM = 1;public static final int STATE_CANCEL = 2;@TableId(type = IdType.AUTO)private Integer id;private String txId;private int count;private int state = STATE_TRY;
}

新增mapper: StockTXMapper

import com.baomidou.mybatisplus.core.mapper.BaseMapper;public interface StockTXMapper extends BaseMapper<StockTX> {
}

修改: StockTCCServiceImpl

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import io.seata.core.context.RootContext;
import io.seata.rm.tcc.api.BusinessActionContext;
import org.springframework.beans.factory.anNOtation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.anNOtation.Transactional;@Service
public class StockTCCServiceImpl implements IStockTCCService {@Autowiredprivate StockMapper stockMapper;@Autowiredprivate StockTXMapper stockTXMapper;@Overridepublic void tryDeduct(String commodityCode, int count) {System.err.println("---------tryDeduct-----------");// 新增业务悬挂判断StockTX stockTX = stockTXMapper.selectOne(new LambdaQueryWrapper<StockTX>().eq(StockTX::getTxId, RootContext.getXID()));if (stockTX != null){// 存在, 说明已经canel执行过类, 拒绝服务return;}// 校验库存Stock one = stockMapper.selectOne(new LambdaQueryWrapper<Stock>().eq(Stock::getCommodityCode, commodityCode));if(one != null && one.getCount() < count){throw new RuntimeException("not enough Count ...");}// 扣减库存stockMapper.update(null, new LambdaUpdateWrapper<Stock>().setSql("count = count-" + count).eq(Stock::getCommodityCode, commodityCode));// 新增库存事务状态记录StockTX tx = new StockTX();tx.setCount(count);tx.setTxId(RootContext.getXID());tx.setState(StockTX.STATE_TRY);stockTXMapper.insert(tx);}@Overridepublic boolean confirm(BusinessActionContext ctx) {System.err.println("---------confirm-----------");// 删除库存事务状态记录int ret = stockTXMapper.delete(new LambdaQueryWrapper<StockTX>().eq(StockTX::getTxId, ctx.getXid()));return ret == 1;}@Overridepublic boolean cancel(BusinessActionContext ctx) {System.err.println("---------cancel-----------");// 新增空回滚校验StockTX stockTX = stockTXMapper.selectOne(new LambdaQueryWrapper<StockTX>().eq(StockTX::getTxId, ctx.getXid()));if (stockTX == null){ // 为空,  空回滚// 新增库存事务状态记录stockTX = new StockTX();stockTX.setTxId(ctx.getXid());stockTX.setState(StockTX.STATE_CANCEL);if(count != null){stockTX.setCount(Integer.parseInt(count));}stockTXMapper.insert(stockTX);return true;}// 幂等处理if(stockTX.getState() == StockTX.STATE_CANCEL){return true;}// 补偿操作: 恢复余额String count = ctx.getActionContext("count").toString();String commodityCode = ctx.getActionContext("commodityCode").toString();stockMapper.update(null, new LambdaUpdateWrapper<Stock>().setSql("count = count + " + count).eq(Stock::getCommodityCode, commodityCode));// 修改库存事务状态记录stockTX.setCount(0);stockTX.setState(StockTX.STATE_CANCEL);int ret = stockTXMapper.updateById(stockTX);return ret == 1;}
}

测试

正常: http://localhost:8088/businesses/purchase?rollback=false&count=2

超库存: http://localhost:8088/businesses/purchase?rollback=false&count=12

超余额: http://localhost:8088/businesses/purchase?rollback=false&count=8

优缺点

优点

• 一阶段完成直接提交事务, 释放数据库资源, 性能好
• 相比AT模型, 无需生成快照, 无需使用全局锁, 性能最强
• 不依赖数据库事务, 而是依赖补偿操作, 可以用于非事务性数据库

缺点

• 代码侵入, 需要认为编写try, confirm和cancel接口, 麻烦
• 没提交/回滚事务前数据是不一致的, 事务属于最终一致
• 需要考虑confirm和cancel失败情况, 要做好幂等处理

SAGA模式

概述

Saga模式是Seata提供的长事务解决方案, 在Saga模式中, 业务流程中每个参与者都提交本地事务, 当出现某一个参与者失败则补偿前面已经成功的参与者, 一阶段正向服务和二阶段补偿服务都由业务开发实现

简单理解:

saga模式也分为2个阶段

• 一阶段: 直接提交本地事务(所有RM)

• 二阶段: 一阶段成功了, 啥都不做, 如果存在某个RM本地事务失败, 则编写补偿业务(反向操作)来实现回滚

原理

左边是所有参与者事务, 右边是补偿反向操作

正常执行顺序: T1–T2–T3–TN

需要回滚执行顺序: T1–T2–T3–TN—回滚—TN—T3—T2—T1

优缺点

优点

• 事务参与者可以基于事件驱动实现异步调用, 吞吐量高
• 一阶段直接提交事务, 无锁, 性能好
• 不用编写TCC中的三哥阶段, 实现简单

缺点

• 一阶段到二阶段时间不定, 时效性差
• 没有锁, 没有事务隔离, 会有脏写可能

模式选择

	XA	AT	TCC	SAGA
一致性	强一致	弱一致	弱一致	最终一致
隔离性	完全隔离	基于全局锁	基于资源预留隔离	无隔离
代码侵入	无	无	有, 编写3个接口	有, 编写状态机和补偿业务
性能	差	好	非常好	非常好
场景	对一致性、隔离性有高要求的业务	居于关系型数据库的大部分分布式事务场景都可以	对性能要求较高的事务, 有非关系型数据参与的事务	业务流程长, 业务流程多, 参与者包含其他公司或者遗留系统服务, 无法提供TCC模式要求的是3个接口

参考资料

Seata四种事务模式AT、TCC、SAGA、XA详解

Seata官方文档

royi分布式事务

看了 5种分布式事务方案，我司最终选择了 Seata，真香！

浅谈分布式事务及解决方案 | 京东物流技术团队

再有人问你分布式事务，把这篇扔给他

2pc实现ByteTCC框架github地址

拜托，面试请不要再问我TCC分布式事务的实现原理！

TCC实现TCC-TRANSACTION官网

TCC实现hmily框架github地址

分布式事务实践——TCC、Saga入门级理解

基于RocketMQ分布式事务 - 完整示例