You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

28 KiB

分布式事务

一、为什么需要使用分布式事务

  我们在分布式环境下一个业务可能会涉及到多个模块之间的调用,为了保证操作的原子性,分布式事务是最好的解决方案。

image.png

二、本地事务

  在系统介绍分布式事务之前,我们还是很有必要回顾下本地事务。在一个服务中生效的事务我们称为本地事务。

1.事务的特性

  事务的概念:事务是逻辑上一组操作,组成这组操作各个逻辑单元,要么一起成功,要么一起失败。

事务的四个特性ACID

  1. 原子性(atomicity):“原子”的本意是“不可再分”,事务的原子性表现为一个事务中涉及到的多个操作在逻辑上缺一不可。事务的原子性要求事务中的所有操作要么都执行,要么都不执行。
  2. 一致性(consistency)一致指的是数据的一致,具体是指:所有数据都处于满足业务规则的一致性状态。一致性原则要求:一个事务中不管涉及到多少个操作,都必须保证事务执行之前数据是正确的,事务执行之后数据仍然是正确的。如果一个事务在执行的过程中,其中某一个或某几个操作失败了,则必须将其他所有操作撤销,将数据恢复到事务执行之前的状态,这就是回滚。
  3. 隔离性(isolation):在应用程序实际运行过程中,事务往往是并发执行的,所以很有可能有许多事务同时处理相同的数据,因此每个事务都应该与其他事务隔离开来,防止数据损坏。隔离性原则要求多个事务在并发执行过程中不会互相干扰
  4. 持久性(durability):持久性原则要求事务执行完成后,对数据的修改永久的保存下来,不会因各种系统错误或其他意外情况而受到影响。通常情况下,事务对数据的修改应该被写入到持久化存储器中。

2.事务的隔离级别

事务并发引起一些读的问题:

  • 脏读 一个事务可以读取另一个事务未提交的数据
  • 不可重复读 一个事务可以读取另一个事务已提交的数据 单条记录前后不匹配
  • 虚读(幻读) 一个事务可以读取另一个事务已提交的数据 读取的数据前后多了点或者少了点

并发写使用mysql默认的锁机制独占锁

解决读问题:设置事务隔离级别

  • read uncommitted(0)
  • read committed(2)
  • repeatable read(4)
  • Serializable(8)

隔离级别越高,性能越低。

image.png

一般情况下:脏读是不可允许的,不可重复读和幻读是可以被适当允许的。

3.事务的传播属性

Spring中的7个事务传播行为:

事务行为 说明
PROPAGATION_REQUIRED 支持当前事务,假设当前没有事务。就新建一个事务
PROPAGATION_SUPPORTS 支持当前事务,假设当前没有事务,就以非事务方式运行
PROPAGATION_MANDATORY 支持当前事务,假设当前没有事务,就抛出异常
PROPAGATION_REQUIRES_NEW 新建事务,假设当前存在事务。把当前事务挂起
PROPAGATION_NOT_SUPPORTED 以非事务方式运行操作。假设当前存在事务,就把当前事务挂起
PROPAGATION_NEVER 以非事务方式运行,假设当前存在事务,则抛出异常
PROPAGATION_NESTED 如果当前存在事务则在嵌套事务内执行。如果当前没有事务则执行与PROPAGATION_REQUIRED类似的操作。

举例说明

ServiceA

ServiceA {   
     void methodA() {
         ServiceB.methodB();
     }
}

ServiceB

ServiceB { 
     void methodB() {
     }  
}

1.PROPAGATION_REQUIRED

  假如当前正要运行的事务不在另外一个事务里,那么就起一个新的事务 比方说ServiceB.methodB的事务级别定义PROPAGATION_REQUIRED, 那么因为执行ServiceA.methodA的时候ServiceA.methodA已经起了事务。这时调用ServiceB.methodBServiceB.methodB看到自己已经执行在ServiceA.methodA的事务内部。就不再起新的事务。而假如ServiceA.methodA执行的时候发现自己没有在事务中他就会为自己分配一个事务。这样在ServiceA.methodA或者在ServiceB.methodB内的不论什么地方出现异常。事务都会被回滚。即使ServiceB.methodB的事务已经被提交可是ServiceA.methodA在接下来fail要回滚ServiceB.methodB也要回滚 在这里插入图片描述

2.PROPAGATION_SUPPORTS

  假设当前在事务中。即以事务的形式执行。假设当前不在一个事务中,那么就以非事务的形式执行

3.PROPAGATION_MANDATORY

  必须在一个事务中执行。也就是说,他仅仅能被一个父事务调用。否则,他就要抛出异常

4.PROPAGATION_REQUIRES_NEW

  这个就比较绕口了。 比方我们设计ServiceA.methodA的事务级别为PROPAGATION_REQUIREDServiceB.methodB的事务级别为PROPAGATION_REQUIRES_NEW。那么当运行到ServiceB.methodB的时候ServiceA.methodA所在的事务就会挂起。ServiceB.methodB会起一个新的事务。等待ServiceB.methodB的事务完毕以后他才继续运行。 他与PROPAGATION_REQUIRED 的事务差别在于事务的回滚程度了。由于ServiceB.methodB是新起一个事务那么就是存在两个不同的事务。假设ServiceB.methodB已经提交那么ServiceA.methodA失败回滚。ServiceB.methodB是不会回滚的。假设ServiceB.methodB失败回滚假设他抛出的异常被ServiceA.methodA捕获ServiceA.methodA事务仍然可能提交。 在这里插入图片描述

5.PROPAGATION_NOT_SUPPORTED

当前不支持事务。比方ServiceA.methodA的事务级别是PROPAGATION_REQUIRED 。而ServiceB.methodB的事务级别是PROPAGATION_NOT_SUPPORTED 那么当执行到ServiceB.methodB时。ServiceA.methodA的事务挂起。而他以非事务的状态执行完再继续ServiceA.methodA的事务。

6.PROPAGATION_NEVER

  不能在事务中执行。 如果ServiceA.methodA的事务级别是PROPAGATION_REQUIRED。 而ServiceB.methodB的事务级别是PROPAGATION_NEVER 那么ServiceB.methodB就要抛出异常了。

7.PROPAGATION_NESTED

如果当前存在事务则在嵌套事务内执行。如果当前没有事务则执行与PROPAGATION_REQUIRED类似的操作。

@Transactional(propagation=Propagation.REQUIRED) 如果有事务, 那么加入事务, 没有的话新建一个(默认情况下) @Transactional(propagation=Propagation.NOT_SUPPORTED) 容器不为这个方法开启事务 @Transactional(propagation=Propagation.REQUIRES_NEW) 不管是否存在事务,都创建一个新的事务,原来的挂起,新的执行完毕,继续执行老的事务 @Transactional(propagation=Propagation.MANDATORY) 必须在一个已有的事务中执行,否则抛出异常 @Transactional(propagation=Propagation.NEVER) 必须在一个没有的事务中执行,否则抛出异常(与Propagation.MANDATORY相反) @Transactional(propagation=Propagation.SUPPORTS) 如果其他bean调用这个方法,在其他bean中声明事务,那就用事务.如果其他bean没有声明事务,那就不用事务.

4.SpringBoot事务代理对象

在SpringBoot中如果一个对象中有多个事务方法相互调用那么事务传播会失效主要原因是当前对象直接调用了自身对象的方法绕过了代理对象的处理造成了事务传播的失效。那么对应的解决方案是 spring-boot-stater-aop 来显示的获取代理对象来调用

引入相关的依赖

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-aop</artifactId>
            <version>2.4.12</version>
        </dependency>

添加aspectj的注解

image.png

然后在需要调用的位置通过 AopContext获取当前的代理对象

    /**
     * 在service中调用自身的其他事务方法的时候事务的传播行为会失效
     *   因为会绕过代理对象的处理
     *
     */
    @Transactional // 事务A
    public void a(){
        OrderServiceImpl o = (OrderServiceImpl) AopContext.currentProxy();
        o.b(); // 事务A
        o.c(); // 事务C
        int a = 10/0;
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public void b(){

    }

    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void c(){

    }

三、分布式事务

1.分布式事务基础

CAP定理

分布式存储系统的CAP原理分布式系统的三个指标

  1. Consistency一致性在分布式系统中的所有数据备份同一时刻是否同样的值

    对于数据分布在不同节点上的数据来说,如果在某个节点更新了数据,那么在其他节点如果都能读取到这个最新的数据,那么就称为强一致,如果有某个节点没有读取到,那就是分布式不一致。

  2. Availability可用性在集群中一部分节点故障后集群整体是否还能响应客户端的读写请求。要求数据需要备份

  3. Partition tolerance分区容忍性大多数分布式系统都分布在多个子网络。每个子网络就叫做一个区partition。分区容错的意思是区间通信可能失败。

CAP理论就是说在分布式存储系统中最多只能实现上面的两点。而由于当前的网络硬件肯定会出现延迟丢包等问题所以分区容忍性是我们无法避免的。所以我们只能在一致性和可用性之间进行权衡没有系统能同时保证这三点。要么选择CP、要么选择AP。

image.png

BASE定理

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

  1. Basically Available基本可用

    基本可用是指分布式系统在出现故障的时候,允许损失部分可用性,即保证核心可用。 电商大促时,为了应对访问量激增,部分用户可能会被引导到降级页面,服务层也可能只提供降级服务。这就是损失部分可用性的体现。

  2. Soft state软状态

    软状态是指允许系统存在中间状态,而该中间状态不会影响系统整体可用性。分布式存储中一般一份数据至少会有三个副本,允许不同节点间副本同步的延时就是软状态的体现。mysql replication的异步复制也是一种体现。

  3. Eventually consistent最终一致性

    最终一致性是指系统中的所有数据副本经过一定时间后,最终能够达到一致的状态。弱一致性和强一致性相反,最终一致性是弱一致性的一种特殊情况。

BASE模型是传统ACID模型的反面不同于ACIDBASE强调牺牲高一致性从而获得可用性数据允许在一段时间内的不一致,只要保证最终一致就可以了

2.分布式事务的解决方案

https://www.processon.com/view/link/62a1ddce0791293ad1a552c0

  分布式事务是企业集成中的一个技术难点,也是每一个分布式系统架构中都会涉及到的一个东西,特别是在微服务架构中,几乎可以说是无法避免

主流的解决方案如下:

  1. 基于XA协议的两阶段提交2PC
  2. 柔性事务-TCC事务
  3. 柔性事务-最终一致性

2.1 两阶段提交2PC

2PC即两阶段提交协议是将整个事务流程分为两个阶段准备阶段Prepare phase、提交阶段commit phase2是指两个阶段P是指准备阶段C是指提交阶段。

image.png

第一阶段:事务协调器要求每个涉及到事务的数据库预提交(precommit)此操作,并反映是否可以提交.

第二阶段:事务协调器要求每个数据库提交数据。

其中,如果有任何一个数据库否决此次提交,那么所有数据库都会被要求回滚它们在此事务中的那部分信息。

目前主流数据库均支持2PC【2 Phase Commit】

XA 是一个两阶段提交协议,又叫做 XA Transactions。

MySQL从5.5版本开始支持SQL Server 2005 开始支持Oracle 7 开始支持。

总的来说XA协议比较简单而且一旦商业数据库实现了XA协议使用分布式事务的成本也比较低。但是XA也有致命的缺点那就是性能不理想特别是在交易下单链路往往并发量很高XA无法满足高并发场景。

  1. 两阶段提交涉及多次节点间的网络通信,通信时间太长!
  2. 事务时间相对于变长了,锁定的资源的时间也变长了,造成资源等待时间也增加好多。
  3. XA目前在商业数据库支持的比较理想在mysql数据库中支持的不太理想mysql的XA实现没有记录prepare阶段日志主备切换会导致主库与备库数据不一致。许多nosql也没有支持XA这让XA的应用场景变得非常狭隘。

2.2 TCC补偿式事务

TCC 是一种编程式分布式事务解决方案。

TCC 其实就是采用的补偿机制其核心思想是针对每个操作都要注册一个与其对应的确认和补偿撤销操作。TCC模式要求从服务提供三个接口Try、Confirm、Cancel。

  • Try主要是对业务系统做检测及资源预留
  • Confirm真正执行业务不作任何业务检查只使用Try阶段预留的业务资源Confirm操作满足幂等性。
  • Cancel释放Try阶段预留的业务资源Cancel操作满足幂等性。

整个TCC业务分成两个阶段完成

image.png

第一阶段主业务服务分别调用所有从业务的try操作并在活动管理器中登记所有从业务服务。当所有从业务服务的try操作都调用成功或者某个从业务服务的try操作失败进入第二阶段。

第二阶段活动管理器根据第一阶段的执行结果来执行confirm或cancel操作。如果第一阶段所有try操作都成功则活动管理器调用所有从业务活动的confirm操作。否则调用所有从业务服务的cancel操作。

举个例子,假如 Bob 要向 Smith 转账100元思路大概是

我们有一个本地方法,里面依次调用

  1. 首先在 Try 阶段要先检查Bob的钱是否充足并把这100元锁住Smith账户也冻结起来。
  2. 在 Confirm 阶段,执行远程调用的转账的操作,转账成功进行解冻。
  3. 如果第2步执行成功那么转账成功如果第二步执行失败则调用远程冻结接口对应的解冻方法 (Cancel)。

缺点:

  • Canfirm和Cancel的幂等性很难保证。
  • 这种方式缺点比较多,通常在复杂场景下是不推荐使用除非是非常简单的场景非常容易提供回滚Cancel而且依赖的服务也非常少的情况。
  • 这种实现方式会造成代码量庞大,耦合性高。而且非常有局限性,因为有很多的业务是无法很简单的实现回滚的,如果串行的服务很多,回滚的成本实在太高。

不少大公司里,其实都是自己研发 TCC 分布式事务框架的专门在公司内部使用。国内开源出去的ByteTCCTCC-transactionHimly。

2.3 消息事务+最终一致性

基于消息中间件的两阶段提交往往用在高并发场景下将一个分布式事务拆成一个消息事务A系统的本地操作+发消息)+B系统的本地操作其中B系统的操作由消息驱动只要消息事务成功那么A操作一定成功消息也一定发出来了这时候B会收到消息去执行本地操作如果本地操作失败消息会重投直到B操作成功这样就变相地实现了A与B的分布式事务。

image.png

虽然上面的方案能够完成A和B的操作但是A和B并不是严格一致的而是最终一致的我们在这里牺牲了一致性换来了性能的大幅度提升。当然这种玩法也是有风险的如果B一直执行不成功那么一致性会被破坏具体要不要玩还是得看业务能够承担多少风险。

适用于高并发最终一致

低并发基本一致:二阶段提交

高并发强一致:没有解决方案

3.seata

分布式事务解决方案seata

官网:https://seata.io/zh-cn/docs/overview/what-is-seata.html

GitHubhttps://github.com/seata/seata

3.1 seata服务的安装

我们可以先导入seata的依赖根据依赖的版本来下载对应的seata安装文件

image.png

https://github.com/seata/seata/releases

下载后解压缩然后进入conf文件夹然后通过registery.conf文件可以更新配置中心和注册中心的信息。

image.png

然后进入 bin 目录 通过 seata-server.bat文件来启动服务

image.png

然后进入nacos注册中心可以看到对应的服务表示OK

image.png

3.2 项目集成Seata

接下来看看如何在商城项目中来集成Seata首先是file.conf文件

网络传输配置:

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
  }
}

事务日志存储配置该部分配置仅在seata-server中使用如果选择db请配合seata.sql使用

## transaction log store, only used in seata-server
store {
  ## store mode: file、db
  mode = "file"

  ## file store property
  file {
    ## store location dir
    dir = "sessionStore"
  }

  ## database store property
  db {
    ## the implement of javax.sql.DataSource, such as DruidDataSource(druid)/BasicDataSource(dbcp) etc.
    datasource = "dbcp"
    ## mysql/oracle/h2/oceanbase etc.
    db-type = "mysql"
    driver-class-name = "com.mysql.jdbc.Driver"
    url = "jdbc:mysql://127.0.0.1:3306/seata"
    user = "mysql"
    password = "mysql"
  }
}

*当前微服务在seata服务器中注册的信息配置

service {
  # 事务分组,默认:${spring.applicaiton.name}-fescar-service-group可以随便写
  vgroup_mapping.${spring.application.name}-fescar-service-group = "default"
  # 仅支持单节点不要配置多地址这里的default要和事务分组的值一致
  default.grouplist = "127.0.0.1:8091" #seata-server服务器地址默认是8091
  # 降级,当前不支持
  enableDegrade = false
  # 禁用全局事务
  disableGlobalTransaction = false
}

客户端相关工作的机制

client {
  rm {
    async.commit.buffer.limit = 10000
    lock {
      retry.internal = 10
      retry.times = 30
      retry.policy.branch-rollback-on-conflict = true
    }
    report.retry.count = 5
    table.meta.check.enable = false
    report.success.enable = true
  }
  tm {
    commit.retry.count = 5
    rollback.retry.count = 5
  }
  undo {
    data.validation = true
    log.serialization = "jackson"
    log.table = "undo_log"
  }
  log {
    exceptionRate = 100
  }
  support {
    # auto proxy the DataSource bean
    spring.datasource.autoproxy = false
  }
}

首先我们需要把registry.conf和file.conf两个配置文件拷贝到对应的项目的属性文件目录中

image.png

然后我们在属性文件中定义 tx-service-group 信息

image.png

然后在file.conf中添加service属性然后关联刚刚设置的tx-service-group信息。注意1.1版本后属性更新为了驼峰命名法:

image.png

3.3 案例演示

  我们在下订单的操作中除了已有的生成订单和订单项已经锁定库存操作外我们还显示的增加的了一个会有积分调整的服务。[1/0],这样一来如果锁定库存成功,但是会员积分调整失败,被分布式事务管理的逻辑中,锁库存的操作会回滚。

image.png

image.png

4.取消订单

  取消订单出现的情况:

  • 下订单后超过30分钟没有支付需要触发关单操作
  • 支付失败,同样的需要关单

实现方式:定时任务和消息中间件,定时任务对系统的性能肯定是有影响

image.png

image.png

image.png

RocketMQhttps://github.com/apache/rocketmq/tree/master/docs/cn

image.png

RocketMQDocker安装

安装NameServer

docker pull rocketmqinc/rocketmq

创建存储目录

mkdir -p /mydata/docker/rocketmq/data/namesrv/logs /mydata/docker/rocketmq/data/namesrv/store

然后安装

docker run -d --restart=always --name rmqnamesrv --privileged=true -p 9876:9876  -v /docker/rocketmq/data/namesrv/logs:/root/logs -v /docker/rocketmq/data/namesrv/store:/root/store -e "MAX_POSSIBLE_HEAP=100000000" rocketmqinc/rocketmq sh mqnamesrv

相关参数说明

image.png

安装Broker

border配置创建 broker.conf 配置文件 vi /mydata/rocketmq/conf/broker.conf ,配置如下:

# 所属集群名称,如果节点较多可以配置多个
brokerClusterName = DefaultCluster 
#broker名称master和slave使用相同的名称表明他们的主从关系 
brokerName = broker-a 
#0表示Master大于0表示不同的
slave brokerId = 0 
#表示几点做消息删除动作默认是凌晨4点 
deleteWhen = 04 
#在磁盘上保留消息的时长,单位是小时 
fileReservedTime = 48 
#有三个值SYNC_MASTERASYNC_MASTERSLAVE同步和异步表示Master和Slave之间同步数据的机 制;
brokerRole = ASYNC_MASTER 
#刷盘策略取值为ASYNC_FLUSHSYNC_FLUSH表示同步刷盘和异步刷盘SYNC_FLUSH消息写入磁盘后 才返回成功状态ASYNC_FLUSH不需要
flushDiskType = ASYNC_FLUSH 
# 设置broker节点所在服务器的ip地址 
brokerIP1 = 192.168.100.130 
#剩余磁盘比例 
diskMaxUsedSpaceRatio=99

安装:

docker run -d --restart=always --name rmqbroker --link rmqnamesrv:namesrv -p 10911:10911 -p 10909:10909 --privileged=true -v /docker/rocketmq/data/broker/logs:/root/logs -v /docker/rocketmq/data/broker/store:/root/store -v /docker/rocketmq/conf/broker.conf:/opt/rocketmq-4.4.0/conf/broker.conf -e "NAMESRV_ADDR=namesrv:9876" -e "MAX_POSSIBLE_HEAP=200000000" rocketmqinc/rocketmq sh mqbroker -c /opt/rocketmq-4.4.0/conf/broker.conf

相关参数说明:

image.png

安装控制台

拉取镜像

docker pull pangliang/rocketmq-console-ng

控制台安装:

docker run -d --restart=always --name rmqadmin -e "JAVA_OPTS=-Drocketmq.namesrv.addr=192.168.100.130:9876 -Dcom.rocketmq.sendMessageWithVIPChannel=false" -p 8080:8080 pangliang/rocketmq-console-ng

访问测试:http://192.168.100.100:8080

image.png

5.释放库存

  需要释放库存的情况:

  • 下订单后手动取消订单获取超时支付订单
  • 支付成功释放库存,更新库存

SpringBoot整合RocketMQ

需要添加对应的依赖

<dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>2.2.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.9.1</version>
        </dependency>

需要添加对应的配置信息

image.png

然后定义对应的消息生产者

@Component
public class OrderMsgProducer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public  void sendOrderMessage(String orderSN){

        rocketMQTemplate.syncSend(OrderConstant.ROCKETMQ_ORDER_TOPIC, MessageBuilder.withPayload(orderSN).build(),5000,4);
    }
}

然后就是延迟消息的消费者,我们得通过对应的监听器来处理

@RocketMQMessageListener(topic = OrderConstant.ROCKETMQ_ORDER_TOPIC,consumerGroup = "${rocketmq.consumer.group}")
@Component
public class OrderMsgConsumer implements RocketMQListener<String> {

    @Override
    public void onMessage(String s) {
        System.out.println("收到的消息:" + s);
    }
}

然后就是执行的效果

image.png