分布式事务-使用Seata

传统数据库事务

A-原子性：①事务中的所有操作，要么全部成功，要么全部失败。②影响事务的操作，一般指的是增删改，也就是一个事务中，有多个增删改的SQL

C-一致性：①事务开始前到事务结束后，数据状态需要一致②例如：转账增减金额和支付减去金额+修改订单状态、减库存

I-隔离性：①多个事务之间的操作互相隔离，互不影响②例如：事务隔离级别(读未提交、读已提交，可重复读，串行化)，不同的隔离级别会产生不同的问题(脏读、不可重复读、幻读)

D-持久性：①事务一旦提交，则永久的保存在磁盘，不可逆

事务控制：JDBC：Connection、Mybatis ：SqlSession、Spring：基于运行时异常的声明式事务

在不同的Java技术框架中如何进行事务控制。每种方式都有其自己的特点和适用场景.：

1. **JDBC（Java Database Connectivity）：** 在JDBC中，事务控制通常是通过`Connection`对象来实现的。可以使用`Connection`对象的方法来开始、提交和回滚事务。通常，需要手动编写事务管理代码，包括处理异常和提交或回滚事务。这种方式是相对底层和手动的。

2. **Mybatis：** Mybatis是一个Java持久化框架，它提供了`SqlSession`对象来管理数据库操作。在Mybatis中，可以通过在`SqlSession`上调用`commit()`和`rollback()`方法来执行事务控制。通常，Mybatis可以与Spring集成，以实现更高级的声明式事务管理。

3. **Spring：** Spring框架提供了一种声明式事务管理的方式，它基于运行时异常（通常是`RuntimeException`的子类）来触发事务回滚。您可以使用Spring的`@Transactional`注解来声明事务方法，然后Spring会在方法执行期间自动管理事务的开始、提交和回滚。这种方式相对更高级，减少了手动编写事务管理代码的需求，并提供了更强的可配置性和可扩展性。

为什么使用分布式事务？-在分布式环境下，出现的事务问题：跨服务、多服务多数据库、多服务单数据库

总结：出现了多个数据库连接，(Idea测试时需要进行多数据库连接)。例如，下订单，扣减库存、扣减金额服务是三个服务，订单创建成功——>库存服务发现库存为0，扣减失败(此时其余两个服务应该进行事务回滚)——>订单和扣减金额依然成功，原因：业务失败或抛出异常，事务却没有控制住。此时应该使用一个全局事务来监管分支子事务，其中一个事务失败，全局事务回滚。

分布式理论

一、CAP定理

分布式系统架构的三个指标：Consistency(一致性)、Availability(可用性)、Partition tolerance(分区容错性)

C——用户访问分布式系统中的任意节点，得到的数据必须一致。数据一致性、数据安全可靠

A——用户访问集群中的任意健康节点，必须能得到响应，而不是超时或拒绝。高性能、服务可靠可用性

P——①Partition（分区）：因为网络故障或其它原因导致分布式系统中的部分节点与其它节点失去连接，形成独立分区。

②Tolerance（容错）：在集群出现分区时，整个系统也要持续对外提供服务。容忍出现分区的问题

CAP权衡

在分布式环境下，这三个指标不可能同时做到，且P是一定要保证的，P是分布式的前提、是场景。没有P就没有分布式这一说了，就是单体架构。为什么 P 一定要保证？因为网络是不可靠的。客户因素，没办法避免。CA多个节点要想数据一致，需要时间同步。那么在这个时间内，一定不能对外访问，所以A不成立。如果可以对外访问，那么数据就不一致，所以C就不成立了

CP：在一定时间内，等待集群节点进行数据同步后，对外提供访问

AP：在任何时间内，都对外访问，但是得到的数据可能不一样

二、BASE理论

对CAP的一种补充、放弃强一致性，追求最终一致性

BA、S、E三个理论：

Basically Available（基本可用）

①分布式系统在出现故障时，允许损失部分可用性，即保证核心可用。

②譬如电商大促，为了应对大流量，暂时停止注册服务，这时注册服务就不可用，但是整个系统是可用的，所以叫基本可用

Soft State（软状态）

①在一定时间内，允许出现中间状态，比如临时的不一致状态。

②比如订单状态：待付款、已付款待发货、已发货、已签收、已结束

Eventually Consistent（最终一致性）：

虽然无法保证强一致性，但是在软状态结束后，最终达到数据一致。

分布式事务的解决思路

AP模式：①最终一致性、②各子事务分别执行和提交，允许出现结果不一致，然后采用弥补措施恢复数据即可，实现最终一致。场景高性能，服务高可用

CP模式：①强一致性②各个子事务执行后互相等待，同时提交，同时回滚，达成强一致。但事务等待过程中，处于弱可用状态。场景要求数据安全、强一致性的业务如金融、银行、支付

子事务：分支事务、各个微服务的子事务，这里的事务可以是一个微服务的多个事务。

全局事务：事务协调者，用来监控和通知各个分支事务

使用Seata架构

TC (Transaction Coordinator)

①事务协调者：维护全局和分支事务的状态，协调全局事务提交或回滚。

②监控和通知各个事务，包括分支事务和全局事务

TM (Transaction Manager)

①事务管理器：定义全局事务的范围、开始全局事务、提交或回滚全局事务。

②执行事务

RM (Resource Manager)

资源管理器：管理分支事务处理的资源，与TC交谈以注册分支事务和报告分支事务的状态，并驱动分支事务提交或回滚。

XA模式

X/Open 组织定义的分布式事务处理（DTP，Distributed Transaction Processing）标准，历史悠久，几乎主流关系型数据库都支持。原理都是基于两阶段提交。

一阶段：①事务协调者通知每个事务参与者执行本地事务②本地事务执行完成后报告事务执行状态给事务协调者，此时事务不提交，继续持有数据库锁

二阶段：事务协调者基于一阶段的报告来判断下一步操作：①如果一阶段都成功，则通知所有事务参与者，提交事务②如果一阶段任意一个参与者失败，则通知所有事务参与者回滚事务

seata的XA模型：就是在XA的模型之上，使用TM进行事务管理

优点：①事务的强一致性，满足ACID原则。②常用数据库都支持，实现简单，并且没有代码侵入

缺点：①因为一阶段需要锁定数据库资源，等待二阶段结束才释放，性能较差②依赖关系型数据库实现事务

使用：①配置文件,注意：每个微服务都需要配置

seata:
data-source-proxy-mode: XA

②全局事务开始方法标记注解=>@GlobalTransactional

重点->seata的AT模式

解释：AT模式同样是分阶段提交的事务模型，不过缺弥补了XA模型中资源锁定周期过长的缺陷。

阶段一RM的工作：①- 注册分支事务②- 记录undo-log（数据快照）③- 执行业务sql并提交④- 报告事务状态

阶段二RM的工作：①删除undo-log即可②一阶段都成功，则提交，删除undo-log③一阶段有失败，则回滚，恢复undo-log日志，删除undo-log

优点：①不会死锁②最终一致性

配置方式

cmd——》seata-server-1.4.2测试

①seata-server-1.4.2/conf/registry.conf——>

registry {
  # 注册中心
  type = "nacos"

  nacos {
    application = "seata-server"
    serverAddr = "127.0.0.1:8848"
    group = "SEATA_GROUP"
    namespace = ""
    cluster = "HZ"
    username = "nacos"
    password = "nacos"
  }
  
}

config {
  # 配置中心
  type = "nacos"

  nacos {
    serverAddr = "127.0.0.1:8848"
    namespace = ""
    group = "SEATA_GROUP"
    username = "nacos"
    password = "nacos"
    dataId = "seataServer.properties"
  }
  
}

②nacos中的配置文件 seataServer.properties

# 数据存储方式，db代表数据库
store.mode=db
store.db.datasource=druid
store.db.dbType=mysql
store.db.driverClassName=com.mysql.jdbc.Driver
store.db.url=jdbc:mysql://127.0.0.1:3306/seata?useUnicode=true&rewriteBatchedStatements=true
store.db.user=root
store.db.password=root
store.db.minConn=5
store.db.maxConn=30
store.db.globalTable=global_table
store.db.branchTable=branch_table
store.db.queryLimit=100
store.db.lockTable=lock_table
store.db.maxWait=5000
# 事务、日志等配置
server.recovery.committingRetryPeriod=1000
server.recovery.asynCommittingRetryPeriod=1000
server.recovery.rollbackingRetryPeriod=1000
server.recovery.timeoutRetryPeriod=1000
server.maxCommitRetryTimeout=-1
server.maxRollbackRetryTimeout=-1
server.rollbackRetryTimeoutUnlockEnable=false
server.undo.logSaveDays=7
server.undo.logDeletePeriod=86400000

# 客户端与服务端传输方式
transport.serialization=seata
transport.compressor=none
# 关闭metrics功能，提高性能
metrics.enabled=false
metrics.registryType=compact
metrics.exporterList=prometheus
metrics.exporterPrometheusPort=9898

　　mysql8 要改一下驱动类名跟url加服务时区

③导入 seata-tc-server.sql 创建数据库和3个表

④启动bin/seata-server.bat

服务配置

①导入依赖

<!--seata-->
<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-seata</artifactId>
    <exclusions>
        <!--版本较低，1.3.0，因此排除--> 
        <exclusion>
            <artifactId>seata-spring-boot-starter</artifactId>
            <groupId>io.seata</groupId>
        </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>io.seata</groupId>
    <artifactId>seata-spring-boot-starter</artifactId>
    <!--seata starter 采用1.4.2版本-->
    <version>1.4.2</version>
</dependency>

②配置文件

seata:
  registry: # TC服务注册中心的配置，微服务根据这些信息去注册中心获取tc服务地址
    type: nacos # 注册中心类型 nacos
    nacos:
      server-addr: 127.0.0.1:8848 # nacos地址
      namespace: "" # namespace，默认为空
      group: SEATA_GROUP # 分组，默认是DEFAULT_GROUP
      application: seata-server # seata服务名称
      username: nacos
      password: nacos
  tx-service-group: seata-demo # 事务组名称
  service:
    vgroup-mapping: # 事务组与cluster的映射关系
      seata-demo: HZ

③在分布式事务的入口方法上增加一个注解

@GlobalTransactional

注意：参与分布式事务的服务都需要进行如上的配置

④数据库操作

ock_table导入到TC服务关联的数据库

undo_log表导入到微服务关联的数据库

⑤微服务操作修改application.yml

seata:
data-source-proxy-mode: AT # 默认就是AT

重启微服务

Seata支持的模式

XA、AT、SAGA(适用于长连接)、TCC（Try....Confirm...Cancel）

二阶段提交：

一阶段：预留资源

二阶段：提交或回滚

手写补偿逻辑

消息队列异步确保——>消费方伪代码

public void doMessage(String msg){
    try{
        //setnx();
        JSON.parseObject(msg, 类名.class);
        //扣库存
        //手动ack
    } catch(Exception e){
        //redis中取出来
        //失败的次数一旦超过达到一次的次数限制。（记录日志，人工干预）
        //记录失败的次数，存入redis
        //nack,让消息重回队列
        channel.nack(,,true);
    }
}