摘要: 原创出处 http://www.iocoder.cn/Sharding-JDBC/transaction-bed/ 「芋道源码」欢迎转载，保留摘要，谢谢！

微信排版崩崩的，建议使用 PC 点击【阅读原文】。

本文主要基于 Sharding-JDBC 1.5.0 正式版

1. 概述
2. 最大努力送达型
3. 柔性事务管理器

3.3.1 创建柔性事务

3.1 概念
3.2 柔性事务配置
3.3 柔性事务

4. 事务日志存储器

4.1 #add()
4.2 #remove()
4.3 #findEligibleTransactionLogs()
4.4 #increaseAsyncDeliveryTryTimes()
4.5 #processData()

5. 最大努力送达型事务监听器
6. 最大努力送达型异步作业

6.1 BestEffortsDeliveryJob
6.2 AsyncSoftTransactionJobConfiguration
6.3 Elastic-Job 是否必须？

7. 适用场景
8. 开发指南 & 开发示例
666. 彩蛋

🙂🙂🙂关注微信公众号：【芋道源码】有福利：
RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表
RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址
您对于源码的疑问每条留言都将得到认真回复。甚至不知道如何读源码也可以请教噢。
新的源码解析文章实时收到通知。每周更新一篇左右。
认真的源码交流微信群。

1. 概述

数据库表分库后，业务场景下的单库本地事务可能变成跨库分布式事务。虽然我们可以通过合适的分库规则让操作的数据在同库下，继续保证单库本地事务，这也是非常推崇的，但不是所有场景下都能适用。如果这些场景对事务的一致性有要求，我们就不得不解决分布式事务的“麻烦”。

分布式事务是个很大的话题，我们来看看 Sharding-JDBC 对她的权衡：

Sharding-JDBC由于性能方面的考量，决定不支持强一致性分布式事务。我们已明确规划线路图，未来会支持最终一致性的柔性事务。

Sharding-JDBC 提供了两种 柔性事务：

最大努力送达型 BED ：已经实现
事务补偿型 TCC ：计划中

本文分享最大努力送达型的实现。建议前置阅读：《Sharding-JDBC 源码分析 —— SQL 执行》。

Sharding-JDBC 正在收集使用公司名单：传送门。

🙂 你的登记，会让更多人参与和使用 Sharding-JDBC。传送门

Sharding-JDBC 也会因此，能够覆盖更多的业务场景。传送门

登记吧，骚年！传送门

2. 最大努力送达型

概念

在分布式数据库的场景下，相信对于该数据库的操作最终一定可以成功，所以通过最大努力反复尝试送达操作。

从概念看，可能不是很直白的理解是什么意思，本文会最大努力让你干净理解。

架构图

执行过程有四种情况：

【红线】执行成功
【棕线】执行失败，同步重试成功
【粉线】执行失败，同步重试失败，异步重试成功
【绿线】执行失败，同步重试失败，异步重试失败，事务日志保留

整体成漏斗倒三角，上一个阶段失败，交给下一个阶段重试：

整个过程通过如下组件完成：

柔性事务管理器
最大努力送达型柔性事务
最大努力送达型事务监听器
事务日志存储器
最大努力送达型异步作业

下面，我们逐节分享每个组件。

3. 柔性事务管理器

3.1 概念

柔性事务管理器，SoftTransactionManager 实现，负责对柔性事务配置( SoftTransactionConfiguration ) 、柔性事务( AbstractSoftTransaction )的管理。

3.2 柔性事务配置

调用 #init() 初始化柔性管理器：

// SoftTransactionManager.java
/**
* 柔性事务配置对象
*/
@Getter
privatefinalSoftTransactionConfiguration transactionConfig;
// SoftTransactionManager.java
/**
* 初始化事务管理器.
*/
publicvoid init()throwsSQLException{
// 初始化 最大努力送达型事务监听器
EventBusInstance.getInstance().register(newBestEffortsDeliveryListener());
// 初始化 事务日志数据库存储表
if(TransactionLogDataSourceType.RDB == transactionConfig.getStorageType()){
Preconditions.checkNotNull(transactionConfig.getTransactionLogDataSource());
       createTable();
}
// 初始化 内嵌的最大努力送达型异步作业
if(transactionConfig.getBestEffortsDeliveryJobConfiguration().isPresent()){
newNestedBestEffortsDeliveryJobFactory(transactionConfig).init();
}
}

将最大努力送达型事务监听器( BestEffortsDeliveryListener )注册到事务总线 ( EventBus )。在『最大努力送达型事务监听器』小节会详细分享
当使用数据库存储事务日志( TransactionLog ) 时，若事务日志表( transaction_log )不存在则进行创建。在『事务日志存储器』小节会详细分享
当配置使用内嵌的最大努力送达型异步作业( NestedBestEffortsDeliveryJob ) 时，进行初始化。在『最大努力送达型异步作业』小节会详细分享

SoftTransactionConfiguration

SoftTransactionConfiguration，柔性事务配置对象。

publicclassSoftTransactionConfiguration{
/**
     * 事务管理器管理的数据源.
     */
@Getter(AccessLevel.NONE)
privatefinalDataSource targetDataSource;
/**
     * 同步的事务送达的最大尝试次数.
     */
privateint syncMaxDeliveryTryTimes =3;
/**
     * 事务日志存储类型.
     */
privateTransactionLogDataSourceType storageType = RDB;
/**
     * 存储事务日志的数据源.
     */
privateDataSource transactionLogDataSource;
/**
     * 内嵌的最大努力送达型异步作业配置对象.
     */
privateOptional<NestedBestEffortsDeliveryJobConfiguration> bestEffortsDeliveryJobConfiguration =Optional.absent();
}

3.3 柔性事务

在 Sharding-JDBC 里，目前柔性事务分成两种：

BEDSoftTransaction ：最大努力送达型柔性事务
TCCSoftTransaction ：TCC型柔性事务

继承 AbstractSoftTransaction

publicabstractclassAbstractSoftTransaction{
/**
     * 分片连接原自动提交状态
     */
privateboolean previousAutoCommit;
/**
     * 分片连接
     */
@Getter
privateShardingConnection connection;
/**
     * 事务类型
     */
@Getter
privateSoftTransactionType transactionType;
/**
     * 事务编号
     */
@Getter
privateString transactionId;
}

AbstractSoftTransaction 实现了开启柔性事务、关闭柔性事务两个方法提供给子类调用：

#beginInternal()

对异常处理的代码：ExecutorExceptionHandler#setExceptionThrown()
对于其他 SQL，不会因为 SQL 错误不执行，会继续执行
对于上层业务，不会因为 SQL 错误终止逻辑，会继续执行。这里有一点要注意下，上层业务不能对该 SQL 执行结果有强依赖，因为 SQL 错误需要重试达到数据最终一致性
对于最大努力型事务( TCC暂未实现 )，会对执行错误的 SQL 进行重试

调用 ExecutorExceptionHandler.setExceptionThrown(false) 设置执行 SQL 错误时，也不抛出异常。
调用 connection.setAutoCommit(true);，设置执行自动提交。使用最大努力型事务时，上层业务执行 SQL 会马上提交，即使调用 Connection#rollback() 也是无法回滚的，这点一定要注意。

/**
* 开启柔性
*
* @param conn 分片连接
* @param type 事务类型
* @throws SQLException
*/
protectedfinalvoid beginInternal(finalConnection conn,finalSoftTransactionType type)throwsSQLException{
// TODO 判断如果在传统事务中，则抛异常
Preconditions.checkArgument(conn instanceofShardingConnection,"Only ShardingConnection can support eventual consistency transaction.");
// 设置执行错误，不抛出异常
ExecutorExceptionHandler.setExceptionThrown(false);
connection =(ShardingConnection) conn;
transactionType = type;
// 设置自动提交状态
previousAutoCommit = connection.getAutoCommit();
connection.setAutoCommit(true);
// 生成事务编号
// TODO 替换UUID为更有效率的id生成器
transactionId = UUID.randomUUID().toString();
}

#end()

事务结束后，一定要记得调用 #end() 清理线程变量。否则，下次请求使用到该线程，会继续在这个柔性事务内。

/**
* 结束柔性事务.
*/
publicfinalvoidend()throwsSQLException{
if(connection !=null){
ExecutorExceptionHandler.setExceptionThrown(true);
connection.setAutoCommit(previousAutoCommit);
SoftTransactionManager.closeCurrentTransactionManager();
}
}
// SoftTransactionManager.java
/**
* 关闭当前的柔性事务管理器.
*/
staticvoid closeCurrentTransactionManager(){
ExecutorDataMap.getDataMap().put(TRANSACTION,null);
ExecutorDataMap.getDataMap().put(TRANSACTION_CONFIG,null);
}

BEDSoftTransaction

BEDSoftTransaction，最大努力送达型柔性事务。

publicclassBEDSoftTransactionextendsAbstractSoftTransaction{
/**
     * 开启柔性事务.
     * 
     * @param connection 数据库连接对象
     */
publicvoidbegin(finalConnection connection)throwsSQLException{
        beginInternal(connection,SoftTransactionType.BestEffortsDelivery);
}
}

TCCSoftTransaction

TCCSoftTransaction，TCC 型柔性事务，暂未实现。实现后，会更新到《Sharding-JDBC 源码分析 —— 分布式事务（二）之事务补偿型》。

3.3.1 创建柔性事务

通过调用 SoftTransactionManager#getTransaction() 创建柔性事务对象：

/**
* {@link ExecutorDataMap#dataMap} 柔性事务对象 key
*/
privatestaticfinalString TRANSACTION ="transaction";
/**
* {@link ExecutorDataMap#dataMap} 柔性事务配置 key
*/
privatestaticfinalString TRANSACTION_CONFIG ="transactionConfig";
// SoftTransactionManager.java
/**
* 创建柔性事务.
* 
* @param type 柔性事务类型
* @return 柔性事务
*/
publicAbstractSoftTransaction getTransaction(finalSoftTransactionType type){
AbstractSoftTransaction result;
switch(type){
caseBestEffortsDelivery:
           result =newBEDSoftTransaction();
break;
caseTryConfirmCancel:
           result =newTCCSoftTransaction();
break;
default:
thrownewUnsupportedOperationException(type.toString());
}
// TODO 目前使用不支持嵌套事务，以后这里需要可配置
if(getCurrentTransaction().isPresent()){
thrownewUnsupportedOperationException("Cannot support nested transaction.");
}
ExecutorDataMap.getDataMap().put(TRANSACTION, result);
ExecutorDataMap.getDataMap().put(TRANSACTION_CONFIG, transactionConfig);
return result;
}

后续可以从 ExecutorDataMap 中获取当前线程的柔性事务和柔性事务配置：

// SoftTransactionManager.java
/**
* 获取当前线程的柔性事务配置.
*
* @return 当前线程的柔性事务配置
*/
publicstaticOptional<SoftTransactionConfiguration> getCurrentTransactionConfiguration(){
Object transactionConfig =ExecutorDataMap.getDataMap().get(TRANSACTION_CONFIG);
return(null== transactionConfig)
?Optional.<SoftTransactionConfiguration>absent()
:Optional.of((SoftTransactionConfiguration) transactionConfig);
}
/**
* 获取当前的柔性事务.
*
* @return 当前的柔性事务
*/
publicstaticOptional<AbstractSoftTransaction> getCurrentTransaction(){
Object transaction =ExecutorDataMap.getDataMap().get(TRANSACTION);
return(null== transaction)
?Optional.<AbstractSoftTransaction>absent()
:Optional.of((AbstractSoftTransaction) transaction);
}

4. 事务日志存储器

柔性事务执行过程中，会通过事务日志( TransactionLog ) 记录每条 SQL 执行状态：

SQL 执行前，记录一条事务日志
SQL 执行成功，移除对应的事务日志

通过实现事务日志存储器接口( TransactionLogStorage )，提供存储功能。目前有两种实现：

MemoryTransactionLogStorage ：基于内存的事务日志存储器。主要用于开发测试，生产环境下不要使用。
RdbTransactionLogStorage ：基于数据库的事务日志存储器。

本节只分析 RdbTransactionLogStorage。对 MemoryTransactionLogStorage 感兴趣的同学可以点击链接传送到达。

TransactionLogStorage 有五个接口方法，下文每个小标题都是一个方法。

4.1 #add()

// TransactionLogStorage.java
/**
* 存储事务日志.
* 
* @param transactionLog 事务日志
*/
void add(TransactionLog transactionLog);
// RdbTransactionLogStorage.java
@Override
publicvoid add(finalTransactionLog transactionLog){
String sql ="INSERT INTO `transaction_log` (`id`, `transaction_type`, `data_source`, `sql`, `parameters`, `creation_time`) VALUES (?, ?, ?, ?, ?, ?);";
try(
// ... 省略你熟悉的代码
}catch(finalSQLException ex){
thrownewTransactionLogStorageException(ex);
}
}

注意：如果插入事务日志失败，SQL 会继续执行，如果此时 SQL 执行失败，则该 SQL 会不见了。建议： #add() 和下文的 #remove() 异常时，都打印下异常日志都文件系统

TransactionLog (transaction_log) 数据库表结构如下：

字段	名字	数据库类型	备注
id	事件编号	VARCHAR(40)	EventBus 事件编号，非事务编号
transaction_type	柔性事务类型	VARCHAR(30)
data_source	真实数据源名	VARCHAR(255)
sql	执行 SQL	TEXT	已经改写过的 SQL
parameters	占位符参数	TEXT	JSON 字符串存储
creation_time	记录时间	LONG
asyncdeliverytry_times	已异步重试次数	INT

4.2 #remove()

// TransactionLogStorage.java
/**
* 根据主键删除事务日志.
* 
* @param id 事务日志主键
*/
void remove(String id);
// RdbTransactionLogStorage.java    
@Override
publicvoid remove(finalString id){
String sql ="DELETE FROM `transaction_log` WHERE `id`=?;";
try(
// ... 省略你熟悉的代码
}catch(finalSQLException ex){
thrownewTransactionLogStorageException(ex);
}
}

4.3 #findEligibleTransactionLogs()

// TransactionLogStorage.java
/**
* 读取需要处理的事务日志.
* 
* <p>需要处理的事务日志为: </p>
* <p>1. 异步处理次数小于最大处理次数.</p>
* <p>2. 异步处理的事务日志早于异步处理的间隔时间.</p>
* 
* @param size 获取日志的数量
* @param maxDeliveryTryTimes 事务送达的最大尝试次数
* @param maxDeliveryTryDelayMillis 执行送达事务的延迟毫秒数.
*/
List<TransactionLog> findEligibleTransactionLogs(int size,int maxDeliveryTryTimes,long maxDeliveryTryDelayMillis);
// RdbTransactionLogStorage.java
@Override
publicList<TransactionLog> findEligibleTransactionLogs(finalint size,finalint maxDeliveryTryTimes,finallong maxDeliveryTryDelayMillis){
List<TransactionLog> result =newArrayList<>(size);
String sql ="SELECT `id`, `transaction_type`, `data_source`, `sql`, `parameters`, `creation_time`, `async_delivery_try_times` "
+"FROM `transaction_log` WHERE `async_delivery_try_times`<? AND `transaction_type`=? AND `creation_time`<? LIMIT ?;";
try(Connection conn = dataSource.getConnection()){
// ... 省略你熟悉的代码
}catch(finalSQLException ex){
thrownewTransactionLogStorageException(ex);
}
return result;
}

4.4 #increaseAsyncDeliveryTryTimes()

// TransactionLogStorage.java
/**
* 增加事务日志异步重试次数.
* 
* @param id 事务主键
*/
void increaseAsyncDeliveryTryTimes(String id);
// RdbTransactionLogStorage.java
@Override
publicvoid increaseAsyncDeliveryTryTimes(finalString id){
String sql ="UPDATE `transaction_log` SET `async_delivery_try_times`=`async_delivery_try_times`+1 WHERE `id`=?;";
try(
// ... 省略你熟悉的代码
}catch(finalSQLException ex){
thrownewTransactionLogStorageException(ex);
}
}

4.5 #processData()

// TransactionLogStorage.java
/**
* 处理事务数据.
*
* @param connection 业务数据库连接
* @param transactionLog 事务日志
* @param maxDeliveryTryTimes 事务送达的最大尝试次数
*/
boolean processData(Connection connection,TransactionLog transactionLog,int maxDeliveryTryTimes);
// RdbTransactionLogStorage.java
@Override
publicboolean processData(finalConnection connection,finalTransactionLog transactionLog,finalint maxDeliveryTryTimes){
// 重试执行失败 SQL
try(
Connection conn = connection;
PreparedStatement preparedStatement = conn.prepareStatement(transactionLog.getSql())){
for(int parameterIndex =0; parameterIndex < transactionLog.getParameters().size(); parameterIndex++){
           preparedStatement.setObject(parameterIndex +1, transactionLog.getParameters().get(parameterIndex));
}
       preparedStatement.executeUpdate();
}catch(finalSQLException ex){
// 重试失败，更新事务日志，增加已异步重试次数
       increaseAsyncDeliveryTryTimes(transactionLog.getId());
thrownewTransactionCompensationException(ex);
}
// 移除重试执行成功 SQL 对应的事务日志
   remove(transactionLog.getId());
returntrue;
}

不同于前四个增删改查接口方法的实现， #processData() 是带有一些逻辑的。根据事务日志( TransactionLog )重试执行失败的 SQL，若成功，移除事务日志；若失败，更新事务日志，增加已异步重试次数
该方法会被最大努力送达型异步作业调用到

5. 最大努力送达型事务监听器

最大努力送达型事务监听器，BestEffortsDeliveryListener，负责记录事务日志、同步重试执行失败 SQL。

// BestEffortsDeliveryListener.java
@Subscribe
@AllowConcurrentEvents
publicvoid listen(finalDMLExecutionEventevent){
if(!isProcessContinuously()){
return;
}
SoftTransactionConfiguration transactionConfig =SoftTransactionManager.getCurrentTransactionConfiguration().get();
TransactionLogStorage transactionLogStorage =TransactionLogStorageFactory.createTransactionLogStorage(transactionConfig.buildTransactionLogDataSource());
BEDSoftTransaction bedSoftTransaction =(BEDSoftTransaction)SoftTransactionManager.getCurrentTransaction().get();
switch(event.getEventExecutionType()){
case BEFORE_EXECUTE:// 执行前，插入事务日志
//TODO 对于批量执行的SQL需要解析成两层列表
           transactionLogStorage.add(newTransactionLog(event.getId(), bedSoftTransaction.getTransactionId(), bedSoftTransaction.getTransactionType(),
event.getDataSource(),event.getSql(),event.getParameters(),System.currentTimeMillis(),0));
return;
case EXECUTE_SUCCESS:// 执行成功，移除事务日志
           transactionLogStorage.remove(event.getId());
return;
case EXECUTE_FAILURE:// 执行失败，同步重试
boolean deliverySuccess =false;
for(int i =0; i < transactionConfig.getSyncMaxDeliveryTryTimes(); i++){// 同步【多次】重试
if(deliverySuccess){
return;
}
boolean isNewConnection =false;
Connection conn =null;
PreparedStatement preparedStatement =null;
try{
// 获得数据库连接
                   conn = bedSoftTransaction.getConnection().getConnection(event.getDataSource(),SQLType.DML);
if(!isValidConnection(conn)){// 因为可能执行失败是数据库连接异常，所以判断一次，如果无效，重新获取数据库连接
                       bedSoftTransaction.getConnection().release(conn);
                       conn = bedSoftTransaction.getConnection().getConnection(event.getDataSource(),SQLType.DML);
                       isNewConnection =true;
}
                   preparedStatement = conn.prepareStatement(event.getSql());
// 同步重试
//TODO 对于批量事件需要解析成两层列表
for(int parameterIndex =0; parameterIndex <event.getParameters().size(); parameterIndex++){
                       preparedStatement.setObject(parameterIndex +1,event.getParameters().get(parameterIndex));
}
                   preparedStatement.executeUpdate();
                   deliverySuccess =true;
// 同步重试成功，移除事务日志
                   transactionLogStorage.remove(event.getId());
}catch(finalSQLException ex){
                   log.error(String.format("Delivery times %s error, max try times is %s", i +1, transactionConfig.getSyncMaxDeliveryTryTimes()), ex);
}finally{
                   close(isNewConnection, conn, preparedStatement);
}
}
return;
default:
thrownewUnsupportedOperationException(event.getEventExecutionType().toString());
}
}

BestEffortsDeliveryListener 通过 EventBus 实现监听 SQL 的执行。Sharding-JDBC 如何实现 EventBus 的，请看《Sharding-JDBC 源码分析 —— SQL 执行》
调用 #isProcessContinuously() 方法判断是否处于最大努力送达型事务中，当且仅当处于该状态才进行监听事件处理
SQL 执行前，插入事务日志
SQL 执行成功，移除事务日志
SQL 执行失败，根据柔性事务配置( SoftTransactionConfiguration )同步的事务送达的最大尝试次数( syncMaxDeliveryTryTimes )进行多次重试直到成功。总体逻辑和 RdbTransactionLogStorage#processData() 方法逻辑类似，区别在于获取分片数据库连接的特殊处理：此处调用失败，数据库连接可能是异常无效的，因此调用了 #isValidConnection() 判断连接的有效性。若无效，则重新获取分片数据库连接。另外，若是重新获取分片数据库连接，需要进行关闭释放 ( Connection#close())：

// BestEffortsDeliveryListener.java
/**
* 通过 SELECT 1 校验数据库连接是否有效
*
* @param conn 数据库连接
* @return 是否有效
*/
privateboolean isValidConnection(finalConnection conn){
try(PreparedStatement preparedStatement = conn.prepareStatement("SELECT 1")){
try(ResultSet rs = preparedStatement.executeQuery()){
return rs.next()&&1== rs.getInt("1");
}
}catch(finalSQLException ex){
returnfalse;
}
}
/**
* 关闭释放预编译SQL对象和数据库连接
*
* @param isNewConnection 是否新创建的数据库连接，是的情况下才释放
* @param conn 数据库连接
* @param preparedStatement 预编译SQL
*/
privatevoid close(finalboolean isNewConnection,finalConnection conn,finalPreparedStatement preparedStatement){
if(null!= preparedStatement){
try{
preparedStatement.close();
}catch(finalSQLException ex){
log.error("PreparedStatement closed error:", ex);
}
}
if(isNewConnection &&null!= conn){
try{
conn.close();
}catch(finalSQLException ex){
log.error("Connection closed error:", ex);
}
}
}

6. 最大努力送达型异步作业

当最大努力送达型事务监听器( BestEffortsDeliveryListener )多次同步重试失败后，交给最大努力送达型异步作业进行多次异步重试，并且多次执行有固定间隔。

Sharding-JDBC 提供了两个最大努力送达型异步作业实现：

NestedBestEffortsDeliveryJob ：内嵌的最大努力送达型异步作业
BestEffortsDeliveryJob ：最大努力送达型异步作业

两者实现代码逻辑基本一致。前者相比后者，用于开发测试，去除对 Zookeeper 依赖，无法实现高可用，因此生产环境下不适合使用。

6.1 BestEffortsDeliveryJob

BestEffortsDeliveryJob 所在 Maven 项目为 sharding-jdbc-transaction-async-job，基于当当开源的 Elastic-Job 实现。如下是官方对该 Maven 项目的简要说明：

由于柔性事务采用异步尝试，需要部署独立的作业和Zookeeper。sharding-jdbc-transaction采用elastic-job实现的sharding-jdbc-transaction-async-job，通过简单配置即可启动高可用作业异步送达柔性事务，启动脚本为start.sh。

BestEffortsDeliveryJob

publicclassBestEffortsDeliveryJobextendsAbstractIndividualThroughputDataFlowElasticJob<TransactionLog>{
/**
     * 最大努力送达型异步作业配置对象
     */
@Setter
privateBestEffortsDeliveryConfiguration bedConfig;
/**
     * 事务日志存储器对象
     */
@Setter
privateTransactionLogStorage transactionLogStorage;
@Override
publicList<TransactionLog> fetchData(finalJobExecutionMultipleShardingContext context){
return transactionLogStorage.findEligibleTransactionLogs(context.getFetchDataCount(),
            bedConfig.getJobConfig().getMaxDeliveryTryTimes(), bedConfig.getJobConfig().getMaxDeliveryTryDelayMillis());
}
@Override
publicboolean processData(finalJobExecutionMultipleShardingContext context,finalTransactionLog data){
try(
Connection conn = bedConfig.getTargetDataSource(data.getDataSource()).getConnection()){
            transactionLogStorage.processData(conn, data, bedConfig.getJobConfig().getMaxDeliveryTryTimes());
}catch(finalSQLException|TransactionCompensationException ex){
            log.error(String.format("Async delivery times %s error, max try times is %s, exception is %s", data.getAsyncDeliveryTryTimes()+1,
                bedConfig.getJobConfig().getMaxDeliveryTryTimes(), ex.getMessage()));
returnfalse;
}
returntrue;
}
@Override
publicboolean isStreamingProcess(){
returnfalse;
}
}

调用 #fetchData() 方法获取需要处理的事务日志 (TransactionLog)，内部调用了 TransactionLogStorage#findEligibleTransactionLogs() 方法
调用 #processData() 方法处理事务日志，重试执行失败的 SQL，内部调用了 TransactionLogStorage#processData()
#fetchData() 和 #processData() 调用是 Elastic-Job 控制的。每一轮定时调度，每条事务日志只执行一次。当超过最大异步调用次数后，该条事务日志不再处理，所以生产使用时，最好增加下相应监控超过最大异步重试次数的事务日志。

6.2 AsyncSoftTransactionJobConfiguration

AsyncSoftTransactionJobConfiguration，异步柔性事务作业配置对象。

publicclassAsyncSoftTransactionJobConfiguration{
/**
     * 作业名称.
     */
privateString name ="bestEffortsDeliveryJob";
/**
     * 触发作业的cron表达式.
     */
privateString cron ="0/5 * * * * ?";
/**
     * 每次作业获取的事务日志最大数量.
     */
privateint transactionLogFetchDataCount =100;
/**
     * 事务送达的最大尝试次数.
     */
privateint maxDeliveryTryTimes =3;
/**
     * 执行事务的延迟毫秒数.
     *
     * <p>早于此间隔时间的入库事务才会被作业执行.</p>
     */
privatelong maxDeliveryTryDelayMillis =60*1000L;
}

6.3 Elastic-Job 是否必须？

Sharding-JDBC 提供的最大努力送达型异步作业实现( BestEffortsDeliveryJob )，通过与 Elastic-Job 集成，可以很便捷并且有质量保证的高可用、高性能使用。一部分团队，可能已经引入或自研了类似 Elastic-Job 的分布式作业中间件解决方案，每多一个中间件，就是多一个学习与运维成本。那么是否可以使用自己的分布式作业解决方案？答案是，可以的。参考 BestEffortsDeliveryJob 的实现，通过调用 TransactionLogStorage 来实现：

// 伪代码(不考虑性能、异常)
List<TransactionLog> transactionLogs = transactionLogStorage.findEligibleTransactionLogs(....);
for(TransactionLog transactionLog : transactionLogs){
       transactionLogStorage.processData(conn, log, maxDeliveryTryTimes);
}

当然，个人还是很推荐 Elastic-Job。

😈 笔者要开始写《Elastic-Job 源码分析》。

另外，如果有支持事务消息的分布式队列系统，可以通过 TransactionLogStorage 实现存储事务消息存储成消息。为什么要支持事务消息？如果 SQL 执行是成功的，需要回滚（删除）事务消息。

7. 适用场景

见《官方文档 - 事务支持》。

8. 开发指南 & 开发示例