建站
咨询
售后
建站咨询
背景
在传统架构中可以使用spring的@Transactional 进行声明式或者编程式的事务管理，但如果我们代码中涉及到多数据源操作，就会发现spring的@Transactional事务管理机制会失灵，这种情况下我们就可以考虑使用两阶段提交的解决方案。
我们以MySQL为例，mysql在5.0版本后支持了XA规范，也就是支持2PC形式的分布式事务。
成都创新互联专注骨干网络服务器租用10多年，服务更有保障！服务器租用，成都温江机房 成都服务器租用，成都服务器托管，骨干网络带宽，享受低延迟，高速访问。灵活、实现低成本的共享或公网数据中心高速带宽的专属高性能服务器。
分布式事务
分布式事务用于在分布式系统中保证不同节点之间的数据一致性。分布式事务的实现有很多种，最具有代表性的是由Oracle Tuxedo系统提出的XA分布式事务协议。
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; -- 查看当前所有处于准备状态的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 创建时间：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)); //初始化数据源 DruidXADataSource xaDataSource_1 = initXADataSource(props, "db1."); //初始化XA连接 XAConnection xaConnection_1 = xaDataSource_1.getXAConnection(); //初始化XA资源 XAResource xaResource_1 = xaConnection_1.getXAResource(); //获得数据库连接 Connection connection_1 = xaConnection_1.getConnection();
connection_1.setAutoCommit(false); //创建XID Xid xid_1 = new MysqlXid("globalid".getBytes(), "branch-1".getBytes(), 0); //关联事务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也就是事务资源协调者,而两个不同的数据源mysql充当了RM资源管理着角色,在我们代码中对每个事务的准备情况进行判断,如果都OK则提交事务,如果有没有准备好的则rollback事务.
修改为一个不能正常执行的sql,来查看他的执行过程
image.png
通过断点分析,我们发现程序在执行到这句的时候就会出现异常,也就是在prepare之前sql语句已经在执行了,只不过我们设置了事务不自动提交,所以在数据库中看不到sql_1的执行结果.
修改正常sql,在prepare阶段加断点
image.png
我们在数据库中查看事务情况,因为我是在一个数据库服务器上做的的跨库数据源,所以我们能看到两条xa记录
image.png
我们继续执行到commit语句,放多第一条commit
image.png
这时候可以发现已经第一个事务的xa信息已经没有了,也就是第一个事务分支已经提交成功了.
image.png
数据库中可以看到新插入成功了一条数据
image.png
这是我们尝试修改结构或者插入一条语句,都会发现数据库处于锁定状态
image.png
等我们把断点放行之后,才可以看到其他语句正常执行,也就是说xa在提交阶段会对数据库进行加锁处理,经过进一步的分析我们发现xa在进入xa end后就对整个表进行加锁操作,因为该sql是update语句,所以在xa end 一直到事务提交或者回滚之前,整个表都处于锁定状态.
image.png
延伸
我们很容易将这种XA机制扩展到到微服务情况,需要各个微服务提供相应的机制,各个微服务提供对应的prepare接口、commit接口、rollback接口。
缺点
xA的性能问题
XA的性能很低。一个数据库的事务和多个数据库间的XA事务性能对比可发现，性能差10倍左右。因此要尽量避免XA事务，例如可以将数据写入本地，用高性能的消息系统分发数据。或使用数据库复制等技术。只有在这些都无法实现，且性能不是瓶颈时才应该使用XA
这种机制假定prepare ok的事务都可以正常commit
也就是进入prepare返回ok后，在执行commit阶段两个事务就有可能出现一些异常情况，比如第一个正常提交了，但第二个却出现了某种异常失败了。
当前标题：JAVA分布式事务的解决方法
文章分享：http://scpingwu.com/article/ijepij.html
