MySQL分布式事务 – 运维那点事

一、分布式事务

在说分布式事务（XA）之前，可以先看一下“以交易系统为例，看分布式事务架构的五大演进”，阐述了分布式事务解决了什么问题？

InnoDB存储引擎提供了对XA事务的支持，并通过XA事务来支持分布式事务的实现。分布式事务指的是允许多个独立的事务资源参与到一个全局的事务中。事务资源通常是关系型数据库系统，但也可以是其他类型的资源。全局事务要求在其中的所有参与的事务要么都提交，要么都回滚，这对于事务原有的ACID要求又有了提高。另外，在使用分布式事务时，InnoDB存储引擎的事务隔离级别必须设置为SERIALIZABLE。

XA事务语允许不同数据库之间的分布式事务，如一台服务器是MySQL数据库的，另一台是Oracle数据库的，又可能还有一台服务器是SQL Server数据库的，只要参与在全局事务中的每个节点都支持XA事务。

考虑下面一种场景：当你发了工资之后，把你的当月工资¥10000从支付宝转到了余额宝。如果在支付宝账户扣除¥10000之后，余额宝系统挂掉了，余额宝的账户并没有增加¥10000，这时候就出现了数据不一致的情况。

# 支付宝
update account set money = money - 10000 where user='dkey';

# 余额宝
update account set money = money + 10000 where user='dkey';

# 支付宝

update account set money = money - 10000 where user='dkey';

# 余额宝

update account set money = money + 10000 where user='dkey';

在很多系统中都能找到上述情况的影子：

在下单的时候，需要在订单表中插入一条数据，然后把库存减去一。
在搜索的时候如果点击了广告，需要先记录该点击事件，然后通知商家系统扣除广告费。

在这种情况下，一定需要使用分布式事务来保证数据的安全。如果发生的操作不能全部提交或回滚，那么任何一个节点出现问题都会导致严重的结果。

在一个分布式事务结束的时候，事务的原子特性要求所有参与该事务的服务器必须全部提交或全部放弃该事务。为了实现这一点，其中一个服务器承担了协调者（coordinater）的角色，由它来保证所有的服务器获得相同的结果。

协调者（coordinater）的工作方式取决于它选用的协议，“两阶段提交”是分布式事务最常用的协议。

两阶段提交协议（Two-phase Commit，2PC）经常被用来实现分布式事务。一般由一个或多个资源管理器（resource managers）、一个事务协调器（transaction coordinater）以及一个应用程序（application program）组成。事务协调器可以和资源管理器在一台机器上。

资源管理器：提供访问事务资源的方法，通常一个数据库就是一个资源管理器。
事务协调器：协调参与全局事务中的各个事务，需要和参与全局事务的所有资源管理器进行通信。
应用程序：定义事务的边界，指定全局事务中的操作。

在MySQL数据库的分布式事务中，资源管理器就是MySQL数据库，事务协调器为连接MySQL服务器的客户端（支持分布式事务的客户端）。下图显示了一个分布式事务的模型。

分布式事务通常采用2PC协议，全称Two Phase Commitment Protocol。该协议主要为了解决在分布式数据库场景下，所有节点间数据一致性的问题。在分布式事务环境下，事务的提交会变得相对比较复杂，因为多个节点的存在，可能存在部分节点提交失败的情况，即事务的ACID特性需要在各个数据库实例中保证。总而言之，在分布式提交时，只要发生一个节点提交失败，则所有的节点都不能提交，只有当所有节点都能提交时，整个分布式事务才允许被提交。

在该协议的第一个阶段，每个参与者投票表决该事务是放弃还是提交，一旦参与者要求提交事务，那么就不允许放弃该事务。因此，在一个参与者要求提交事务之前，它必须保证最终能够执行分布式事务中自己的那部分，即使该参与者出现故障而被中途替换掉。

一个事务的参与者如果最终能提交事务，那么可以说参与者处于事务的准备好（prepared）状态。为了保证能够提交，每个参与者必须将事务中所有发生改变的对象以及自身的状态（prepared）保存到持久性存储中。

在该协议的第二个阶段，事务的每个参与者执行最终统一的决定。如果任何一个参与者投票放弃事务，那么最终的决定是放弃事务，则所有的节点都被告知需要回滚。如果所有的参与者都投票提交事务，那么最终的决定是提交事务。

1. 我们的应用程序（client）发起一个开始请求到TC；

2. TC先将<prepare>消息写到本地日志，之后向所有的RM发起<prepare>消息。以支付宝转账到余额宝为例，TC给A的prepare消息是通知支付宝数据库相应账目扣款1万，TC给B的prepare消息是通知余额宝数据库相应账目增加1w。为什么在执行任务前需要先写本地日志，主要是为了故障后恢复用，本地日志起到现实生活中凭证的效果，如果没有本地日志（凭证），出问题容易死无对证；

3. RM收到<prepare>消息后，执行具体本机事务，但不会进行commit，如果成功返回<yes>，不成功返回<no>。同理，返回前都应把要返回的消息写到日志里，当作凭证。

4. TC收集所有执行器返回的消息，如果所有执行器都返回yes，那么给所有执行器发生送commit消息，执行器收到commit后执行本地事务的commit操作；如果有任一个执行器返回no，那么给所有执行器发送rollback消息，执行器收到rollback消息后执行事务rollback操作。

注：TC或RM把发送或接收到的消息先写到日志里，主要是为了故障后恢复用。如某一RM从故障中恢复后，先检查本机的日志，如果已收到<commit >，则提交，如果<rollback>则回滚。如果是<yes>，则再向TC询问一下，确定下一步。如果什么都没有，则很可能在<prepare>阶段RM就崩溃了，因此需要回滚。

可见与本地事务不同的是，分布式事务需要多一次的PREPARE操作，待收到所有节点的同意信息后，再进行COMMIT或是ROLLBACK操作。

现如今实现基于两阶段提交的分布式事务也没那么困难了，如果使用java，那么可以使用开源软件atomikos(http://www.atomikos.com/)来快速实现。

不过但凡使用过的上述两阶段提交的同学都可以发现性能实在是太差，根本不适合高并发的系统。为什么？

两阶段提交涉及多次节点间的网络通信，通信时间太长！
事务时间相对于变长了，锁定的资源的时间也变长了，造成资源等待时间也增加好多！

正是由于分布式事务存在很严重的性能问题，大部分高并发服务都在避免使用，往往通过其他途径来解决数据一致性问题。比如使用消息队列来避免分布式事务。

二、MySQL分布式事务操作

XA事务语法

# 在mysql实例中开启一个XA事务,指定一个全局唯一标识；
mysql> XA START 'any_unique_id';

# XA事务的操作结束；
mysql> XA END 'any_unique_id';

# 告知mysql准备提交这个xa事务；
mysql> XA PREPARE 'any_unique_id';

# 告知mysql提交这个xa事务；
mysql> XA COMMIT 'any_unique_id';

# 告知mysql回滚这个xa事务；
mysql> XA ROLLBACK 'any_unique_id';

# 查看本机mysql目前有哪些xa事务处于prepare状态；
mysql> XA RECOVER;

# 在mysql实例中开启一个XA事务,指定一个全局唯一标识；

mysql> XA START 'any_unique_id';

# XA事务的操作结束；

mysql> XA END 'any_unique_id';

# 告知mysql准备提交这个xa事务；

mysql> XA PREPARE 'any_unique_id';

# 告知mysql提交这个xa事务；

mysql> XA COMMIT 'any_unique_id';

# 告知mysql回滚这个xa事务；

mysql> XA ROLLBACK 'any_unique_id';

# 查看本机mysql目前有哪些xa事务处于prepare状态；

mysql> XA RECOVER;

XA事务演示

在单个节点上运行分布式事务是没有意义的，起码两个节点才有意义。但是要在MySQL数据库的命令行下演示多个节点参与的分布式事务也是行不通的。通常来说，都是通过编程语言来完成分布式事务操作的。当前Java的JTA可以很好地支持MySQL的分布式事务。下面用一个简单的例子来演示：

public class XaDemo {
    public static MysqlXADataSource getDataSource(String connStr, String user, String pwd) {
        try {

            MysqlXADataSource ds = new MysqlXADataSource();
            ds.setUrl(connStr);
            ds.setUser(user);
            ds.setPassword(pwd);

            return ds;
        } catch (Exception e) {
            e.printStackTrace();
        }

        return null;
    }

    public static void main(String[] arg) {
        String connStr1 = "jdbc:mysql://192.168.0.1:3306/test";
        String connStr2 = "jdbc:mysql://192.168.0.2:3306/test";

        try {
            //从不同数据库获取数据库数据源
            MysqlXADataSource ds1 = getDataSource(connStr1, "root", "123456");
            MysqlXADataSource ds2 = getDataSource(connStr2, "root", "123456");

            //数据库1获取连接
            XAConnection xaConnection1 = ds1.getXAConnection();
            XAResource xaResource1 = xaConnection1.getXAResource();
            Connection connection1 = xaConnection1.getConnection();
            Statement statement1 = connection1.createStatement();
            
            //数据库2获取连接
            XAConnection xaConnection2 = ds2.getXAConnection();
            XAResource xaResource2 = xaConnection2.getXAResource();
            Connection connection2 = xaConnection2.getConnection();
            Statement statement2 = connection2.createStatement();

            //创建事务分支的xid
            Xid xid1 = new MysqlXid(new byte[] { 0x01 }, new byte[] { 0x02 }, 100);
            Xid xid2 = new MysqlXid(new byte[] { 0x011 }, new byte[] { 0x012 }, 100);

            try {
                //事务分支1关联分支事务sql语句
                xaResource1.start(xid1, XAResource.TMNOFLAGS);
                int update1Result = statement1.executeUpdate("update account_from set money=money - 50 where id=1");
                xaResource1.end(xid1, XAResource.TMSUCCESS);

                //事务分支2关联分支事务sql语句
                xaResource2.start(xid2, XAResource.TMNOFLAGS);
                int update2Result = statement2.executeUpdate("update account_to set money= money + 50 where id=1");
                xaResource2.end(xid2, XAResource.TMSUCCESS);
                
                // 两阶段提交协议第一阶段
                int ret1 = xaResource1.prepare(xid1);
                int ret2 = xaResource2.prepare(xid2);

                // 两阶段提交协议第二阶段
                if (XAResource.XA_OK == ret1 && XAResource.XA_OK == ret2) {
                    xaResource1.commit(xid1, false);
                    xaResource2.commit(xid2, false);

                    System.out.println("reslut1:" + update1Result + ", result2:" + update2Result);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

public class XaDemo {

public static MysqlXADataSource getDataSource(String connStr, String user, String pwd) {

try {

MysqlXADataSource ds = new MysqlXADataSource();

ds.setUrl(connStr);

ds.setUser(user);

ds.setPassword(pwd);

return ds;

} catch (Exception e) {

e.printStackTrace();

}

return null;

}

public static void main(String[] arg) {

String connStr1 = "jdbc:mysql://192.168.0.1:3306/test";

String connStr2 = "jdbc:mysql://192.168.0.2:3306/test";

try {

//从不同数据库获取数据库数据源

MysqlXADataSource ds1 = getDataSource(connStr1, "root", "123456");

MysqlXADataSource ds2 = getDataSource(connStr2, "root", "123456");

//数据库1获取连接

XAConnection xaConnection1 = ds1.getXAConnection();

XAResource xaResource1 = xaConnection1.getXAResource();

Connection connection1 = xaConnection1.getConnection();

Statement statement1 = connection1.createStatement();

//数据库2获取连接

XAConnection xaConnection2 = ds2.getXAConnection();

XAResource xaResource2 = xaConnection2.getXAResource();

Connection connection2 = xaConnection2.getConnection();

Statement statement2 = connection2.createStatement();

//创建事务分支的xid

Xid xid1 = new MysqlXid(new byte[] { 0x01 }, new byte[] { 0x02 }, 100);

Xid xid2 = new MysqlXid(new byte[] { 0x011 }, new byte[] { 0x012 }, 100);

try {

//事务分支1关联分支事务sql语句

xaResource1.start(xid1, XAResource.TMNOFLAGS);

int update1Result = statement1.executeUpdate("update account_from set money=money - 50 where id=1");

xaResource1.end(xid1, XAResource.TMSUCCESS);

//事务分支2关联分支事务sql语句

xaResource2.start(xid2, XAResource.TMNOFLAGS);

int update2Result = statement2.executeUpdate("update account_to set money= money + 50 where id=1");

xaResource2.end(xid2, XAResource.TMSUCCESS);

// 两阶段提交协议第一阶段

int ret1 = xaResource1.prepare(xid1);

int ret2 = xaResource2.prepare(xid2);

// 两阶段提交协议第二阶段

if (XAResource.XA_OK == ret1 && XAResource.XA_OK == ret2) {

xaResource1.commit(xid1, false);

xaResource2.commit(xid2, false);

System.out.println("reslut1:" + update1Result + ", result2:" + update2Result);

}

} catch (Exception e) {

e.printStackTrace();

}

} catch (Exception e) {

e.printStackTrace();

}

XA事务恢复

如果执行分布式事务的某个mysql crash了，MySQL按照如下逻辑进行恢复：

a. 如果这个xa事务commit了，那么什么也不用做。

b. 如果这个xa事务还没有prepare，那么直接回滚它。

c. 如果这个xa事务prepare了，还没commit，那么把它恢复到prepare的状态，由用户去决定commit或rollback。

当mysql crash后重新启动之后，执行“XA RECOVER；”查看当前处于prepare状态的xa事务，然后commit或rollback它们即可。如果不去处理，那么它们占用的资源就一直不会释放，比如锁。

三、MySQL分布式事务限制

a. XA事务和本地事务以及锁表操作是互斥的

开启了xa事务就无法使用本地事务和锁表操作

mysql> xa start 't1xa';
Query OK, 0 rows affected (0.04 sec)

mysql> begin;
ERROR 1399 (XAE07): XAER_RMFAIL: The command cannot be executed when global transaction is in the  ACTIVE state

mysql> lock table t read;
ERROR 1399 (XAE07): XAER_RMFAIL: The command cannot be executed when global transaction is in the  ACTIVE state

mysql> xa start 't1xa';

Query OK, 0 rows affected (0.04 sec)

mysql> begin;

ERROR 1399 (XAE07): XAER_RMFAIL: The command cannot be executed when global transaction is in the ACTIVE state

mysql> lock table t read;

ERROR 1399 (XAE07): XAER_RMFAIL: The command cannot be executed when global transaction is in the ACTIVE state

开启了本地事务就无法使用xa事务

mysql> begin;
Query OK, 0 rows affected (0.00 sec)

mysql> xa start 'rrrr';
ERROR 1400 (XAE09): XAER_OUTSIDE: Some work is done outside global transaction

mysql> begin;

Query OK, 0 rows affected (0.00 sec)

mysql> xa start 'rrrr';

ERROR 1400 (XAE09): XAER_OUTSIDE: Some work is done outside global transaction

b. xa start之后必须xa end，否则不能执行xa commit和xa rollback

所以如果在执行xa事务过程中有语句出错了，你也需要先xa end一下，然后才能xa rollback。

mysql> xa start 'tt';
Query OK, 0 rows affected (0.00 sec)

mysql> xa rollback 'tt';
ERROR 1399 (XAE07): XAER_RMFAIL: The command cannot be executed when global transaction is in the  ACTIVE state

mysql> xa end 'tt';
Query OK, 0 rows affected (0.00 sec)

mysql> xa rollback 'tt';
Query OK, 0 rows affected (0.00 sec)

mysql> xa start 'tt';

Query OK, 0 rows affected (0.00 sec)

mysql> xa rollback 'tt';

ERROR 1399 (XAE07): XAER_RMFAIL: The command cannot be executed when global transaction is in the ACTIVE state

mysql> xa end 'tt';

Query OK, 0 rows affected (0.00 sec)

mysql> xa rollback 'tt';

Query OK, 0 rows affected (0.00 sec)

四、MySQL 5.7对分布式事务的支持

一直以来，MySQL数据库是支持分布式事务的，但是只能说是有限的支持，具体表现在：

已经prepare的事务，在客户端退出或者服务宕机的时候，2PC的事务会被回滚。
在服务器故障重启提交后，相应的Binlog被丢失。

上述问题存在于MySQL数据库长达数十年的时间，直到MySQL-5.7.7版本，官方才修复了该问题。下面将会详细介绍下该问题的具体表现和官方修复方法，这里分别采用官方MySQL-5.6.27版本(未修复)和MySQL-5.7.9版本(已修复)进行验证。

先来看下存在的问题，我们先创建一个表如下：

CREATE TABLE t(
  id INT AUTO_INCREMENT PRIMARY KEY,
  a INT
) ENGINE=InnoDB;

CREATE TABLE t(

id INT AUTO_INCREMENT PRIMARY KEY,

a INT

) ENGINE=InnoDB;

对于上述表，通过如下操作进行数据插入：

mysql> XA START 'mysql56';
Query OK, 0 rows affected (0.00 sec)

mysql> INSERT INTO t VALUES(1,1);
Query OK, 1 row affected (0.02 sec)

mysql> XA END 'mysql56';
Query OK, 0 rows affected (0.00 sec)

mysql> XA PREPARE 'mysql56';
Query OK, 0 rows affected (0.00 sec)

mysql> XA START 'mysql56';

Query OK, 0 rows affected (0.00 sec)

mysql> INSERT INTO t VALUES(1,1);

Query OK, 1 row affected (0.02 sec)

mysql> XA END 'mysql56';

Query OK, 0 rows affected (0.00 sec)

mysql> XA PREPARE 'mysql56';

Query OK, 0 rows affected (0.00 sec)

通过上面的操作，用户创建了一个分布式事务，并且prepare没有返回错误，说明该分布式事务可以被提交。通过命令XA RECOVER查看显示如下结果：

mysql> XA RECOVER;
+----------+--------------+--------------+---------+
| formatID | gtrid_length | bqual_length | data    |
+----------+--------------+--------------+---------+
| 1        | 7            | 0            | mysql56 |
+----------+--------------+--------------+---------+

mysql> XA RECOVER;

+----------+--------------+--------------+---------+

+----------+--------------+--------------+---------+

| 1 | 7 | 0 | mysql56 |

+----------+--------------+--------------+---------+

若这时候用户退出客户端后重连，通过命令xa recover会发现刚才创建的2PC事务不见了。即prepare成功的事务丢失了，不符合2PC协议规范！！！

产生上述问题的主要原因在于：MySQL 5.6版本在客户端退出的时候，自动把已经prepare的事务回滚了，那么MySQL为什么要这样做？这主要取决于MySQL的内部实现，MySQL 5.7以前的版本，对于prepare的事务，MySQL是不会记录binlog的（官方说是减少fsync，起到了优化的作用）。只有当分布式事务提交的时候才会把前面的操作写入binlog信息，所以对于binlog来说，分布式事务与普通的事务没有区别，而prepare以前的操作信息都保存在连接的IO_CACHE中，如果这个时候客户端退出了，以前的binlog信息都会被丢失，再次重连后允许提交的话，会造成Binlog丢失，从而造成主从数据的不一致，所以官方在客户端退出的时候直接把已经prepare的事务都回滚了！

官方的做法，貌似干得很漂亮，牺牲了一点标准化的东西，至少保证了主从数据的一致性。但其实不然，若用户已经prepare后在客户端退出之前，MySQL发生了宕机，这个时候又会怎样？

MySQL在某个分布式事务prepare成功后宕机，宕机前操作该事务的连接并没有断开，这个时候已经prepare的事务并不会被回滚，所以在MySQL重新启动后，引擎层通过recover机制能恢复该事务。当然该事务的Binlog已经在宕机过程中被丢失，这个时候，如果去提交，则会造成主从数据的不一致，即提交没有记录Binlog，从上丢失该条数据。所以对于这种情况，官方一般建议直接回滚已经prepare的事务。

以上是MySQL 5.7以前版本MySQL在分布式事务上的各种问题，那么MySQL 5.7版本官方做了哪些改进？这个可以从官方的WL#6860描述上得到一些信息，我们还是本着没有实践就没有发言权的态度，从具体的操作上来分析下MySQL 5.7的改进方法。还是以上面同样的表结构进行同样的操作如下：

mysql> XA START 'mysql57';
Query OK, 0 rows affected (0.00 sec)

mysql> INSERT INTO t VALUES(1,1);
Query OK, 1 row affected (0.02 sec)

mysql> XA END 'mysql57';
Query OK, 0 rows affected (0.00 sec)

mysql> XA PREPARE 'mysql57';
Query OK, 0 rows affected (0.00 sec)

mysql> XA START 'mysql57';

Query OK, 0 rows affected (0.00 sec)

mysql> INSERT INTO t VALUES(1,1);

Query OK, 1 row affected (0.02 sec)

mysql> XA END 'mysql57';

Query OK, 0 rows affected (0.00 sec)

mysql> XA PREPARE 'mysql57';

Query OK, 0 rows affected (0.00 sec)

这个时候，我们通过mysqlbinlog来查看下Master上的Binlog，结果如下：

同时也对比下Slave上的Relay log，如下：

通过上面的操作，明显发现在prepare以后，从XA START到XA PREPARE之间的操作都被记录到了Master的Binlog中，然后通过复制关系传到了Slave上。也就是说MySQL 5.7开始，MySQL对于分布式事务，在prepare的时候就完成了写Binlog的操作，通过新增一种叫XA_prepare_log_event的event类型来实现，这是与以前版本的主要区别(以前版本prepare时不写Binlog)。

当然仅靠这一点是不够的，因为我们知道Slave通过SQL thread来回放Relay log信息，由于prepare的事务能阻塞整个session，而回放的SQL thread只有一个(不考虑并行回放)，那么SQL thread会不会因为被分布式事务的prepare阶段所阻塞，从而造成整个SQL thread回放出现问题？这也正是官方要解决的第二个问题：怎么样能使SQL thread在回放到分布式事务的prepare阶段时，不阻塞后面event的回放？其实这个实现也很简单(在xa.cc::applier_reset_xa_trans)，只要在SQL thread回放到prepare的时候，进行类似于客户端断开连接的处理即可(把相关cache与SQL thread的连接句柄脱离)。最后在Slave服务器上，用户通过命令XA RECOVER可以查到如下信息：

mysql> XA RECOVER;
+----------+--------------+--------------+---------+
| formatID | gtrid_length | bqual_length | data    |
+----------+--------------+--------------+---------+
| 1        | 7            | 0            | mysql57 |
+----------+--------------+--------------+---------+