分布式事务剖析
# 分布式事务剖析
ShardingSphere > 用户手册 > ShardingSphere-JDBC > 特殊 API > 分布式事务 (opens new window)
分布式理论见:分布式事务解决方案 (opens new window)
# 整合XA原理
Java 通过定义 JTA 接口实现了 XA 模型,JTA 接口中的 ResourceManager
需要数据库厂商提供 XA 驱动实现, 而TransactionManager
则需要事务管理器的厂商实现,传统的事务管理器需要同应用服务器绑定,因此使用的成本很高。 而嵌入式的事务管器可以通过 jar 形式提供服务,同 Apache ShardingSphere 集成后,可保证分片后跨库事务强一致性。
ShardingSphere 在整合 XA 事务时,采用分离 XA 事务管理和连接池管理的方式,做到对应用程序的零侵入。主要支持一下以下功能:
- 支持数据分片后的跨库XA事务
- 两阶段提交保证操作的原子性和数据的强一致性
- 服务宕机重启后,提交/回滚中的事务可自动恢复
- SPI机制整合主流的XA事务管理器,默认 Atomikos
- 同时支持XA和非XA的连接池
- 提供spring-boot和namespace的接入端
整合原理图:
Begin(开启XA全局事务):XA Sharding Transaction Manager会调用具体的XA事务管理器开启XA的全局事务。
执行物理SQL:ShardingSphere进行解析/优化/路由后会生成SQL操作,执行引擎为每个物理SQL创建连接的同时,物理连接所对应的XA Resource也会被注册到当前XA事务中。事务管理器会在此阶段发送XA Resource.start命令给数据库,数据库在收到XAResource.end命令之前的所有SQL操作,会被标记为XA事务。 例如:
XAResource1.start ## Enlist阶段执行 statement.execute("sql1"); ## 模拟执行一个分片SQL1 , 会被标记为XA事务 statement.execute("sql2"); ## 模拟执行一个分片SQL2 ,会被标记为XA事务 XAResource1.end ## 提交阶段执行
Commit/Rollback(提交XA事务):XA Sharding Transaction Manager收到接入端的提交命令后,会委托实际的XA事务管理进行提交动作,这时事务管理器会收集当前线程里所有注册的XA Resource,首先发送XAResource.end指令,用以标记此XA事务的边界。 接着会依次发送prepare指令,收集所有参与XA Resource投票,如果所有XAResource的反馈结果都是OK,则会再次调用commit指令进行最终提交,如果有一个XA Resource的反馈结果为No,则会调用rollback指令进行回滚。 在事务管理器发出提交指令后,任何XA Resource产生的异常都会通过recovery日志进行重试,来保证提交阶段的操作原子性和数据强一致性。
例如:
# commit XAResource1.prepare ## ack: yes XAResource2.prepare ## ack: yes XAResource1.commit XAResource2.commit # rollback XAResource1.prepare ## ack: yes XAResource2.prepare ## ack: no XAResource1.rollback XAResource2.rollback
# 整合Saga原理
ShardingSphere的柔性事务已通过第三方servicecomb-saga组件实现的,通过SPI机制注入使用。ShardingSphere是基于反向SQL技术实现的反向补偿操作,它将对数据库进行更新操作的SQL自动生成反向SQL,并交由Saga-actuator引擎执行。使用方则无需再关注如何实现补偿方法,将柔性事务管理器的应用范畴成功的定位回了事务的本源——数据库层面。
ShardingSphere支持以下功能:
- 完全支持跨库事务
- 支持失败SQL重试及最大努力送达
- 支持反向SQL、自动生成更新快照以及自动补偿
- 默认使用关系型数据库进行快照及事务日志的持久化,支持使用SPI的方式加载其他类型的持久化
实现原理:Saga柔性事务的实现类为SagaShardingTransactionMananger, ShardingSphere通过Hook的方式拦截逻辑SQL的解析和路由结果,这样,在分片物理SQL执行前,可以生成逆向SQL,在事务提交阶段再把SQL调用链交给Saga引擎处理。如下图:
具体流程概述:
- Init(Saga引擎初始化):包含Saga柔性事务的应用启动时,saga-actuator引擎会根据saga.properties的配置进行初始化的 流程。
- Begin(开启Saga全局事务):每次开启Saga全局事务时,将会生成本次全局事务的上下文(SagaTransactionContext),事务 上下文记录了所有子事务的正向SQL和逆向SQL,作为生成事务调用链的元数据使用。
- 执行物理SQL:在物理SQL执行前,ShardingSphere根据SQL的类型生成逆向SQL,这里是通过Hook的方式拦截Parser的解析结果进行实现。
- Commit/Rollback(提交Saga事务):提交阶段会生成Saga执行引擎所需的调用链路图,commit操作产生ForwardRecovery(正向SQL补偿)任务,rollback操作产生BackwardRecovery任务(逆向SQL补偿)。
# 整合Seata原理
分布式事务的实现目前主要分为两阶段的XA强事务和BASE柔性事务。如下图:
Seata AT事务作为BASE柔性事务的一种实现,可以无缝接入到ShardingSphere生态中。在整合Seata-AT事务时,需要把TM,RM,TC的模型融入到ShardingSphere 分布式事务的SPI的生态中。
在数据库资源上,Seata通过对接DataSource接口,让JDBC操作可以同TC进行RPC通信。同样,ShardingSphere也是面向DataSource接口对用户配置的物理DataSource进行了聚合,因此把物理DataSource二次包装为Seata 的DataSource后,就可以把Seata-AT事务融入到ShardingSphere的分片中。原理如下图:
详细流程如下:
- Init(Seata引擎初始化):包含Seata柔性事务的应用启动时,用户配置的数据源会按seata.conf的配置,适配成Seata事务所 需的DataSourceProxy,并且注册到RM中。
- Begin(开启Seata全局事务):TM控制全局事务的边界,TM通过向TC发送Begin指令,获取全局事务ID,所有分支事务通过此全 局事务ID,参与到全局事务中;全局事务ID的上下文存放在当前线程变量中。
- 执行分片物理SQL:处于Seata全局事务中的分片SQL通过RM生成undo快照,并且发送participate指令到TC,加入到全局事务中。ShardingSphere的分片物理SQL是按多线程方式执行,因此整合Seata AT事务时,需要在主线程和子线程间进行全局事务ID的上下文传递,这同服务间的上下文传递思路完全相同。
- Commit/Rollback(提交Seata事务):提交Seata事务时,TM会向TC发送全局事务的commit和rollback指令,TC根据全局事务ID协调所有分支事务进行commit和rollback。
# 实战
ShardingSphere整合了XA、Saga和Seata模式后,为分布式事务控制提供了极大的便利,我们可以在应用程序编程时,采用以下统一模式进行使用。
# 第一步:引入 Maven 依赖
<!-- 使用 XA 事务时,需要引入此模块 -->
<dependency>
<groupId>org.apache.shardingsphere</groupId>
<artifactId>shardingsphere-transaction-xa-core</artifactId>
<version>${shardingsphere.version}</version>
</dependency>
<!-- 使用 Saga 事务时,需要引入此模块 -->
<dependency>
<groupId>io.shardingsphere</groupId>
<artifactId>sharding-transaction-base-saga</artifactId>
<version>${shardingsphere-spi-impl.version}</version>
</dependency>
<!-- 使用 BASE 事务时,需要引入此模块 -->
<dependency>
<groupId>org.apache.shardingsphere</groupId>
<artifactId>shardingsphere-transaction-base-seata-at</artifactId>
<version>${shardingsphere.version}</version>
</dependency>
目前ShardingSphere没有整合Saga事务的实现,但是仍然可以使用第三方实现的Saga事务。
项目地址: shardingsphere-spi-impl (opens new window),文中涉及${shardingsphere-spi-impl.version} 的jar暂未发布到maven中央仓,因此需要根据源码自行部署。
# 第二步:配置事务管理器
ShardingSphere默认的XA事务管理器为Atomikos,通过在项目的classpath中添加jta.properties来定制化Atomikos配置项。具体的配置规则如下,更多可以 Atomikos 官方文档 (opens new window)。
#指定是否启动磁盘日志,默认为true。在生产环境下一定要保证为true,否则数据的完整性无法保证 com.atomikos.icatch.enable_logging=true #JTA/XA资源是否应该自动注册 com.atomikos.icatch.automatic_resource_registration=true #JTA事务的默认超时时间,默认为10000ms com.atomikos.icatch.default_jta_timeout=10000 #事务的最大超时时间,默认为300000ms。这表示事务超时时间由 UserTransaction.setTransactionTimeout()较大者决定。4.x版本之后,指定为0的话则表示不设置超时时间 com.atomikos.icatch.max_timeout=300000 #指定在两阶段提交时,是否使用不同的线程(意味着并行)。3.7版本之后默认为false,更早的版本 默认为true。如果为false,则提交将按照事务中访问资源的顺序进行。 com.atomikos.icatch.threaded_2pc=false #指定最多可以同时运行的事务数量,默认值为50,负数表示没有数量限制。在调用 UserTransaction.begin()方法时,可能会抛出一个”Max number of active transactions reached”异常信息,表示超出最大事务数限制 com.atomikos.icatch.max_actives=50 #是否支持subtransaction,默认为true com.atomikos.icatch.allow_subtransactions=true #指定在可能的情况下,否应该join 子事务(subtransactions),默认值为true。如果设置为 false,对于有关联的不同subtransactions,不会调用XAResource.start(TM_JOIN) com.atomikos.icatch.serial_jta_transactions=true #指定JVM关闭时是否强制(force)关闭事务管理器,默认为false com.atomikos.icatch.force_shutdown_on_vm_exit=false #在正常关闭(no-force)的情况下,应该等待事务执行完成的时间,默认为Long.MAX_VALUE com.atomikos.icatch.default_max_wait_time_on_shutdown=9223372036854775807 ========= 日志记录配置======= #事务日志目录,默认为./。 com.atomikos.icatch.log_base_dir=./ #事务日志文件前缀,默认为tmlog。事务日志存储在文件中,文件名包含一个数字后缀,日志文件 以.log为扩展名,如tmlog1.log。遇到checkpoint时,新的事务日志文件会被创建,数字增加。 com.atomikos.icatch.log_base_name=tmlog #指定两次checkpoint的时间间隔,默认为500 com.atomikos.icatch.checkpoint_interval=500 =========日志恢复配置============= #指定在多长时间后可以清空无法恢复的事务日志(orphaned),默认86400000ms com.atomikos.icatch.forget_orphaned_log_entries_delay=86400000 #指定两次恢复扫描之间的延迟时间。默认值为与com.atomikos.icatch.default_jta_timeout 相同 com.atomikos.icatch.recovery_delay=${com.atomikos.icatch.default_jta_timeout} #提交失败时,再抛出一个异常之前,最多可以重试几次,默认值为5 com.atomikos.icatch.oltp_max_retries=5 #提交失败时,每次重试的时间间隔,默认10000ms com.atomikos.icatch.oltp_retry_interval=10000
Saga可以通过在项目的classpath中添加 saga.properties 来定制化Saga事务的配置项。配置项的属性及说明如下:
属性名称 默认值 说明 saga.actuator.executor.size 5 使用的线程池大小 saga.actuator.transaction.max.retries 5 失败SQL的最大重试次数 saga.actuator.compensation.max.retries 5 失败SQL的最大尝试补偿次数 saga.actuator.transaction.retry.delay.milliseconds 5000 失败SQL的重试间隔,单位毫秒 saga.actuator.compensation.retry.delay.milliseconds 3000 失败SQL的补偿间隔,单位毫秒 saga.persistence.enabled false 快照及执行日志是否进行持久化 saga.actuator.recovery.policy ForwardRecovery 失败事务的补偿策略,ForwardRecovery为最大努力送达,BackwardRecovery为反向SQL补偿 saga.persistence.ds.url 无 快照及日志持久化的数据库JDBC链接 saga.persistence.ds.username 无 持久化的数据库用户名 saga.persistence.ds.password 无 持久化的数据库密码 saga.persistence.ds.max.pool.size 50 持久化的数据库链接池最大连接数 saga.persistence.ds.min.pool.size 1 持久化的数据库链接池最小连接数 saga.persistence.ds.max.life.time.milliseconds 0(无限制) 持久化的数据库链接最大存活时间,单位毫秒 saga.persistence.ds.idle.timeout.milliseconds 60 * 1000 持久化的数据库链接空闲回收时间,单位毫秒 saga.persistence.ds.connection.timeout.milliseconds 30 * 1000 持久化的数据库链接超时时间,单位毫秒
# 第三步:使用分布式事务
目前SharingSphere有两种方式声明使用分布式事务:Java编码方式以及注解方式:
// Java编发方式支持 TransactionType.LOCAL, TransactionType.XA, TransactionType.BASE
public void method() {
......
TransactionTypeHolder.set(TransactionType.XA);
try (Connection conn = dataSource.getConnection()) { // 使用 ShardingSphereDataSource
conn.setAutoCommit(false);
PreparedStatement ps = conn.prepareStatement("INSERT INTO t_order (user_id, status) VALUES (?, ?)");
ps.setObject(1, 1000);
ps.setObject(2, "init");
ps.executeUpdate();
conn.commit();
}
......
}
// 注解方式, 支持TransactionType.LOCAL, TransactionType.XA, TransactionType.BASE
@Transactional
@ShardingSphereTransactionType(TransactionType.XA)
public void insert() {
jdbcTemplate.execute("INSERT INTO t_order (user_id, status) VALUES (?, ?)", (PreparedStatementCallback<Object>) ps -> {
ps.setObject(1, i);
ps.setObject(2, "init");
ps.executeUpdate();
});
}