分布式事务原理及解决方案

时间 2019-11-07

标签分布式事务原理解决方案栏目系统架构繁體版

原文原文链接

1 引言

分布式事务是企业集成中的一个技术难点，也是每个分布式系统架构中都会涉及到的一个东西，特别是在这几年愈来愈火的微服务架构中，几乎能够说是没法避免，本文就围绕单机事务，分布式事务以及分布式事务的处理方式来展开。java

2 事务

事务提供一种“要么什么都不作，要么作全套（All or Nothing）”的机制，她有ACID四大特性mysql

原子性（Atomicity）：事务做为一个总体被执行，包含在其中的对数据库的操做要么所有被执行，要么都不执行。
一致性（Consistency）：事务应确保数据库的状态从一个一致状态转变为另外一个一致状态。一致状态是指数据库中的数据应知足完整性约束。除此以外，一致性还有另一层语义，就是事务的中间状态不能被观察到（这层语义也有说应该属于原子性）。
隔离性（Isolation）：多个事务并发执行时，一个事务的执行不该影响其余事务的执行，如同只有这一个操做在被数据库所执行同样。
持久性（Durability）：已被提交的事务对数据库的修改应该永久保存在数据库中。在事务结束时，此操做将不可逆转。

2.1 单机事务

以mysql的InnoDB存储引擎为例，来了解单机事务是如何保证ACID特性的。 git

事务的隔离性是经过数据库锁的机制实现的，持久性经过redo log（重作日志）来实现，原子性和一致性经过Undo log来实现。

2.2 分布式事务

单机事务是经过将操做限制在一个会话内经过数据库自己的锁以及日志来实现ACID，那么分布式环境下该如何保证ACID特性那?github

2.2.1 XA协议实现分布式事务

2.2.1.1 XA描述

X/Open DTP(X/Open Distributed Transaction Processing Reference Model) 是X/Open 这个组织定义的一套分布式事务的标准，也就是了定义了规范和API接口，由各个厂商进行具体的实现。 X/Open DTP 定义了三个组件： AP，TM，RM
web

AP(Application Program)：也就是应用程序，能够理解为使用DTP的程序
RM(Resource Manager)：资源管理器，这里能够理解为一个DBMS系统，或者消息服务器管理系统，应用程序经过资源管理器对资源进行控制。资源必须实现XA定义的接口
TM(Transaction Manager)：事务管理器，负责协调和管理事务，提供给AP应用程序编程接口以及管理资源管理器

其中在DTP定义了如下几个概念spring

事务：一个事务是一个完整的工做单元，由多个独立的计算任务组成，这多个任务在逻辑上是原子的
全局事务：对于一次性操做多个资源管理器的事务，就是全局事务
分支事务：在全局事务中，某一个资源管理器有本身独立的任务，这些任务的集合做为这个资源管理器的分支任务
控制线程：用来表示一个工做线程，主要是关联AP,TM,RM三者的一个线程，也就是事务上下文环境。简单的说，就是须要标识一个全局事务以及分支事务的关系

若是一个事务管理器管理着多个资源管理器，DTP是经过两阶段提交协议来控制全局事务和分支事务。sql

第一阶段：准备阶段事务管理器通知资源管理器准备分支事务，资源管理器告之事务管理器准备结果
第二阶段：提交阶段事务管理器通知资源管理器提交分支事务，资源管理器告之事务管理器结果

2.2.1.2 XA的ACID特性

原子性：XA议使用2PC原子提交协议来保证分布式事务原子性
隔离性：XA要求每一个RMs实现本地的事务隔离，子事务的隔离来保证整个事务的隔离。
一致性：经过原子性、隔离性以及自身一致性的实现来保证“数据库从一个一致状态转变为另外一个一致状态”；经过MVCC来保证中间状态不能被观察到。

2.2.1.3 XA的优缺点

优势：
对业务无侵入，对RM要求高
缺点：
同步阻塞：在二阶段提交的过程当中，全部的节点都在等待其余节点的响应，没法进行其余操做。这种同步阻塞极大的限制了分布式系统的性能。

单点问题：协调者在整个二阶段提交过程当中很重要，若是协调者在提交阶段出现问题，那么整个流程将没法运转。更重要的是，其余参与者将会处于一直锁定事务资源的状态中，而没法继续完成事务操做。

数据不一致：假设当协调者向全部的参与者发送commit请求以后，发生了局部网络异常，或者是协调者在还没有发送完全部 commit请求以前自身发生了崩溃，致使最终只有部分参与者收到了commit请求。这将致使严重的数据不一致问题。

容错性很差：若是在二阶段提交的提交询问阶段中，参与者出现故障，致使协调者始终没法获取到全部参与者的确认信息，这时协调者只能依靠其自身的超时机制，判断是否须要中断事务。显然，这种策略过于保守。换句话说，二阶段提交协议没有设计较为完善的容错机制，任意一个节点是失败都会致使整个事务的失败。

2.2.2 TCC协议实现分布式事务

2.2.2.1 TCC描述

TCC（Try-Confirm-Cancel）分布式事务模型相对于 XA 等传统模型，其特征在于它不依赖资源管理器（RM）对分布式事务的支持，而是经过对业务逻辑的分解来实现分布式事务。
数据库

第一阶段：CanCommit

3PC的CanCommit阶段其实和2PC的准备阶段很像。协调者向参与者发送commit请求，参与者若是能够提交就返回Yes响应，不然返回No响应。
事务询问：协调者向参与者发送CanCommit请求。询问是否能够执行事务提交操做。而后开始等待参与者的响应
响应反馈：参与者接到CanCommit请求以后，正常状况下，若是其自身认为能够顺利执行事务，则返回Yes响应，并进入预备状态；不然反馈No。apache

第二阶段：PreCommit
协调者在获得全部参与者的响应以后，会根据结果执行2种操做：执行事务预提交，或者中断事务

执行事务预提交
发送预提交请求：协调者向全部参与者节点发出 preCommit 的请求，并进入 prepared 状态。
事务预提交：参与者受到 preCommit 请求后，会执行事务操做，对应 2PC 准备阶段中的 “执行事务”，也会 Undo 和 Redo 信息记录到事务日志中。
各参与者响应反馈:若是参与者成功执行了事务，就反馈 ACK 响应，同时等待指令：提交（commit）或终止（abort）编程

中断事务
发送中断请求：协调者向全部参与者节点发出 abort 请求。
中断事务：参与者若是收到 abort 请求或者超时了，都会中断事务。

第三阶段：Do Commit
该阶段进行真正的事务提交，也能够分为如下两种状况

执行提交
发送提交请求：协调者接收到各参与者发送的ACK响应，那么他将从预提交状态进入到提交状态。并向全部参与者发送 doCommit 请求。
事务提交：参与者接收到 doCommit 请求以后，执行正式的事务提交。并在完成事务提交以后释放全部事务资源。
响应反馈：事务提交完以后，向协调者发送 ACK 响应。
完成事务：协调者接收到全部参与者的 ACK 响应以后，完成事务。

中断事务
协调者没有接收到参与者发送的 ACK 响应（多是接受者发送的不是ACK响应，也可能响应超时），那么就会执行中断事务。
发送中断请求：协调者向全部参与者发送 abort 请求。
事务回滚：参与者接收到 abort 请求以后，利用其在阶段二记录的 undo 信息来执行事务的回滚操做，并在完成回滚以后释放全部的事务资源。
反馈结果：参与者完成事务回滚以后，向协调者发送 ACK 消息。
中断事务：协调者接收到参与者反馈的 ACK 消息以后，完成事务的中断。

2.2.2.2 TCC的ACID特性

原子性：TCC 模型也使用 2PC 原子提交协议来保证事务原子性。Try 操做对应2PC 的一阶段准备（Prepare）；Confirm 对应 2PC 的二阶段提交（Commit），Cancel 对应 2PC 的二阶段回滚（Rollback），能够说 TCC 就是应用层的 2PC。
隔离性：隔离的本质是控制并发，放弃在数据库层面加锁经过在业务层面加锁来实现。【好比在帐户管理模块设计中，增长可用余额和冻结金额的设置】
一致性：经过原子性保证事务的原子提交、业务隔离性控制事务的并发访问，实现分布式事务的一致性状态转变；事务的中间状态不能被观察到这点并不保证[本协议是基于柔性事务理论提出的]。

2.2.2.3 TCC的优缺点

优势：
相对于二阶段提交，三阶段提交主要解决的单点故障问题，并减小了阻塞的时间。由于一旦参与者没法及时收到来自协调者的信息以后，他会默认执行 commit。而不会一直持有事务资源并处于阻塞状态。
缺点：
三阶段提交也会致使数据一致性问题。因为网络缘由，协调者发送的 Cancel 响应没有及时被参与者接收到，那么参与者在等待超时以后执行了 commit 操做。这样就和其余接到 Cancel 命令并执行回滚的参与者之间存在数据不一致的状况。

2.2.3 SAGA协议实现分布式事务

2.2.3.1 SAGA协议介绍

Saga的组成：

每一个Saga由一系列sub-transaction Ti 组成
每一个Ti 都有对应的补偿动做Ci，补偿动做用于撤销Ti形成的结果

saga的执行顺序有两种：

T1, T2, T3, ..., Tn
T1, T2, ..., Tj, Cj,..., C2, C1，其中0 < j < n

Saga定义了两种恢复策略：

backward recovery，向后恢复，即上面提到的第二种执行顺序，其中j是发生错误的sub-transaction，这种作法的效果是撤销掉以前全部成功的sub-transation，使得整个Saga的执行结果撤销。
forward recovery，向前恢复，适用于必需要成功的场景，执行顺序是相似于这样的：T1, T2, ..., Tj(失败), Tj(重试),..., Tn，其中j是发生错误的sub-transaction。该状况下不须要Ci。

Saga的注意事项

Ti和Ci是幂等的。举个例子，假设在执行Ti的时候超时了，此时咱们是不知道执行结果的，若是采用forward recovery策略就会再次发送Ti，那么就有可能出现Ti被执行了两次，因此要求Ti幂等。若是采用backward recovery策略就会发送Ci，而若是Ci也超时了，就会尝试再次发送Ci，那么就有可能出现Ci被执行两次，因此要求Ci幂等。
Ci必须是可以成功的，若是没法成功则须要人工介入。若是Ci不能执行成功就意味着整个Saga没法彻底撤销，这个是不容许的。但总会出现一些特殊状况好比Ci的代码有bug、服务长时间崩溃等，这个时候就须要人工介入了
Ti - Ci和Ci - Ti的执行结果必须是同样的：sub-transaction被撤销了。举例说明，仍是考虑Ti执行超时的场景，咱们采用了backward recovery，发送一个Ci，那么就会有三种状况：
1：Ti的请求丢失了，服务以前没有、以后也不会执行Ti
2：Ti在Ci以前执行
3：Ci在Ti以前执行
对于第1种状况，容易处理。对于第二、3种状况，则要求Ti和Ci是可交换的（commutative)，而且其最终结果都是sub-transaction被撤销。

Saga架构

Saga Execution Component解析请求JSON并构建请求图
TaskRunner 用任务队列确保请求的执行顺序
TaskConsumer 处理Saga任务，将事件写入saga log，并将请求发送到远程服务

2.2.3.2 SAGA的ACID特性

原子性：经过SAGA协调器实现
一致性：本地事务+SAGA Log
持久性：SAGA Log
隔离性：不保证（同TCC）

3 分布式事务的处理方案

3.1 XA

仅在同一个事务上下文中须要协调多种资源（即数据库，以及消息主题或队列）时，才有必要使用 X/Open XA 接口。数据库接入XA须要使用XA版的数据库驱动，消息队列要实现XA须要实现javax.transaction.xa.XAResource接口。

3.1.1 jotm的分布式事务

代码以下：

public class UserService {
    @Autowired
    private UserDao userDao;
    @Autowired
    private LogDao logDao;
    @Transactional
    public void save(User user){
        userDao.save(user);
        logDao.save(user);
        throw new RuntimeException();
    }
}
@Resource
public class UserDao {
    @Resource(name="jdbcTemplateA")
    private JdbcTemplate jdbcTemplate;
    public void save(User user){
        jdbcTemplate.update("insert into user(name,age) values(?,?)",user.getName(),user.getAge());
    }
}
@Repository
public class LogDao {
    @Resource(name="jdbcTemplateB")
    private JdbcTemplate jdbcTemplate;
    public void save(User user){
        jdbcTemplate.update("insert into log(name,age) values(?,?)",user.getName(),user.getAge());
    }
}
复制代码

配置：

<bean id="jotm" class="org.objectweb.jotm.Current" />
    <bean id="transactionManager" class="org.springframework.transaction.jta.JtaTransactionManager">
        <property name="userTransaction" ref="jotm" />
    </bean>
    <tx:annotation-driven transaction-manager="transactionManager"/>
    <!-- 配置数据源 -->
    <bean id="dataSourceA" class="org.enhydra.jdbc.pool.StandardXAPoolDataSource" destroy-method="shutdown">
        <property name="dataSource">
            <bean class="org.enhydra.jdbc.standard.StandardXADataSource" destroy-method="shutdown">
                <property name="transactionManager" ref="jotm" />
                <property name="driverName" value="com.mysql.jdbc.Driver" />
                <property name="url" value="jdbc:mysql://localhost:3306/test?useUnicode=true&amp;characterEncoding=utf-8" />
            </bean>
        </property>
        <property name="user" value="xxx" />
        <property name="password" value="xxx" />
    </bean>
    <!-- 配置数据源 -->
    <bean id="dataSourceB" class="org.enhydra.jdbc.pool.StandardXAPoolDataSource" destroy-method="shutdown">
        <property name="dataSource">
            <bean class="org.enhydra.jdbc.standard.StandardXADataSource" destroy-method="shutdown">
                <property name="transactionManager" ref="jotm" />
                <property name="driverName" value="com.mysql.jdbc.Driver" />
                <property name="url" value="jdbc:mysql://localhost:3306/test2?useUnicode=true&amp;characterEncoding=utf-8" />
            </bean>
        </property>
        <property name="user" value="xxx" />
        <property name="password" value="xxx" />
    </bean>
    <bean id="jdbcTemplateA" class="org.springframework.jdbc.core.JdbcTemplate">
        <property name="dataSource" ref="dataSourceA" />
    </bean>
    <bean id="jdbcTemplateB" class="org.springframework.jdbc.core.JdbcTemplate">
        <property name="dataSource" ref="dataSourceB" />
    </bean>
复制代码

使用到的JAR包：

compile 'org.ow2.jotm:jotm-core:2.3.1-M1'
  compile 'org.ow2.jotm:jotm-datasource:2.3.1-M1'
  compile 'com.experlog:xapool:1.5.0'
复制代码

事务配置: 咱们知道分布式事务中须要一个事务管理器即接口javax.transaction.TransactionManager、面向开发人员的javax.transaction.UserTransaction。对于jotm来讲，他们的实现类都是Current

public class Current implements UserTransaction, TransactionManager

咱们若是想使用分布式事务的同时，又想使用Spring带给咱们的@Transactional便利，就须要配置一个JtaTransactionManager，而该JtaTransactionManager是须要一个userTransaction实例的，因此用到了上面的Current，以下配置:

<bean id="jotm" class="org.objectweb.jotm.Current" />
<bean id="transactionManager" class="org.springframework.transaction.jta.JtaTransactionManager">  
    <property name="userTransaction" ref="jotm" />  
</bean>
<tx:annotation-driven transaction-manager="transactionManager"/>
复制代码

同时上述StandardXADataSource是须要一个TransactionManager实例的，因此上述StandardXADataSource配置把jotm加了进去.

执行过程：

第一步：事务拦截器开启事务
咱们知道加入了@Transactional注解，同时开启tx:annotation-driven，会对本对象进行代理，加入事务拦截器。在事务拦截器中，获取javax.transaction.UserTransaction，这里即org.objectweb.jotm.Current，而后使用它开启事务，并和当前线程进行绑定，绑定关系数据存放在org.objectweb.jotm.Current中。
第二步：使用jdbcTemplate进行业务操做
dbcTemplateA要从dataSourceA中获取Connection,和当前线程进行绑定，同时以对应的dataSourceA做为key。同时判断当前线程是否含有事务，经过dataSourceA中的org.objectweb.jotm.Current发现当前线程有事务，则把Connection自动提交设置为false,同时将该链接归入当前事务中。
jdbcTemplateB要从dataSourceB中获取Connection,和当前线程进行绑定，同时以对应的dataSourceB做为key。同时判断当前线程是否含有事务，经过dataSourceB中的org.objectweb.jotm.Current发现当前线程有事务，则把Connection自动提交设置为false,同时将该链接归入当前事务中。
第三步：异常回滚一旦抛出异常，则须要进行事务的回滚操做。回滚就是将当前事务进行回滚，该事务的回滚会调用和它关联的全部Connection的回滚。

3.1.2 Atomikos的分布式事务

代码同上，配置为：

<bean id="atomikosUserTransaction" class="com.atomikos.icatch.jta.UserTransactionImp">
        <property name="transactionTimeout" value="300" />
    </bean>

    <bean id="springTransactionManager" class="org.springframework.transaction.jta.JtaTransactionManager">
        <property name="userTransaction" ref="atomikosUserTransaction" />
    </bean>

    <tx:annotation-driven transaction-manager="springTransactionManager"/>

    <!-- 配置数据源 -->
    <bean id="dataSourceC" class="com.atomikos.jdbc.AtomikosDataSourceBean" init-method="init" destroy-method="close">
        <property name="uniqueResourceName" value="XA1DBMS" />
        <property name="xaDataSourceClassName" value="com.mysql.jdbc.jdbc2.optional.MysqlXADataSource" />
        <property name="xaProperties">
            <props>
                <prop key="URL">jdbc:mysql://localhost:3306/test?useUnicode=true&amp;characterEncoding=utf-8</prop>
                <prop key="user">xxx</prop>
                <prop key="password">xxx</prop>
            </props>
        </property>
        <property name="poolSize" value="3" />
        <property name="minPoolSize" value="3" />
        <property name="maxPoolSize" value="5" />
    </bean>

    <!-- 配置数据源 -->
    <bean id="dataSourceD" class="com.atomikos.jdbc.AtomikosDataSourceBean" init-method="init" destroy-method="close">
        <property name="uniqueResourceName" value="XA2DBMS" />
        <property name="xaDataSourceClassName" value="com.mysql.jdbc.jdbc2.optional.MysqlXADataSource" />
        <property name="xaProperties">
            <props>
                <prop key="URL">jdbc:mysql://localhost:3306/test2?useUnicode=true&amp;characterEncoding=utf-8</prop>
                <prop key="user">xxx</prop>
                <prop key="password">xxx</prop>
            </props>
        </property>
        <property name="poolSize" value="3" />
        <property name="minPoolSize" value="3" />
        <property name="maxPoolSize" value="5" />
    </bean>

    <bean id="jdbcTemplateC" class="org.springframework.jdbc.core.JdbcTemplate">
        <property name="dataSource" ref="dataSourceC" />
    </bean>

    <bean id="jdbcTemplateD" class="org.springframework.jdbc.core.JdbcTemplate">
        <property name="dataSource" ref="dataSourceD" />
    </bean>
复制代码

事务配置：
咱们知道分布式事务中须要一个事务管理器即接口javax.transaction.TransactionManager、面向开发人员的javax.transaction.UserTransaction。对于Atomikos来讲分别对应以下：

com.atomikos.icatch.jta.UserTransactionImp
com.atomikos.icatch.jta.UserTransactionManager 咱们若是想使用分布式事务的同时，又想使用Spring带给咱们的@Transactional便利，就须要配置一个JtaTransactionManager，而该JtaTransactionManager是须要一个userTransaction实例的

<bean id="userTransaction" class="com.atomikos.icatch.jta.UserTransactionImp">  
    <property name="transactionTimeout" value="300" />  
</bean>
<bean id="springTransactionManager" class="org.springframework.transaction.jta.JtaTransactionManager">  
    <property name="userTransaction" ref="userTransaction" />   
</bean>
<tx:annotation-driven transaction-manager="springTransactionManager"/>
复制代码

能够对比下jotm的案例配置jotm的分布式事务配置。能够看到jotm中使用的xapool中的StandardXADataSource是须要一个transactionManager的，而Atomikos使用的AtomikosNonXADataSourceBean则不须要。咱们知道，StandardXADataSource中有了transactionManager就能够获取当前线程的事务，同时把XAResource加入进当前事务中去，而AtomikosNonXADataSourceBean却没有，它是怎么把XAResource加入进当前线程绑定的事务呢？这时候就须要能够经过静态方法随时获取当前线程绑定的事务。使用到的JAR包：

compile 'com.atomikos:transactions-jdbc:4.0.0M4'
复制代码

3.2 单机事务+同步回调（异步）

以订单子系统和支付子系统为例，以下图：

如上图，payment是支付系统，trade是订单系统，两个系统对应的数据库是分开的。支付完成以后，支付系统须要通知订单系统状态变动。
对于payment要执行的操做能够用伪代码表示以下：

begin tx;
  count = update account set amount = amount - ${cash} where uid = ${uid} and amount >= amount if (count <= 0) return false update payment_record set status = paid where trade_id = ${tradeId}
commit;
复制代码

对于trade要执行的操做能够用伪代码表示以下：

begin tx;
  count = update trade_record set status = paid where trade_id = ${trade_id} and status = unpaid if (count <= 0) return false do other things ... commit;
复制代码

可是对于这两段代码如何串起来是个问题，咱们增长一个事务表，即图中的tx_info，来记录成功完成的支付事务，tx_info中须要有能够标示被支付系统处理状态的字段，为了和支付信息一致，须要放入事务中，代码以下：

begin tx;
  count = update account set amount = amount - ${cash} where uid = ${uid} and amount >= amount if (count <= 0) return false update payment_record set status = paid where trade_id = ${tradeId}
    insert into tx_info values(${trade_id},${amount}...) commit;
复制代码

支付系统边界到此为止，接下来就是订单系统轮询访问tx_info，拉取已经支付成功的订单信息，对每一条信息都执行trade系统的逻辑，伪代码以下：

foreach trade_id in tx_info
  do trade_tx
  save tx_info.id to some store
复制代码

事无延迟取决于时间程序轮询间隔，这样咱们作到了一致性，最终订单都会在支付以后的最大时间间隔内完成状态迁移。
固然，这里也能够采用支付系统经过RPC方式同步通知订单系统的方式来实现，处理状态经过tx_info中的字段来表示。
另外，交易系统每次拉取数据的起点以及消费记录须要记录下来，这样才能不遗漏不重复地执行，因此须要增长一张表用于排重，即上图中的tx_duplication。可是每次对tx_duplication表的插入要在trade_tx的事务中完成，伪代码以下：

begin tx;
  c = insert ignore tx_duplication values($trade_id...) if (c <= 0) return false count = update trade_record set status = paid where trade_id = ${trade_id} and status = unpaid if (count <= 0) return false do other things ... commit;
复制代码

另外，tx_duplication表中trade_id表上必须有惟一键，这个算是结合以前的幂等篇来保证trade_tx的操做是幂等的。

3.3 MQ作中间表角色

在上面的方案中，tx_info表所起到的做用就是队列做用，记录一个系统的表更，做为通知给须要感知的系统的事件。而时间程序去拉取只是系统去获取感兴趣事件的一个方式，而对应交易系统的本地事务只是对应消费事件的一个过程。在这样的描述下，这些功能就是一个MQ——消息中间件。以下图

这样tx_info表的功能就交给了MQ，消息消费的偏移量也不须要关心了，MQ会搞定的，可是tx_duplication仍是必须存在的，由于MQ并不能避免消息的重复投递，这其中的缘由有不少，主要是仍是分布式的CAP形成的，再次不详细描述。
这要求MQ必须支持事务功能，能够达到本地事务和消息发出是一致性的，可是没必要是强一致的。一般使用的方式以下的伪代码：

sendPrepare();
  isCommit = local_tx()
  if (isCommit) sendCommit()
    else sendRollback()
复制代码

在作本地事务以前，先向MQ发送一个prepare消息，而后执行本地事务，本地事务提交成功的话，向MQ发送一个commit消息，不然发送一个abort消息，取消以前的消息。MQ只会在收到commit确认才会将消息投递出去，因此这样的形式能够保证在一切正常的状况下，本地事务和MQ能够达到一致性。
可是分布式存在异常状况，网络超时，机器宕机等等，好比当系统执行了local_tx()成功以后，还没来得及将commit消息发送给MQ，或者说发送出去了，网络超时了等等缘由，MQ没有收到commit，即commit消息丢失了，那么MQ就不会把prepare消息投递出去。若是这个没法保证的话，那么这个方案是不可行的。针对这种状况，须要一个第三方异常校验模块来对MQ中在必定时间段内没有commit/abort 的消息和发消息的系统进行检查，确认该消息是否应该投递出去或者丢弃，获得系统的确认以后，MQ会作投递仍是丢弃，这样就彻底保证了MQ和发消息的系统的一致性，从而保证了接收消息系统的一致性。
这个方案要求MQ的系统可用性必须很是高，至少要超过使用MQ的系统（推荐rocketmq，kafka都支持发送预备消息和业务回查），这样才能保证依赖他的系统能稳定运行。

3.4 SAGA方案

项目地址：github.com/apache/serv… Saga处理场景是要求相关的子事务提供事务处理函数同时也提供补偿函数。Saga协调器alpha会根据事务的执行状况向omega发送相关的指令，肯定是否向前重试或者向后恢复。

成功场景

成功场景下，每一个事务都会有开始和有对应的结束事件。

异常场景

异常场景下，omega会向alpha上报中断事件，而后alpha会向该全局事务的其它已完成的子事务发送补偿指令，确保最终全部的子事务要么都成功，要么都回滚。

超时场景

超时场景下，已超时的事件会被alpha的按期扫描器检测出来，与此同时，该超时事务对应的全局事务也会被中断。

例子

假设要租车、预订酒店知足分布式事务。
租车服务

@Service
class CarBookingService {
  private Map<Integer, CarBooking> bookings = new ConcurrentHashMap<>();

  @Compensable(compensationMethod = "cancel")
  void order(CarBooking booking) {
    booking.confirm();
    bookings.put(booking.getId(), booking);
  }

  void cancel(CarBooking booking) {
    Integer id = booking.getId();
    if (bookings.containsKey(id)) {
      bookings.get(id).cancel();
    }
  }

  Collection<CarBooking> getAllBookings() {
    return bookings.values();
  }

  void clearAllBookings() {
    bookings.clear();
  }
}
复制代码

酒店预订

@Service
class HotelBookingService {
  private Map<Integer, HotelBooking> bookings = new ConcurrentHashMap<>();

  @Compensable(compensationMethod = "cancel")
  void order(HotelBooking booking) {
    if (booking.getAmount() > 2) {
      throw new IllegalArgumentException("can not order the rooms large than two");
    }
    booking.confirm();
    bookings.put(booking.getId(), booking);
  }

  void cancel(HotelBooking booking) {
    Integer id = booking.getId();
    if (bookings.containsKey(id)) {
      bookings.get(id).cancel();
    }
  }

  Collection<HotelBooking> getAllBookings() {
    return bookings.values();
  }

  void clearAllBookings() {
    bookings.clear();
  }
}
复制代码

主服务

@RestController
public class BookingController {

  @Value("${car.service.address:http://car.servicecomb.io:8080}")
  private String carServiceUrl;

  @Value("${hotel.service.address:http://hotel.servicecomb.io:8080}")
  private String hotelServiceUrl;

  @Autowired
  private RestTemplate template;

  @SagaStart
  @PostMapping("/booking/{name}/{rooms}/{cars}")
  public String order(@PathVariable String name, @PathVariable Integer rooms, @PathVariable Integer cars) {
    template.postForEntity(
        carServiceUrl + "/order/{name}/{cars}",
        null, String.class, name, cars);

    postCarBooking();

    template.postForEntity(
        hotelServiceUrl + "/order/{name}/{rooms}",
        null, String.class, name, rooms);

    postBooking();

    return name + " booking " + rooms + " rooms and " + cars + " cars OK";
  }

  // This method is used by the byteman to inject exception here
  private void postCarBooking() {

  }

  // This method is used by the byteman to inject the faults such as the timeout or the crash
  private void postBooking() {

  }
}
复制代码

执行流程

在Alpha目录执行 mvn clean package -DskipTests -Pdemo
执行 java -Dspring.profiles.active=prd -D"spring.datasource.url=jdbc:postgresql:// ${host_address}:5432/saga?useSSL=false" -jar alpha-server-$ {saga_version}-exec.jar
在saga spring demo目录执行 mvn clean package -DskipTests -Pdemo
java -Dserver.port=8081 -Dalpha.cluster.address= ${alpha_address}:8080 -jar hotel-$ {saga_version}-exec.jar
java -Dserver.port=8082 -Dalpha.cluster.address= ${alpha_address}:8080 -jar car-$ {saga_version}-exec.jar
java -Dserver.port=8083 -Dalpha.cluster.address= ${alpha_address}:8080 -Dcar.service.address=$ {host_address}:8082 -Dhotel.service.address= ${host_address}:8081 -jar booking-$ {saga_version}-exec.jar[alpha_address不带http其余地址要带上http]

3.5 TCC方案

项目地址https://github.com/QNJR-GROUP/EasyTransaction[对比tcc-transaction，Hmily，ByteTCC来讲EasyTransaction性能最好，压测未发现错误], 固然你也可使用上面提到的SAGA项目，也是支持TCC协议的。下面咱们举个例子来看TCC是如何处理业务逻辑的。

eg：订单支付

1：订单服务->修改订单状态
2：库存服务->扣减库存
3：积分服务->增长积分
4：仓库服务->建立出库单

try阶段

1：订单服务->状态变动为“UpDating”
2：库存服务->可用库存减小1，冻结库存增长1
3：积分服务->积分不变，增长预备积分
4：仓库服务->建立出库单，状态设置为“UnKnown”

confirm阶段

1：订单服务->状态变动为“已支付”
2：库存服务->冻结库存清零
3：积分服务->积分增长，预备积分清零
4：仓库服务->状态设置为“出库单已建立”

cancel阶段

1：订单服务->状态变动为“已取消”
2：库存服务->可用库存增长，冻结库存清零
3：积分服务->预备积分清零
4：仓库服务->状态设置为“已取消”

4 小结

基本概念	优势	缺点
本地事务。事务由资源管理器（如DBMS）本地管理	严格的ACID	不具有分布事务处理能力
全局事务（DTP模型） TX协议：应用或应用服务器与事务管理器的接口 XA协议：全局事务管理器与资源管理器的接口	严格的ACID	效率很是低
JTA:面向应用、应用服务器与资源管理器的高层事务接口 JTS:JTA事务管理器的实现标准，向上支持JTA，向下经过CORBA OTS实现跨事务域的互操做性 EJB	简单一致的编程模型跨域分布处理的ACID保证	DTP模型自己的局限缺乏充分公开的大规模、高可用、密集事务应用的成功案例
基于MQ	消息数据独立存储、独立伸缩下降业务系统与消息系统间的耦合	一次消息发送须要两次请求业务处理服务需实现消息状态回查接口
二阶段提交	原理简单，实现方便	同步阻塞：在二阶段提交的过程当中，全部的节点都在等待其余节点的响应，没法进行其余操做。这种同步阻塞极大的限制了分布式系统的性能。单点问题：协调者在整个二阶段提交过程当中很重要，若是协调者在提交阶段出现问题，那么整个流程将没法运转。更重要的是，其余参与者将会处于一直锁定事务资源的状态中，而没法继续完成事务操做。数据不一致：假设当协调者向全部的参与者发送commit请求以后，发生了局部网络异常，或者是协调者在还没有发送完全部 commit请求以前自身发生了崩溃，致使最终只有部分参与者收到了commit请求。这将致使严重的数据不一致问题。容错性很差：若是在二阶段提交的提交询问阶段中，参与者出现故障，致使协调者始终没法获取到全部参与者的确认信息，这时协调者只能依靠其自身的超时机制，判断是否须要中断事务。显然，这种策略过于保守。换句话说，二阶段提交协议没有设计较为完善的容错机制，任意一个节点是失败都会致使整个事务的失败。
TCC	相对于二阶段提交，三阶段提交主要解决的单点故障问题，并减小了阻塞的时间。由于一旦参与者没法及时收到来自协调者的信息以后，他会默认执行 commit。而不会一直持有事务资源并处于阻塞状态。	三阶段提交也会致使数据一致性问题。因为网络缘由，协调者发送的 abort 响应没有及时被参与者接收到，那么参与者在等待超时以后执行了 commit 操做。这样就和其余接到 abort 命令并执行回滚的参与者之间存在数据不一致的状况。
SAGA	简单业务使用TCC须要修改原来业务逻辑，saga只须要添加一个补偿动做因为没有预留动做因此不用担忧资源释放的问题异常处理简单	因为没有预留动做致使补偿处理麻烦

业务各有各的不一样，有些业务能容忍短时间不一致，有些业务的操做能够幂等，不管什么样的分布式事务解决方案都有其优缺点，没有一个银弹可以适配全部。所以，业务须要什么样的解决方案，还须要结合自身的业务需求、业务特色、技术架构以及各解决方案的特性，综合分析，才能找到最适合的方案。