Seata--分布式事务

/ 架构设计 / 没有评论 / 1602浏览

分布式事务基础

事务

事务指的就是一个操作单元,在这个操作单元中的所有操作最终要保持一致的行为,要么所有操作都成功,要么所有的操作都被撤销。简单地说,事务提供一种“要么什么都不做,要么做全套”机制。

本地事务

本地事物其实可以认为是数据库提供的事务机制。说到数据库事务就不得不说,数据库事务中的四大特性:

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

分布式事务

分布式事务指事务的参与者、支持事务的服务器、资源服务器以及事务管理器分别位于不同的分布式系统的不同节点之上。

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

本质上来说,分布式事务就是为了保证不同数据库的数据一致性。

分布式事务的场景

单体系统访问多个数据库

一个服务需要调用多个数据库实例完成数据的增删改操作

截屏2021-08-29 下午9.46.17

多个微服务访问同一个数据库

多个服务需要调用一个数据库实例完成数据的增删改操作

截屏2021-08-29 下午9.46.44

多个微服务访问多个数据库

多个服务需要调用一个数据库实例完成数据的增删改操作

截屏2021-08-29 下午9.47.14

分布式事务解决方案

全局事务

全局事务基于DTP模型实现。DTP是由X/Open组织提出的一种分布式事务模型——X/Open Distributed Transaction Processing Reference Model。它规定了要实现分布式事务,需要三种角色:

整个事务分成两个阶段:

滚。

截屏2021-08-29 下午9.48.58

优点

缺点

可靠消息服务

基于可靠消息服务的方案是通过消息中间件保证上、下游应用数据操作的一致性。假设有A和B两个系 统,分别可以处理任务A和任务B。此时存在一个业务流程,需要将任务A和任务B在同一个事务中处 理。就可以使用消息中间件来实现这种分布式事务。

截屏2021-08-29 下午9.50.40

第一步:消息由系统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系统回滚,主要是考虑到整个系统设计的复杂度问题。

基于可靠消息服务的分布式事务,前半部分使用异步,注重性能;后半部分使用同步,注重开发成本。

最大努力通知

最大努力通知也被称为定期校对,其实是对第二种解决方案的进一步优化。它引入了本地消息表来记录错误消息,然后加入失败消息的定期校对功能,来进一步保证消息会被下游系统消费。

截屏2021-08-29 下午9.54.48

第一步:消息由系统A投递到中间件

  1. 处理业务的同一事务中,向本地消息表中写入一条记录

  2. 准备专门的消息发送者不断地发送本地消息表中的消息到消息中间件,如果发送失败则重试

第二步:消息由中间件投递到系统B

  1. 消息中间件收到消息后负责将该消息同步投递给相应的下游系统,并触发下游系统的任务执行

  2. 当下游系统处理成功后,向消息中间件反馈确认应答,消息中间件便可以将该条消息删除,从而该 事务完成

  3. 对于投递失败的消息,利用重试机制进行重试,对于重试失败的,写入错误消息表

  4. 消息中间件需要提供失败消息的查询接口,下游系统会定期查询失败消息,并将其消费

这种方式的优缺点:

优点: 一种非常经典的实现,实现了最终一致性。

缺点: 消息表会耦合到业务系统中,如果没有封装好的解决方案,会有很多杂活需要处理。

TCC事务

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

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

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

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

截屏2021-08-29 下午9.59.42

截屏2021-08-29 下午9.59.51

TCC两阶段提交与XA两阶段提交的区别是:

TCC事务的优缺点:

Seata介绍

2019 年 1 月,阿里巴巴中间件团队发起了开源项目 Fescar(Fast & EaSy Commit AndRollback), 其愿景是让分布式事务的使用像本地事务的使用一样,简单和高效,并逐步解决开发者们遇到的分布式 事务方面的所有难题。后来更名为 Seata,意为:Simple Extensible Autonomous Transaction Architecture,是一套分布式事务解决方案。

Seata的设计目标是对业务无侵入,因此从业务无侵入的2PC方案着手,在传统2PC的基础上演进。它把 一个分布式事务理解成一个包含了若干分支事务的全局事务。全局事务的职责是协调其下管辖的分支事 务达成一致,要么一起成功提交,要么一起失败回滚。此外,通常分支事务本身就是一个关系数据库的 本地事务。

截屏2021-08-29 下午10.05.03

Seata主要由三个重要组件组成:

截屏2021-08-29 下午10.06.00

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协调其管辖之下的所有分支事务, 决定是否回滚

Seata实现2PC与传统2PC的差别:

  1. 架构层次方面,传统2PC方案的 RM 实际上是在数据库层,RM本质上就是数据库自身,通过XA协 议实现,而 Seata的RM是以jar包的形式作为中间件层部署在应用程序这一侧的。
  2. 两阶段提交方面,传统2PC无论第二阶段的决议是commit还是rollback,事务性资源的锁都要保 持到Phase2完成才释放。而Seata的做法是在Phase1 就将本地事务提交,这样就可以省去Phase2 持锁的时间,整体提高效率

Seata实现分布式事务控制

本示例通过Seata中间件实现分布式事务,模拟电商中的下单和扣库存的过程

我们通过订单微服务执行下单操作,然后由订单微服务调用商品微服务扣除库存

截屏2021-08-29 下午10.08.39

案例基本代码

修改order微服务

controller

@RestController
@Slf4j
public class OrderController5 {
  @Autowired
  private OrderServiceImpl5 orderService;
  //下单
  @RequestMapping("/order/prod/{pid}")
  public Order order(@PathVariable("pid") Integer pid) {
    log.info("接收到{}号商品的下单请求,接下来调用商品微服务查询此商品信息", pid);
    return orderService.createOrder(pid); 
  }
}

OrderService

@Service
@Slf4j
public class OrderServiceImpl5{

    @Autowired
    private OrderDao orderDao;
    @Autowired
    private ProductService productService;
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @GlobalTransactional
    public Order createOrder(Integer pid) {
            //1 调用商品微服务,查询商品信息
            Product product = productService.findByPid(pid); 
            log.info("查询到{}号商品的信息,内容是:{}", pid, JSON.toJSONString(product)); 
            //2 下单(创建订单)
            Order order = new Order();
            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);
            return order;
       }
}
 

ProductService

@FeignClient(value = "service-product") 
public interface ProductService {
  //减库存 
  @RequestMapping("/product/reduceInventory")
  void reduceInventory(@RequestParam("pid") Integer pid,@RequestParam("num")
}

修改Product微服务

controller

//减少库存 
@RequestMapping("/product/reduceInventory")
public void reduceInventory(Integer pid, int num) {
  productService.reduceInventory(pid, num); 
}

service

 
@Override
public void reduceInventory(Integer pid, int num) { 
  Product product = productDao.findById(pid).get(); 
  product.setStock(product.getStock() - num); 
  //减库存
  productDao.save(product); 
}

异常模拟

在ProductServiceImpl的代码中模拟一个异常, 然后调用下单接口

@Override
public void reduceInventory(Integer pid, Integer number) { 
  Product product = productDao.findById(pid).get();
  if (product.getStock() < number) {
  		throw new RuntimeException("库存不足"); 
  }
  int i = 1 / 0; 
  product.setStock(product.getStock() - number); 
  productDao.save(product);
}

启动Seata

下载seata

下载地址:https://github.com/seata/seata/releases/v0.9.0/

修改配置文件

将下载得到的压缩包进行解压,进入conf目录,调整下面的配置文件:

registry.conf

registry {
  # file 、nacos 、eureka、redis、zk、consul、etcd3、sofa
  type = "nacos"

  nacos {
    serverAddr = "localhost"
    namespace = ""
    cluster = "default"
  }
}
config {
  # file、nacos 、apollo、zk、consul、etcd3
  type = "nacos"
  nacos {
    serverAddr = "localhost"
    namespace = ""
  }
}

nacos-confifig.txt

 service.vgroup_mapping.service-product=default 
 service.vgroup_mapping.service-order=default

初始化seata在nacos的配置

# 初始化seata 的nacos配置
# 注意: 这里要保证nacos是已经正常运行的 
cd conf
nacos-config.sh 127.0.0.1

执行成功后可以打开Nacos的控制台,在配置列表中,可以看到初始化了很多Group为SEATA_GROUP 的配置。

启动seata服务

cd bin
seata-server.bat -p 9000 -m file

启动后在 Nacos 的服务列表下面可以看到一个名为 serverAddr 的服务。

使用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;

添加配置

在需要进行分布式控制的微服务中进行下面几项配置:

添加依赖

 
<dependency>
  <groupId>com.alibaba.cloud</groupId> 
  <artifactId>spring-cloud-starter-alibaba-seata</artifactId> 
</dependency>
<dependency>
  <groupId>com.alibaba.cloud</groupId> 
  <artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId> 
</dependency>

DataSourceProxyConfifig

Seata 是通过代理数据源实现事务分支的,所以需要配置 io.seata.rm.datasource.DataSourceProxy 的 Bean,且是 @Primary默认的数据源,否则事务不会回滚,无法实现分布式事务

@Configuration
public class DataSourceProxyConfig {
    @Bean
    @ConfigurationProperties(prefix = "spring.datasource") public DruidDataSource druidDataSource() {
        return new DruidDataSource();
    }
    @Primary
    @Bean
    public DataSourceProxy dataSource(DruidDataSource druidDataSource) {
        return new DataSourceProxy(druidDataSource);
    }
}

registry.conf

在resources下添加Seata的配置文件 registry.conf

registry {
    type = "nacos"
    nacos {
      serverAddr = "localhost"
      namespace = "public"
      cluster = "default"
    }
}
config {
    type = "nacos"
    nacos {
      serverAddr = "localhost"
      namespace = "public"
      cluster = "default"
    }
}

bootstrap.yaml

spring:
 application:
  name: service-product
 cloud:
  nacos:
   config:
  	 server-addr: localhost:8848 # nacos的服务端地址 
  	 namespace: public
      group: SEATA_GROUP
  alibaba:
   seata:
    tx-service-group: ${spring.application.name }

在order微服务开启全局事务

@GlobalTransactional//全局事务控制
public Order createOrder(Integer pid) {}

测试

再次下单测试

seata运行流程分析

截屏2021-08-29 下午10.32.57

要点说明:

  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 并执行,以完成分支事务回滚到之前的状态,如果 回滚失败则会重试回滚操作