分布式事务

Java_慈祥

发布于 2024-08-06 15:14:14

800

发布于 2024-08-06 15:14:14

文章被收录于专栏：Web前后端、全栈出发

分布式事务

前言：

本篇学习借鉴了以下文章： 感谢大佬 https://blog.csdn.net/bjweimengshu/article/details/79607522 https://www.cnblogs.com/savorboard/p/distributed-system-transaction-consistency.html https://www.cnblogs.com/monkeyblog/p/10449363.html 当然这里绝对不是打广告！ 而且，我看的可不止这么多，这个是我看的比较好的… 为了方便我后面自己忘记了可以回顾的。
本篇，存在一些大佬的文章搬运，主要是写的太好了，copy来了… 如果大佬不允许会及时删除！

什么是事务

举个生活中的例子：

你去小卖铺买东西，“一手交钱，一手交货”就是一个事务的例子
交钱和交货 必须全部成功，事务才算成功任一个活动失败，事务将撤销所有已成功的活动。
事务可以看做是一次大的活动，它由不同的小活动组成，这些活动要么全部成功，要么全部失败。

数据库事务的四大特性 ACID：

A（Atomic）：原子性

构成事务的所有操作，要么都执行完成，要么全部不执行，不可能出现部分成功部分失败的情况。

C（Consistency）：一致性

在事务执行前后，数据库的一致性约束没有被破坏。
比如张三100元，李四100元，一共200。李四给张三50 李四50元，张三150元，一共还是200元！

I（Isolation）：隔离性

数据库中的事务一般都是并发的，
隔离性是指并发的两个事务的执行互不干扰，一个事务不能看到其他事务运行过程的中间状态。
通过配置事务隔离级别可以避脏读、重复读等问题

D（Durability）：持久性

事务完成之后，该事务对数据的更改会被持久化到数据库，且不会被回滚

本地事务 `Local Transaction`

起初，事务仅限于对单一数据库资源的访问控制 架构服务化以后，事务的概念延伸到了服务中。
倘若将一个单一的服务操作作为一个事务，那么整个服务操作只能涉及一个单一的数据库资源。
这类基于单个服务单一数据库资源访问的事务，被称为**本地事务**

分布式事务 | 产生的场景

随着互联网的快速发展，软件系统由原来的 单体应用 转变为**分布式应用**
分布式系统会把一个应用系统拆分为可独立部署的多个服务，不同的服务还会有不同的库 因此需要服务与服务之间远程协作才能完成事务操作
这种分布式系统环境下**由不同的服务之间通过网络远程协作，在不同的数据库之间，完成事务**称之为分布式事务

单一服务分布式事务

最早的分布式事务应用架构很简单
不涉及服务间的访问调用，仅仅是服务内操作涉及到对多个数据库资源的访问。

多服务分布式事务

一个服务操作访问不同的数据库资源 对于上面介绍的分布式事务应用架构，尽管一个服务操作会访问多个数据库资源，但是毕竟整个事务还是控制在单一服务的内部。
一个服务操作需要调用另外一个服务，**这时的事务就需要跨越多个服务了**

多服务多数据源分布式事务

在多个服务之间，且不同服务存在不同的数据库，的环境下的分布式事务 好牛啊！

事务的作用：

保证每个事务的数据一致性。

分布式事务基础理论

CAP理论

理解CAP

CAP是 Consistency、Availability、Partition tolerance三个词语的缩写分别表示：一致性、可用性、分区容忍性

C (一致性)

一致性是指: 写操作后的读操作可以**读取到最新的数据状态** 实时更新！
对于数据分布在不同节点上的数据来说如果在某个节点更新了数据，那么在其他节点如果都能读取到这个最新的数据，那么就称为 强一致 如果有某个节点没有读取到，那就是 分布式不一致

A (可用性)

可用性，就是提高程序的高可用… 提高安全… 类似于搭建集群，一个服务挂了，其它服务继续工作不影响使用！，但这必然影响了C一致性的性能速度！
但注意，可用行会导致一致性的效率，但也要保证 最终一致性 这是事务必须的！ 非故障的节点在合理的时间内返回合理的响应　(不是错误和超时的响应)
所以分布式系统理论上不可能选择 CA 架构, 只能选择 CP 或者 AP 架构。

P (分区容错性)

分布式系统的各各结点部署在不同的子网，这就是**网络分区 不可避免的会出现由于网络问题而导致结点之间通信失败，此时仍可对外提供服务，这叫**分区容忍性
如何实现分区容忍性？ 尽量使用异步取代同步操作 　例如使用异步方式将数据从主数据库同步到从数据，这样结点之间能有效的实现松耦合。 添加从数据库结点，其中一个从结点挂掉其它从结点提供服务。
分布式分区容忍性的特点： 分区容忍性分是布式系统具备的基本能力。

CAP有哪些组合方式呢？

AP：

放弃一致性，追求分区容忍性和可用性。
例如： 商品管理，完全可以实现AP，前提是只要用户可以接受所查询的到数据在一定时间内不是最新的即可。 一些业务场景比如： 订单退款，今日退款成功，明日账户到账，只要用户可以接受在一定时间内到账即可。

CP:

放弃可用性，追求一致性和分区容错性，我们的zookeeper其实就是追求的强一致
一些业务场景比如： 抢购商品秒杀！

CA：

放弃分区容忍性，即不进行分区，不考虑由于网络不通或结点挂掉的问题，则可以实现一致性和可用性。
那么系统将不是一个标准的分布式系统，我们最常用的 关系型数据就满足了CA

面试题：ZooKeeper 和 eureka 的区别

ZooKeeper 是 cp eureka 是ap 的架构….

总结

CAP理论告诉我们一个分布式系统最多只能同时满足一致性（Consistency）、可用性（Availability）和分区容忍性（Partition tolerance）这三项中的两项.
在实际生产中很多场景都要实现一致性 比如我们举的例子：主数据库向从数据库同步数据，即使不要一致性，但是最终也要将数据同步成功来保证数据一致
其中AP在实际应用中较多，AP即舍弃一致性，保证可用性和分区容忍性

BASE理论

BASE 是 Basically Available(基本可用)、Soft state(软状态)和 Eventually consistent (最终一致性)三个短语的缩写。
BASE理论是对CAP中AP的一个扩展，通过牺牲强一致性来获得可用性 当出现故障允许部分不可用但要保证核心功能可用，允许数据在一段时间内是不一致的，但最终达到一致状态。
满足BASE理论的事务，我们称之为 “柔性事务”

基本可用:

分布式系统在出现故障时，允许损失部分可用功能，保证核心功能可用。
如，电商网站交易付款出现问题了，商品依然可以正常浏览。

软状态:

由于不要求强一致性，所以BASE允许系统中存在中间状态（也叫软状态）
这个状态不影响系统可用性，如订单的"支付中"、“数据同步中”等状态，待数据最终一致后状态改为“成功”状态。

最终一致:

最终一致是指经过一段时间后，所有节点数据都将会达到一致。 如订单的"支付中"状态，最终会变为“支付成功”或者"支付失败"，使订单状态与实际交易结果达成一致，但需要一定时间的延迟、等待。

分布式事务解决方案

XA分布式事务协议

分布式事务常见的解决方案有：**2pc传统方案** 2PC的传统方案是在数据库层面实现的，如Oracle、MySQL都支持2PC协议
为了统一标准减少行业内不必要的对接成本，需要制定标准化的处理模型及接口标准
国际开放标准组织**Open Group 定义了 分布式事务处理模型 DTP** Distributed Transaction Processing Reference Model

DTP模型

DTP模型定义如下角色：

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

基于XA协议的两阶段提交(2PC)

两阶段提交协议 Two Phase Commitment Protocol 涉及到两种角色

一个**事务协调者（coordinator）** TM事务管理器负责协调多个参与者进行事务投票及提交(回滚)
多个**事务参与者（participants）** RM资源管理器即本地事务执行者

总共处理步骤有两个

投票阶段**（voting phase） 参与者操作协调者将通知，事务参与者准备提交或取消事务，然后进入表决过程投票阶段参与者将告知协调者自己的决策** true： 事务参与者，本地事务执行成功,但未提交 flase： 本地事务执行故障
提交阶段**（commit phase） 协调者操作收到参与者的通知后，协调者再向参与者发出通知，根据反馈**投票**情况决定，各参与者是否要提交还是回滚 多个参与者，只要有一个false , 就表示事务执行失败，通知所有的参与者未提交的事务进行回滚！**

2PC总结：

整个2PC的事务流程涉及到三个角色AP、RM、TM。
AP指的是使用2PC分布式事务的应用程序；
RM指的是资源管理器，它控制着分支事务；
TM指的是事务管理器，它控制着整个全局事务。

缺点：

在 投票阶段**，RM执行实际的业务操作，但不提交事务，**资源锁定
协调者单点故障问题 3PC阶段解此问题！ 事务协调者是整个XA模型的核心，一旦事务协调者节点挂掉，参与者收不到提交或是回滚通知。参与者会一直处于中间状态无法完成事务。
丢失消息导致的不一致问题。 在XA协议的第二个阶段，如果发生局部网络问题，一部分事务参与者收到了提交消息 另一部分事务参与者没收到提交消息，那么就导致了节点之间数据的不一致…!!

Seata 实现2PC

Seata方案

Seata是由阿里中间件团队发起的开源项目 Fescar， 后更名为Seata Simple Extensible Autonomous Transaction Architecture 一套一站式分布式事务解决方案。

解决分布式事务问题，有两个设计初衷

对业务无侵入 即减少技术架构上的微服务化所带来的分布式事务问题对业务的侵入，实际开发中只要一个注解就搞定了！**@Configuration**
高性能 减少分布式事务解决方案所带来的性能消耗，添加 undo_log 表，**本地放心提交事务，可以依靠undo_log进行回滚处理..**

seata中有两种分布式事务实现方案：**AT** 及**TCC**

本人目前只会AT的…😢 Seata AT模式是基于 XA事务演进而来的一个分布式事务中间件

Seata的设计思想：

Seata的设计目标其一是对业务无侵入，因此从业务无侵入的2PC方案着手 在传统2PC的基础上演进，并解决 2PC方案面临的问题, 第二阶段资源占用！

与传统2PC 的模型类似，Seata定义了3个组件来协议分布式事务的处理过程：

Transaction Manager (TM )：事务管理器 控制全局事务的边界，负责开启一个全局事务，并最终向TC发起全局提交或全局回滚的**决议。**
Transaction Coordinator (TC)：事务协调器 事务协调器，维护全局事务的运行状态，负责**协调并驱动**全局事务的提交或回滚。
Resource Manager (RM)：控制分支事务 控制分支事务，负责分支注册、状态汇报（投票），并接收事务协调器的指令，驱动分支（本地）事务的提交和回滚。

具体执行流程：

第一阶段

在一阶段，Seata 会拦截“业务 SQL”
首先解析 SQL 语义，找到“业务 SQL”要更新的业务数据，在业务数据被更新前，将其保存成“before image”快照 存在undo_log表中！
然后执行“业务 SQL”更新业务数据，在业务数据更新之后，再将其保存成“after image”，最后生成行锁。 数据库行锁：确保多线程情况下，改记录只能由一个线程操作！
以上操作全部在一个数据库事务内完成，这样保证了一阶段操作的原子性。
TM 向 TC 申请开启一个全局事务，全局事务创建成功并生成一个全局唯一的XID 确保后面执行...

第二阶段

提交

第二阶段如果是提交的话，因为业务SQL在一阶段已经提交至数据库（已通过）
所以 Seata 框架只需将一阶段保存的快照数据和行锁删掉，完成数据清理即可。 极大提高了第二阶段的执行性能！

回滚

第二阶段如果是回滚的话 Seata就需要回滚一阶段已执行的的业务SQL。当然回滚方式是使用before image镜像还原业务数据。
RM 收到协调器发来的回滚请求，通过 XID线程id 和 Branch ID 分支id 找到相应的回滚日志记录
在还原之前还要进行校验脏写 判断当前的数据，和undo表中，执行后的数据是否一致. 事务执行后的数据是否被更改过!

如果两份数据完全一致就说明没有脏写，可以还原业务数据

如果不一致就说明有脏写，出现脏写就需要转人工处理。

Seata　2PC与传统2PC的差别

架构层次方面

传统2PC方案的 RM 实际上是在数据库层，RM 本质上就是数据库自身 通过 XA 协议实现
而Seata的 RM 是以jar包的形式作为中间件层部署在应用程序这一侧的。

两阶段提交方面

传统2PC无论第二阶段的决议是commit还是rollback 事务性资源的锁都要保持到 第二阶段完成才释放。
而Seata的做法是在阶段一就将本地事务提交将提交前的数据信息，保存在undo_log表中...，这样就可以省去阶段二持锁的时间，整体提高效率。

Spring Cloud 快速集成 Seata

上面理论，了解即可…具体的本人还不是很清除后面可能会整理学习…**目前会用即可！**
Seata使用起来还是非常简单的！
Github 集成文档

依赖：

一般给要管理的微服加入即可，但因为几乎所有的微服都需要，所有就放在公共的**util微服里了！**

	<dependency>
         <groupId>com.alibaba.cloud</groupId>
         <artifactId>spring-cloud-alibaba-seata</artifactId>
         <version>2.1.0.RELEASE</version>
     </dependency>

或

<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-alibaba-dependencies</artifactId>
            <version>2.1.0.RELEASE</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

因为：

spring-cloud-starter-alibaba-seata 这个依赖中只依赖了spring-cloud-alibaba-seata

所以在项目中添加spring-cloud-starter-alibaba-seata和spring-cloud-alibaba-seata是一样的

添加配置文件

同上一般都放在公共的 util微服模块的, resourece资源文件目录下：

registry.conf

该配置用于指定 TC 的注册中心和配置文件，默认都是 file;
如果使用其他的注册中心，要求 Seata-Server 也注册到该配置中心上

registry {
  # file 、nacos 、eureka、redis、zk、consul、etcd3、sofa
  type = "file"			#默认 file文件类型

  nacos {
    serverAddr = "localhost"
    namespace = "public"
    cluster = "default"
  }
  eureka {
    serviceUrl = "http://localhost:8761/eureka"
    application = "default"
    weight = "1"
  }
  redis {
    serverAddr = "localhost:6379"
    db = "0"
  }
  zk {
    cluster = "default"
    serverAddr = "127.0.0.1:2181"
    session.timeout = 6000
    connect.timeout = 2000
  }
  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
  type = "file"

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

file.conf

该配置用于指定TC的相关属性；如果使用注册中心也可以将配置添加到配置中心

transport {
  # tcp udt unix-domain-socket
  type = "TCP"
  #NIO NATIVE
  server = "NIO"
  #enable heartbeat
  heartbeat = true
  #thread factory for netty
  thread-factory {
    boss-thread-prefix = "NettyBoss"
    worker-thread-prefix = "NettyServerNIOWorker"
    server-executor-thread-prefix = "NettyServerBizHandler"
    share-boss-worker = false
    client-selector-thread-prefix = "NettyClientSelector"
    client-selector-thread-size = 1
    client-worker-thread-prefix = "NettyClientWorkerThread"
    # netty boss thread size,will not be used for UDT
    boss-thread-size = 1
    #auto default pin or 8
    worker-thread-size = 8
  }
  shutdown {
    # when destroy server, wait seconds
    wait = 3
  }
  serialization = "seata"
  compressor = "none"
}
service {
  #vgroup->rgroup
  vgroup_mapping.my_test_tx_group = "default"
  #only support single node
  default.grouplist = "127.0.0.1:8091"
  #degrade current not support
  enableDegrade = false
  #disable
  disable = false
  #unit ms,s,m,h,d represents milliseconds, seconds, minutes, hours, days, default permanent
  max.commit.retry.timeout = "-1"
  max.rollback.retry.timeout = "-1"
}

client {
  async.commit.buffer.limit = 10000
  lock {
    retry.internal = 10
    retry.times = 30
  }
  report.retry.count = 5
}

## transaction log store
store {
  ## store mode: file、db
  mode = "file"

  ## file store
  file {
    dir = "sessionStore"

    # branch session size , if exceeded first try compress lockkey, still exceeded throws exceptions
    max-branch-session-size = 16384
    # globe session size , if exceeded throws exceptions
    max-global-session-size = 512
    # file buffer size , if exceeded allocate new buffer
    file-write-buffer-cache-size = 16384
    # when recover batch read size
    session.reload.read_size = 100
    # async, sync
    flush-disk-mode = async
  }

  ## database store
  db {
    ## the implement of javax.sql.DataSource, such as DruidDataSource(druid)/BasicDataSource(dbcp) etc.
    datasource = "dbcp"
    ## mysql/oracle/h2/oceanbase etc.
    db-type = "mysql"
    url = "jdbc:mysql://127.0.0.1:3306/seata"
    user = "mysql"
    password = "mysql"
    min-conn = 1
    max-conn = 3
    global.table = "global_table"
    branch.table = "branch_table"
    lock-table = "lock_table"
    query-limit = 100
  }
}
lock {
  ## the lock store mode: local、remote
  mode = "remote"

  local {
    ## store locks in user's database
  }

  remote {
    ## store locks in the seata's server
  }
}
recovery {
  committing-retry-delay = 30
  asyn-committing-retry-delay = 30
  rollbacking-retry-delay = 30
  timeout-retry-delay = 30
}

transaction {
  undo.data.validation = true
  undo.log.serialization = "jackson"
}

## metrics settings
metrics {
  enabled = false
  registry-type = "compact"
  # multi exporters use comma divided
  exporter-list = "prometheus"
  exporter-prometheus-port = 9898
}

修改应用程序yml

在各个需要分布式事务的模块添加yml，并且指定file.conf中配置通信指定组名my_test_tx_group

.yml

spring:
  cloud:
    alibaba:
      seata:
        tx-service-group: my_test_tx_group

注入数据源

同上，为了方便操作 一般直接放在公共的util 模块中
Seata 通过代理数据源的方式实现分支事务，MyBatis和JPA都需要注入 io.seata.rm.datasource.DataSourceProxy
不同的是MyBatis 还需要额外注入 org.apache.ibatis.session.SqlSessionFactory

DataSourceProxyConfig.Java

import com.alibaba.druid.pool.DruidDataSource;
import io.seata.rm.datasource.DataSourceProxy;
import org.apache.ibatis.session.SqlSessionFactory;
import org.mybatis.spring.SqlSessionFactoryBean;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import javax.sql.DataSource;

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

    @Bean
    public DataSourceProxy dataSourceProxy(DataSource dataSource) {
        return new DataSourceProxy(dataSource);
    }
	//JPA 不需要注入sqlSessionFactoryBean
    @Bean
    public SqlSessionFactory sqlSessionFactoryBean(DataSourceProxy dataSourceProxy) throws Exception {
        SqlSessionFactoryBean sqlSessionFactoryBean = new SqlSessionFactoryBean();
        sqlSessionFactoryBean.setDataSource(dataSourceProxy);
        return sqlSessionFactoryBean.getObject();
    }
}

添加 undo_log 表

在业务相关的数据库中添加 undo_log 表，**用于保存需要回滚的数据** 每个参与分布式事务都要加这个库！
分布式事务是多个数据库的操作，给每个数据库加入一个 undo_log**日志表！**

undo_log.sql

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

启动 Seata-Server

需要的朋友，在 https://github.com/seata/seata/releases 下载相应版本的 Seata-Server
修改 registry.conf为相应的配置(如果使用 file 则不需要修改)
Seata解压即用，一般不需要配置任何东西！
启动 sh ./bin/seata-server.sh

使用@GlobalTransactional开启事务

在业务的发起方的方法上使用**@GlobalTransactional开启全局事务** 就是你业务执行的总方法！
Seata 会将事务的 xid 通过拦截器添加到调用其他服务的请求中，实现分布式事务
在事务调度的总方法上加 @GlobalTransactional 全局事务注解

基于XA协议的三阶段提交(3PC)

3PC三阶段，提交是在二阶段提交上的改进版本，主要是加入了**超时机制**。同时在协调者和参与者中都引入超时机制

XA三阶段提交在两阶段提交的基础上增加了CanCommit阶段，并且引入了超时机制。
一旦事物参与者迟迟没有接到协调者的commit请求，会自动进行本地commit。这样有效解决了协调者单点故障的问题。
但是性能问题和不一致的问题仍然没有根本解决。需要开发者介入！

三阶段将二阶段 准备阶段**拆分为2个阶段**

在原先的准备提交can Commit 后面 插入了一个preCommit 预提交阶段
以此来处理原先：二阶段参与者准备后，协调者发生崩溃或错误 导致参与者无法知晓是否提交或回滚的不确定状态所引起的延时问题。

阶段一 canCommit

不变一切正常
协调者向参与者发送 commit 请求，参与者如果可以提交就返回 yes 响应(参与者不执行事务操作)，否则返回 no 响应;

阶段二 preCommit

协调者根据阶段 1 canCommit 参与者的反应情况来决定是否可以进行基于事务的 preCommit 操作。
根据响应情况，有以下**两种可能**

情况 1：阶段 1 所有参与者均反馈 yes，参与者预执行事务

协调者向所有参与者发出 preCommit 请求，进入准备阶段。
参与者收到 preCommit 请求后，执行事务操作，将 undo 和 redo 信息记入事务日志中（但不提交事务）。
各参与者向协调者反馈 ack 成功响应或 no 失败响应，并等待最终指令。

情况 2：阶段 1 任何一个参与者反馈 no，或者等待超时后协调者尚无法收到所有参与者的反馈，即中断事务

协调者向所有参与者发出 abort 请求。
无论收到协调者发出的 abort 请求，或者在等待协调者请求过程中出现超时，参与者均会中断事务。
即： 阶段一返回一个 no 中断所有事务 参与者未接收到协调者消息中断本身事务！ 解决了协调者突然挂了的情况！

阶段三 do Commit

该阶段进行真正的事务提交 也可以分为以下两种情况。

情况一阶段 2 所有参与者均反馈 ack 响应，执行真正的事务提交

如果协调者处于工作状态，则向所有参与者发出 do Commit 请求。
参与者收到 do Commit 请求后，会正式执行事务提交，并释放整个事务期间占用的资源。
各参与者向协调者反馈 ack 完成的消息。
协调者收到所有参与者反馈的 ack 消息后，即完成事务提交。

情况二阶段 2 任何一个参与者反馈 no，或者 等待超时后协调者尚无法收到所有参与者的反馈**，即中断事务**

如果协调者处于工作状态，向所有参与者发出 abort 请求。
参与者使用阶段 1 中的 undo 信息执行回滚操作，并释放整个事务期间占用的资源。
各参与者向协调者反馈 ack 完成的消息。
协调者收到所有参与者反馈的 ack 消息后，即完成事务中断。

注意：

进入阶段 3 后，无论协调者出现问题，或者协调者与参与者网络出现问题 都会导致参与者无法接收到协调者发出的 do Commit 请求或 abort 请求。
此时，参与者都会在等待超时之后，**继续执行事务提交。**

优点：

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

缺点：

数据不一致问题依然存在
当在参与者收到 preCommit 请求后等待 do commite 指令时，此时如果协调者请求中断事务
而协调者无法与参与者正常通信，会导致参与者继续提交事务，造成数据不一致。
即：阶段三如果协调者出现故障，仍会导致数据事务不一致！！需要开发者介入！

TCC三段提交

TCC是Try**尝试**、Confirm**确认**、Cancel**撤销**三个词语的缩写 Try操作做业务检查及资源预留 Confirm做业务确认操作 Cancel实现一个与Try相反的操作 即回滚操作
TM首先发起所有的分支事务的try操作，任何一个分支事务的try操作执行失败
TM将会发起所有分支事务的Cancel操作，
若try操作全部成功，TM将会发起所有分支事务的Confirm操作
其中Confirm/Cancel操作若执行失败，TM会进行重试。

其实从思想上看和 2PC 差不多，都是先试探性的执行，如果都可以那就真正的执行，如果不行就回滚。

流程还是很简单的

难点在于业务上的定义，对于每一个操作你都需要定义三个动作分别对应Try - Confirm - Cancel 因此 TCC 对业务的侵入较大和业务紧耦合 需要根据特定的场景和业务逻辑来设计相应的操作。 很多时候需要手动补偿代码！
注意撤销和确认操作的执行可能需要重试，因此还需要保证操作的幂等。

幂等： 无论程序执行n次，保证最终执行结果唯一！

2PC 和 3PC 都是数据库层面的，而 TCC 是**业务层面的分布式事务**

开发者，手动编写逻辑，进行提交回滚，补偿代码 发送短信等…
目前本人没有具体的了解… 后面也许会更新！

本地消息表 (MQ+Table) `最终一致性`

学习之前需要了解： MQ 定时框架 本人使用的RabbitMQ 这个问题不大！

可靠消息最终一致性事务

利用消息中间件来异步完成事务的后一半更新，实现系统的最终一致性 这个方式避免了像XA协议那样的性能问题。

方案简介:

本地消息表的方案最初是由 eBay 提出, 核心思路是将分布式事务拆分成 一个个，本地事务进行处理。
事务发起方服务 数据库额外新建 事务执行消息表
事务发起方，发起处理业务开启事务 ，并记录消息在 事务消息表中
通过 定时查看，事务消息表的数据发送事务消息
事务被动方基于——消息中间件——消费事务消息表中的事务。 进行处理！

优点：

可以避免： 业务处理成功 + 事务消息发送失败 或 业务处理失败 + 事务消息发送成功 多个系统事务的数据 最终 一致性

缺点：

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

业务实现：

上面的说法可能不是很清晰，这里可以结合业务场景进行处理！
某学习平台，用户下单购买商品教程——产生订单———用户： 学习模块添加任务 学习课程！ 随便编的业务别杠

支付成功后，订单服务向本地数据库更新订单状态，并向 消息表 写入“添加选课消息” task_his.sql：任务信息 mq 交换机/队列 version版本... 这里的操作是本地事务执行要么全部失败，要么全部成功！ 且　 如果这里，就事务执行失败了，直接回滚 事务失败！
通过 定时框架 定时扫描 task_his.sql 表信息，向MQ中发送消息 避免了如果在发送消息时候，网络动荡消息发送失败！**定时发送...**
MQ 通过 Confirm 消息确认 100%发送消息 确保消息一定会发送到MQ 上！对于这里，因为定时框架会每隔一段时间，就扫描 消息表 循环向mq 上发消息！ mq 发送成功就成功！ 删除，消息表中的任务... mq 发送失败, 也不会立刻，重新发送**等待下次定时任务... 当然如果直接重新发送也行 不同场景不同处理！ 反正无论如何都会发的MQ 上！**
学习模块实时监听 MQ的消息队列，只要有新消息就接收，并继续执行自己的业务操作 同时获取自己的结果消息 事务成功/失败 根据MQ的 消息确认接收机制（ACK） 消息一旦被消费者接收返回 ack，队列中的消息就会被删除。 使用手动ACK： 消息接收后，不会立刻发送ACK , 学习模块事务执行完毕才返回 ack 确保了消息的消费！ 学习模块的事务无论执行成功/失败，都会在像MQ发送一条消息！

注意这里还是要做 幂等的操作！ 无论程序执行n次，对于相同的请求保证结果唯一！

方式很多，可以是根据任务请求内容，获取订单id 判断是否执行过…

订单模块接收MQ 上消息，判断事务是否执行成功！回滚/提交。 并删除**消息表的消息！ 就避免了消息在次 定时发送!**

MQ: ack 和 Confirm 消息确认

ack 保证消息一定被消费

消息一旦被消费者接收，队列中的消息就会被删除。
RabbitMQ怎么知道消息被接收了呢？ 如果消费者领取消息后，还没执行操作就挂掉了呢？或者抛出了异常？消息消费失败，但是RabbitMQ无从得知，这样消息就丢失了！ 因此，RabbitMQ有一个ACK机制。 当消费者获取消息后，会向RabbitMQ发送回执ACK，告知消息已经被接收。不过这种回执ACK分两种情况： 自动ACK： 消息一旦被接收，消费者自动发送ACK 手动ACK： 消息接收后，不会发送ACK，需要手动调用
使用场景： 如果消息不太重要，丢失也没有影响，那么自动ACK会比较方便 如果消息非常重要，不容丢失。那么最好在消费完成后手动ACK，否则接收消息后就自动ACK

Confirm 保证消息一定发送成功

消息的确认，是指生产者投递消息后，如果 MQ 收到消息，则会给我们生产者一个应答。
生产者进行接收应答，用来确定这条消息是否正常的发送到 MQ，这种方式也是消息的可靠性投递的核心保障!
消息发送成功和失败, 会执行对应发送者的 回调方法！

常见的消息 `任务`表：

不是绝对的，这个消息表根据实际开发中更改即可!

task_his.sql

DROP TABLE IF EXISTS `task_his`;

CREATE TABLE `task_his` (
  `id` varchar(32) NOT NULL COMMENT '任务id',
  `create_time` datetime DEFAULT NULL,
  `update_time` datetime DEFAULT NULL,
  `delete_time` datetime DEFAULT NULL,
  `task_type` varchar(32) DEFAULT NULL COMMENT '任务类型',
  `mq_exchange` varchar(64) DEFAULT NULL COMMENT '交换机名称',
  `mq_routingkey` varchar(64) DEFAULT NULL COMMENT 'routingkey',
  `request_body` varchar(512) DEFAULT NULL COMMENT '任务请求的内容',
  `version` int(10) DEFAULT '0' COMMENT '乐观锁版本号',
  `status` varchar(32) DEFAULT NULL COMMENT '任务状态',
  `errormsg` varchar(512) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

任务创建时间更新时间删除时间 要操作MQ 交换机队列信息！
为方式分布式多服务 version 设置乐观锁版本号！
…

Spring Task定时任务

依赖：
在Spring boot启动类上添加注解：@EnableScheduling
任务类
Quartz 是一个异步任务调度框架，功能丰富，可以实现按日历调度

定时任务类：

仅供参考…

MessageTaskJob.Java

@Component
public class MessageTaskJob {

    @Autowired
    private TbTaskService taskService;     //消息表业务对象！

    @Scheduled(cron = "0/30 * * * * ?")    //定时每三十秒扫描一次 消息表！
    public void showTask() {
        System.out.println("查询任务数据！");
        try {
            //获取一分钟前消息表数据！
            List<TbTask> list = taskService.getBeforTaskList();

            //循环变量发送MQ消息...
            for (TbTask tbTask : list) {
                //使用乐观锁解决高并发下的信息发送...后面解释！
                if (taskService.updateVersionLock(tbTask.getId(), tbTask.getVersion()) > 0) {
                    //发送MQ...
                    taskService.publishTaskMessage(tbTask);
                    //发送之后修改,当前 订单消息表的记录时间..当前时间即可！
                        //因为定时会循环,如果消息发送失败..则直接下次在轮询重新发送即可！
                    taskService.updateTaskUpdateTime(tbTask.getId());
                    System.out.println("发送消息并,修改时间!");
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

修改时间，是为了, 防止消息发送失败等原因, 将时间修改为当前时间，**定时操作执行在次发送...** 确保最终一致性！
为啥MQ 存在消息确认机制，还要这么做. 不同的人不同的写法… 反之最终要发到 mq 上就ok了… 这里将时间重排，重新定时发送 个人觉得为了公平… 你去银行排队办理业务，缺少文件回去办理…回来了当然要重新排队.

Mybatis sql参考

    <!-- 获取一分钟前消息表的所有数据！ -->
    <select id="getBeforTaskList" resultType="com.zb.entity.TbTask">
        select
        id as id,
        create_time as createTime,
        update_time as updateTime,
        delete_time as deleteTime,
        task_type as taskType,getBeforTaskList
        mq_exchange as mqExchange,
        mq_routingkey as mqRoutingkey,
        request_body as requestBody,
        status as status,
        errormsg as errormsg,
        version as version
        from tb_task
         WHERE TIMESTAMPDIFF(MINUTE,update_time, NOW())>1
    </select>

    <!-- 发送消息成功更改时间... -->
    <update id="updateNowTime" >
        update tb_task set update_time =now() where id=#{id}
    </update>
    <!-- 乐观锁：多线程情况下防止,统一消息发送多次处理...
            根据 id version版本 来进行修改..
    -->
    <update id="updateVersioLock">
        update tb_task set version =#{version}+1 where id=#{id} and version=#{version}
    </update>

乐观锁：解决分布式锁.

考虑订单服务将来会集群部署
为了避免任务在 定时任务 内重复执行，这里使用乐观锁 A B 线程执行任务都查到了 消息集合 A 发送了任务1 B 也发送了任务1 重复操作！
乐观锁处理: A B线程都进入方法获取到 消息集合 Ａ先执行并带着版本号+id去给版本+1 B因为查的 消息集合并不是最新的id+版本号去修改影响行数小于0 不执行发送任务！

但如果：A线程执行特别快修改了版本，但是B执行慢才查到，**获取了最新的消息集合**

就悲剧了…
所以，可以在A修改版本号同时，修改当前时间…
这样：B无论执行快慢都避免了重复操作！快：因为版本 id 修改不了，影响行数不大于 0 慢：因为时间更改了，直接就获取不到最新的数据了…
这只是我的个人想法： 实际开发另算根据场景来定~

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2024-08-06，如有侵权请联系 cloudcommunity@tencent.com 删除

事务

本文分享自作者个人站点/博客前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体同步曝光计划，欢迎热爱写作的你一起参与！

分布式事务

分布式事务

分布式事务

前言：

什么是事务

数据库事务的四大特性 ACID：

本地事务 Local Transaction

分布式事务 | 产生的场景

单一服务分布式事务

多服务分布式事务

多服务多数据源分布式事务

事务的作用：

分布式事务基础理论

CAP理论

理解CAP

CAP有哪些组合方式呢？

面试题：ZooKeeper 和 eureka 的区别

总结

BASE理论

基本可用:

软状态:

最终一致:

分布式事务解决方案

XA分布式事务协议

DTP模型

基于XA协议的两阶段提交(2PC)

2PC总结：

Seata 实现2PC

Seata方案

Seata的设计思想：

具体执行流程：

Seata 2PC与传统2PC的差别

Spring Cloud 快速集成 Seata

依赖：

添加配置文件

修改应用程序yml

注入数据源

添加 undo_log 表

启动 Seata-Server

使用@GlobalTransactional开启事务

基于XA协议的三阶段提交(3PC)

阶段一 canCommit

阶段二 preCommit

阶段三 do Commit

注意：

优点：

缺点：

TCC三段提交

本地消息表 (MQ+Table) 最终一致性

可靠消息最终一致性事务

业务实现：

MQ: ack 和 Confirm 消息确认

常见的消息 任务表：

Spring Task定时任务

定时任务类：

乐观锁：解决分布式锁.

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

本地事务 `Local Transaction`

Seata　2PC与传统2PC的差别

本地消息表 (MQ+Table) `最终一致性`

常见的消息 `任务`表：