分布式事务(seata AT)

seata是一款开源的分布式事务解决方案,致力于提供高性能和简单易用的分布式务服务.

seate将为用户提供了AT,TCC,SAGA和XA 事务模式.

Seata AT事务方案

seata 的AT模式是一种无侵入的分布式事务解决方案

当用户下订单时，执行以下三步流程：

订单系统保存订单
订单系统调用库存服务，减少商品库存
订单系统调用账户服务，扣减用户金额
这三步要作为一个整体事务进行管理，要么整体成功，要么整体失败。

Seata AT基本原理
Seata AT 事务分两个阶段来管理全局事务：
第一阶段：执行各分支事务
第二阶段：控制全局事务最终提交或回滚

seata AT是通过通过事务协调:

1,执行分布式事务

2.控制全局事务,最终提交或回滚

3.对每个全局事务分配一个事务id

4.RM(资源)上报状态,上报给TC

5.TC要收集每个模块的状态

6.访问库存RM 要先向TC注册

总结:确定有的订单模块都执行成功了,TC才会执行第二阶段回滚或提交,或执行失败了TC会向所有RM发送回滚操作指令,RM会完成最终回滚操作

TC通对数据库的undo_log事务日志表,里面有操作前,操作后的两个数据在这里面,若分布事务下单个模块有一个执行失败,那么TC会通过undo_log事务日志表,回滚到操作前的数据,然后删除事务日志表,这步也叫第二阶段提交.

TC事务协调器,确定所有订单模块都执行成功了,TC才会执行第二阶段回滚或提交.若失败了TC会完成最终回滚操作

seata AT 事务

1.启动TC事务协调器

2.修改三个配置文件

registrg.conf /bin目录下

file.conf /bin 目录下这个是seata server运行过程中产生的日志数据存储到哪儿

seata-server.bat 这个如查是苹果电脑就用seata-server.bat

在seata文件夹里的bin文件里输入:cmd 打开小黑窗口

运行必:

设置JAVA_HOME PATH

JDK必须是1.8

在eureka网页里有seata-server就表示成功了

然后就是在idae中的哪个类先执行,就在哪个类里的业务方法添加事务注解:@Transactional //控制本地事务.同时加上GlobalTransactional启动全局事务.比如我们现在有一个模块,一个订单,一个用户,一个扣减金额.那个肯定是有了订单,订单分别调动用户和账户所以在订单模块里加

file.conf文件

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

  nacos {
    serverAddr = "localhost"
    namespace = ""
    cluster = "default"
  }
  eureka {
    serviceUrl = "http://localhost:8761/eureka"  ######################
    # application = "default"
    # weight = "1"
  }
  redis {
    serverAddr = "localhost:6379"
    db = "0"
    password = ""
    cluster = "default"
    timeout = "0"
  }
  zk {
    cluster = "default"
    serverAddr = "127.0.0.1:2181"
    session.timeout = 6000
    connect.timeout = 2000
    username = ""
    password = ""
  }
  consul {
    cluster = "default"
    serverAddr = "127.0.0.1:8500"
  }
  etcd3 {
    cluster = "default"
    serverAddr = "http://localhost:2379"
  }
  sofa {
    serverAddr = "127.0.0.1:9603"
    application = "default"
    region = "DEFAULT_ZONE"
    datacenter = "DefaultDataCenter"
    cluster = "default"
    group = "SEATA_GROUP"
    addressWaitTime = "3000"
  }
  file {
    name = "file.conf"
  }
}

config {
  # file、nacos 、apollo、zk、consul、etcd3、springCloudConfig
  type = "file"

  nacos {
    serverAddr = "localhost"
    namespace = ""
    group = "SEATA_GROUP"
  }
  consul {
    serverAddr = "127.0.0.1:8500"
  }
  apollo {
    app.id = "seata-server"
    apollo.meta = "http://192.168.1.204:8801"
    namespace = "application"
  }
  zk {
    serverAddr = "127.0.0.1:2181"
    session.timeout = 6000
    connect.timeout = 2000
    username = ""
    password = ""
  }
  etcd3 {
    serverAddr = "http://localhost:2379"
  }
  file {
    name = "file.conf"
  }
}

registry.conf

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

  nacos {
    serverAddr = "localhost"
    namespace = ""
    cluster = "default"
  }
  eureka {
    serviceUrl = "http://localhost:8761/eureka"  ######################
    # application = "default"
    # weight = "1"
  }
  redis {
    serverAddr = "localhost:6379"
    db = "0"
    password = ""
    cluster = "default"
    timeout = "0"
  }
  zk {
    cluster = "default"
    serverAddr = "127.0.0.1:2181"
    session.timeout = 6000
    connect.timeout = 2000
    username = ""
    password = ""
  }
  consul {
    cluster = "default"
    serverAddr = "127.0.0.1:8500"
  }
  etcd3 {
    cluster = "default"
    serverAddr = "http://localhost:2379"
  }
  sofa {
    serverAddr = "127.0.0.1:9603"
    application = "default"
    region = "DEFAULT_ZONE"
    datacenter = "DefaultDataCenter"
    cluster = "default"
    group = "SEATA_GROUP"
    addressWaitTime = "3000"
  }
  file {
    name = "file.conf"
  }
}

config {
  # file、nacos 、apollo、zk、consul、etcd3、springCloudConfig
  type = "file"

  nacos {
    serverAddr = "localhost"
    namespace = ""
    group = "SEATA_GROUP"
  }
  consul {
    serverAddr = "127.0.0.1:8500"
  }
  apollo {
    app.id = "seata-server"
    apollo.meta = "http://192.168.1.204:8801"
    namespace = "application"
  }
  zk {
    serverAddr = "127.0.0.1:2181"
    session.timeout = 6000
    connect.timeout = 2000
    username = ""
    password = ""
  }
  etcd3 {
    serverAddr = "http://localhost:2379"
  }
  file {
    name = "file.conf"
  }
}

order下的application.yml文件

spring:
  application:
    name: order

  datasource:
    driver-class-name: com.mysql.cj.jdbc.Driver
    url: jdbc:mysql:///seata_order?useUnicode=true&characterEncoding=UTF-8&serverTimezone=GMT%2B8
    username: root
    password: root
    jdbcUrl: ${spring.datasource.url}
  cloud:
    alibaba:
      seata:
        tx-service-group: order_tx_group

server:
  port: 8083

eureka:
  client:
    service-url:
      defaultZone: http://localhost:8761/eureka
  instance:
    prefer-ip-address: true
    instance-id: ${spring.cloud.client.ip-address}:${spring.application.name}:${server.port}

mybatis-plus:
  type-aliases-package: cn.tedu.order.entity
  mapper-locations:
    - classpath:/mapper/*Mapper.xml
  configuration:
    map-underscore-to-camel-case: true

logging:
  level:
    cn.tedu.order.mapper: debug
ribbon:
  MaxAutoRetriesNextServer: 0

bootstrap.yml

获取自己电脑上ip的方式

在用户account和金额storage里分别也加上这三个东西

file.conf registry.conf bootstrap.yml 这三个文件 ,只用复制就行了,DSAutoConf类也复制

其它与启动类同包下的

DSAutoConf类

package cn.tedu.storage;

import com.zaxxer.hikari.HikariDataSource;
import io.seata.rm.datasource.DataSourceProxy;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;

import javax.sql.DataSource;

@Configuration
public class DSAutoConf {
    //创建原始数据源
    @Bean
    @ConfigurationProperties(prefix = "spring.datasource")
    public DataSource getDataSource(){
        return new HikariDataSource();
    }
    //创建代理对象
    @Bean
    @Primary //首选对象
    public DataSource dataSourceProxy(DataSource ds){
        return new DataSourceProxy(ds);
    }
}

分别在account和storage的实现类里的方法上加上

@Transactional注解

按顺序启动服务：

Eureka
Seata Server
Easy Id Generator
Order

调用保存订单，地址：
http://localhost:8083/create?userId=1&productId=1&count=10&money=100

观察控制台，看到全局事务和订单的分支事务已经启动，并可以看到全局事务ID（XID）和分支事务ID（Branch ID）：

然后观察数据库中新添加的订单数据：

测试出现异常，回滚的情况

在业务代码中加一个模拟异常再试一下：

package cn.tedu.order.service;

import cn.tedu.order.entity.Order;
import cn.tedu.order.feign.AccountClient;
import cn.tedu.order.feign.EasyIdClient;
import cn.tedu.order.feign.StorageClient;
import cn.tedu.order.mapper.OrderMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Random;

@Service
public class OrderServiceImpl implements OrderService{



    @Autowired
    private OrderMapper orderMapper;
    @Autowired
    private EasyIdClient easyIdClient;
    @Autowired
    private AccountClient accountClient;
    @Autowired
    private StorageClient storageClient;

    @Override
    public void create(Order order) {
        //远程调用id发号器,生 订单ID
        String s = easyIdClient.getId("order_business");
        Long orderId = Long.valueOf(s);

        order.setId(orderId);
        orderMapper.create(order);
        //TODO: 远程调用库存,减少库存
        storageClient.decrease(order.getProductId(),order.getCount());
        //TODO: 远程调用用户,扣减金额
        accountClient.decrease(order.getUserId(),order.getMoney());
    }
}

重启 order 项目，并调用保存订单：
http://localhost:8083/create?userId=1&productId=1&count=10&money=100

订单启动全局事务部分完成，在继续之前，先把模拟异常注释掉：

配置

与订单项目中添加的配置完全相同，请参考订单配置章节配置下面三个文件：

application.yml
registry.conf
file.conf

创建 seata 数据源代理
与订单项目中数据源代理完全相同，请参考订单中数据源代理章节，在 cn.tedu.storage 包下创建数据源配置类 DatasourceConfiguration。主程序注解排除 DataSourceAutoConfiguration 自动配置类。

测试出现异常，回滚的情况

在业务代码中加一个模拟异常再试一下：

package cn.tedu.storage.service;
import cn.tedu.storage.mapper.StorageMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Service
public class StorageServiceImpl implements StorageService {
    @Autowired
    private StorageMapper storageMapper;
    @Transactional
    @Override
    public void decrease(Long productId, Integer count) {
        storageMapper.decrease(productId, count);
    }
}

重启 storage 项目，并调用保存订单：
http://localhost:8083/create?userId=1&productId=1&count=10&money=100

查看数据库表 order 和 storage，如果执行成功会新增订单、减少库存，如果执行失败则数据没有变化，被回滚了。

storage 分支事务部分完成，在继续之前，先把模拟异常注释掉：

account账户服务添加 Seata AT 事务

配置

与订单项目中添加的配置完全相同，请参考订单配置章节配置下面三个文件：

application.yml
registry.conf
file.conf
创建 seata 数据源代理
与订单项目中数据源代理完全相同，请参考订单中数据源代理章节，在 cn.tedu.account 包下创建数据源配置类 DatasourceConfiguration。主程序注解排除 DataSourceAutoConfiguration 自动配置类。.............................................................................................................................................................................................................................................................................................................................与storage里设置一样
重启 account 项目，并调用保存订单：
http://localhost:8083/create?userId=1&productId=1&count=10&money=100

失败时，在 order 和 storage 控制台可以看到回滚日志。

这是 storage 的回滚日志：

项目地址:: 仓库 - 微儿VR (weier-vr) - Gitee.com

分布式事务(seata AT)

大数据系统相关栏目本月热门文章