JAVA分布式事務的解決方法

背景

在傳統(tǒng)架構(gòu)中可以使用spring的@Transactional 進行聲明式或者編程式的事務管理，但如果我們代碼中涉及到多數(shù)據(jù)源操作，就會發(fā)現(xiàn)spring的@Transactional事務管理機制會失靈，這種情況下我們就可以考慮使用兩階段提交的解決方案。
我們以mysql為例，mysql在5.0版本后支持了XA規(guī)范，也就是支持2PC形式的分布式事務。

分布式事務

分布式事務用于在分布式系統(tǒng)中保證不同節(jié)點之間的數(shù)據(jù)一致性。分布式事務的實現(xiàn)有很多種，最具有代表性的是由Oracle Tuxedo系統(tǒng)提出的XA分布式事務協(xié)議。

mysql XA

相關sql語句

XA start 'global_id','branch_id';
update user set age=22 where id=12;
update order set amount=1000.01 where id=1234;
XA end  'global_id','branch_id';

XA prepare   'global_id','branch_id';
XA RECOVER;  -- 查看當前所有處于準備狀態(tài)的XA事務
XA commit;-- 真正提交事務
XA rollback;-- 回滾事務

Java 代碼

使用druid管理連接池,其支持XA
import com.alibaba.druid.pool.xa.DruidXADataSource;
import com.mysql.jdbc.jdbc2.optional.MysqlXADataSource;

import com.alibaba.druid.pool.xa.DruidXADataSource;import com.mysql.jdbc.jdbc2.optional.MysqlXid;import javax.sql.XAConnection;import javax.transaction.xa.XAResource;import javax.transaction.xa.Xid;import java.sql.Connection;import java.sql.Statement;import java.util.Properties;/**
 * @author Jam Fang  https://www.jianshu.com/u/0977ede560d4
 * @version 創(chuàng)建時間：2019/4/14 13:58
 */public class TwoPhaseCommitApplication {    public void multiDataSourceTest() throws Exception {        String propertyfile = "/app.properties";        Properties props = new Properties();
        props.load(getClass().getResourceAsStream(propertyfile));        //初始化數(shù)據(jù)源        DruidXADataSource xaDataSource_1 = initXADataSource(props, "db1.");        //初始化XA連接        XAConnection xaConnection_1 = xaDataSource_1.getXAConnection();        //初始化XA資源        XAResource xaResource_1 = xaConnection_1.getXAResource();        //獲得數(shù)據(jù)庫連接        Connection connection_1 = xaConnection_1.getConnection();
        connection_1.setAutoCommit(false);        //創(chuàng)建XID        Xid xid_1 = new MysqlXid("globalid".getBytes(), "branch-1".getBytes(), 0);        //關聯(lián)事務start end
        xaResource_1.start(xid_1, XAResource.TMNOFLAGS);        Statement stmt = connection_1.createStatement();        String sql_1 = "INSERT INTO `order`(orderid,amount,product) values('00001','3000.00','蘋果筆記本');";//"delete from test3 where pk_t=3;";
        stmt.executeUpdate(sql_1);
        xaResource_1.end(xid_1, XAResource.TMSUCCESS);        //事務準備        int result_1 = xaResource_1.prepare(xid_1);        DruidXADataSource xaDataSource_2 = initXADataSource(props, "db2.");        XAConnection xaConnection_2 = xaDataSource_2.getXAConnection();        XAResource xaResource_2 = xaConnection_2.getXAResource();        Connection connection_2 = xaConnection_2.getConnection();
        connection_2.setAutoCommit(false);        Xid xid_2 = new MysqlXid("globalid".getBytes(), "branch-2".getBytes(), 0);
        xaResource_2.start(xid_2, XAResource.TMNOFLAGS);        Statement stmt2 = connection_2.createStatement();        String sql_2 = "update shipping set address='北京黃浦江畔' where id=1;";
        stmt2.executeUpdate(sql_2);
        xaResource_2.end(xid_2, XAResource.TMSUCCESS);        int result_2 = xaResource_2.prepare(xid_2);        //XA事務 準備階段        if (result_1 == XAResource.XA_OK &&
                result_2 == XAResource.XA_OK) {            //都返回OK的話，進行提交階段
            xaResource_1.commit(xid_1, false);
            xaResource_2.commit(xid_2, false);        } else {            //回滾事務
            xaResource_1.rollback(xid_1);
            xaResource_2.rollback(xid_2);        }    }    DruidXADataSource initXADataSource(Properties props, String prefix) {        DruidXADataSource xaDataSource = new DruidXADataSource();
        xaDataSource.setDbType(props.getProperty(prefix + "dbtype"));
        xaDataSource.setUrl(props.getProperty(prefix + "url"));
        xaDataSource.setUsername(props.getProperty(prefix + "username"));
        xaDataSource.setPassword(props.getProperty(prefix + "password"));        return xaDataSource;    }    public static void main(String args[]) {        try {            new TwoPhaseCommitApplication().multiDataSourceTest();        } catch (Exception e) {
            e.printStackTrace();        }    }}

app.properties文件

db1.dbtype=mysql
db1.url=jdbc:mysql://127.0.0.1:3306/archdemo1?useUnicode=true&characterEncoding=utf8&useSSL=false&serverTimezone=Asia/Shanghai
db1.username=root
db1.password=123456
db2.dbtype=mysql

db2.url=jdbc:mysql://127.0.0.1:3306/archdemo2?useUnicode=true&characterEncoding=utf8&useSSL=false&serverTimezone=Asia/Shanghai
db2.username=root
db2.password=123456

分析

在這種方案下,我們的代碼充當了TM也就是事務資源協(xié)調(diào)者,而兩個不同的數(shù)據(jù)源mysql充當了RM資源管理著角色,在我們代碼中對每個事務的準備情況進行判斷,如果都OK則提交事務,如果有沒有準備好的則rollback事務.

修改為一個不能正常執(zhí)行的sql,來查看他的執(zhí)行過程

image.png

通過斷點分析,我們發(fā)現(xiàn)程序在執(zhí)行到這句的時候就會出現(xiàn)異常,也就是在prepare之前sql語句已經(jīng)在執(zhí)行了,只不過我們設置了事務不自動提交,所以在數(shù)據(jù)庫中看不到sql_1的執(zhí)行結(jié)果.

修改正常sql,在prepare階段加斷點

image.png

我們在數(shù)據(jù)庫中查看事務情況,因為我是在一個數(shù)據(jù)庫服務器上做的的跨庫數(shù)據(jù)源,所以我們能看到兩條xa記錄

image.png

我們繼續(xù)執(zhí)行到commit語句,放多第一條commit

image.png

這時候可以發(fā)現(xiàn)已經(jīng)第一個事務的xa信息已經(jīng)沒有了,也就是第一個事務分支已經(jīng)提交成功了.

image.png

數(shù)據(jù)庫中可以看到新插入成功了一條數(shù)據(jù)

image.png

這是我們嘗試修改結(jié)構(gòu)或者插入一條語句,都會發(fā)現(xiàn)數(shù)據(jù)庫處于鎖定狀態(tài)

image.png

等我們把斷點放行之后,才可以看到其他語句正常執(zhí)行,也就是說xa在提交階段會對數(shù)據(jù)庫進行加鎖處理,經(jīng)過進一步的分析我們發(fā)現(xiàn)xa在進入xa end后就對整個表進行加鎖操作,因為該sql是update語句,所以在xa end 一直到事務提交或者回滾之前,整個表都處于鎖定狀態(tài).

image.png

延伸

我們很容易將這種XA機制擴展到到微服務情況,需要各個微服務提供相應的機制,各個微服務提供對應的prepare接口、commit接口、rollback接口。

缺點

xA的性能問題

XA的性能很低。一個數(shù)據(jù)庫的事務和多個數(shù)據(jù)庫間的XA事務性能對比可發(fā)現(xiàn)，性能差10倍左右。因此要盡量避免XA事務，例如可以將數(shù)據(jù)寫入本地，用高性能的消息系統(tǒng)分發(fā)數(shù)據(jù)?；蚴褂脭?shù)據(jù)庫復制等技術。只有在這些都無法實現(xiàn)，且性能不是瓶頸時才應該使用XA

這種機制假定prepare ok的事務都可以正常commit

也就是進入prepare返回ok后，在執(zhí)行commit階段兩個事務就有可能出現(xiàn)一些異常情況，比如第一個正常提交了，但第二個卻出現(xiàn)了某種異常失敗了。