数据库中间件 Sharding-JDBC 源码分析 —— 事务（一）之BED

// SoftTransactionManager.java
/**
* 柔性事务配置对象
*/
@Getter
private final SoftTransactionConfiguration transactionConfig;  

// SoftTransactionManager.java
/**
* 初始化事务管理器.
*/
public void init() throws SQLException {
   // 初始化 最大努力送达型事务监听器
   EventBusInstance.getInstance().register(new BestEffortsDeliveryListener());
   // 初始化 事务日志数据库存储表
   if (TransactionLogDataSourceType.RDB == transactionConfig.getStorageType()) {
       Preconditions.checkNotNull(transactionConfig.getTransactionLogDataSource());
       createTable();
   }
   // 初始化 内嵌的最大努力送达型异步作业
   if (transactionConfig.getBestEffortsDeliveryJobConfiguration().isPresent()) {
       new NestedBestEffortsDeliveryJobFactory(transactionConfig).init();
   }
}

public class SoftTransactionConfiguration {
    /**
     * 事务管理器管理的数据源.
     */
    @Getter(AccessLevel.NONE)
    private final DataSource targetDataSource;

    /**
     * 同步的事务送达的最大尝试次数.
     */
    private int syncMaxDeliveryTryTimes = 3;

    /**
     * 事务日志存储类型.
     */
    private TransactionLogDataSourceType storageType = RDB;
    /**
     * 存储事务日志的数据源.
     */
    private DataSource transactionLogDataSource;

    /**
     * 内嵌的最大努力送达型异步作业配置对象.
     */
    private Optional<NestedBestEffortsDeliveryJobConfiguration> bestEffortsDeliveryJobConfiguration = Optional.absent();
}

public abstract class AbstractSoftTransaction {
    /**
     * 分片连接原自动提交状态
     */
    private boolean previousAutoCommit;
    /**
     * 分片连接
     */
    @Getter
    private ShardingConnection connection;
    /**
     * 事务类型
     */
    @Getter
    private SoftTransactionType transactionType;
    /**
     * 事务编号
     */
    @Getter
    private String transactionId;
}

/**
* 开启柔性
*
* @param conn 分片连接
* @param type 事务类型
* @throws SQLException
*/
protected final void beginInternal(final Connection conn, final SoftTransactionType type) throws SQLException {
 // TODO 判断如果在传统事务中，则抛异常
 Preconditions.checkArgument(conn instanceof ShardingConnection, "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();
}

/**
* 结束柔性事务.
*/
public final void end() throws SQLException {
 if (connection != null) {
 ExecutorExceptionHandler.setExceptionThrown(true);
      connection.setAutoCommit(previousAutoCommit);
 SoftTransactionManager.closeCurrentTransactionManager();
 }
}

// SoftTransactionManager.java
/**
* 关闭当前的柔性事务管理器.
*/
static void closeCurrentTransactionManager() {
 ExecutorDataMap.getDataMap().put(TRANSACTION, null);
 ExecutorDataMap.getDataMap().put(TRANSACTION_CONFIG, null);
}

public class BEDSoftTransaction extends AbstractSoftTransaction {

    /**
     * 开启柔性事务.
     * 
     * @param connection 数据库连接对象
     */
    public void begin(final Connection connection) throws SQLException {
        beginInternal(connection, SoftTransactionType.BestEffortsDelivery);
    }
}

/**
* {@link ExecutorDataMap#dataMap} 柔性事务对象 key
*/
private static final String TRANSACTION = "transaction";
/**
* {@link ExecutorDataMap#dataMap} 柔性事务配置 key
*/
private static final String TRANSACTION_CONFIG = "transactionConfig";

// SoftTransactionManager.java
/**
* 创建柔性事务.
* 
* @param type 柔性事务类型
* @return 柔性事务
*/
public AbstractSoftTransaction getTransaction(final SoftTransactionType type) {
   AbstractSoftTransaction result;
   switch (type) {
       case BestEffortsDelivery: 
           result = new BEDSoftTransaction();
           break;
       case TryConfirmCancel:
           result = new TCCSoftTransaction();
           break;
       default: 
           throw new UnsupportedOperationException(type.toString());
   }
   // TODO 目前使用不支持嵌套事务，以后这里需要可配置
   if (getCurrentTransaction().isPresent()) {
       throw new UnsupportedOperationException("Cannot support nested transaction.");
   }
   ExecutorDataMap.getDataMap().put(TRANSACTION, result);
   ExecutorDataMap.getDataMap().put(TRANSACTION_CONFIG, transactionConfig);
   return result;
}

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

/**
* 获取当前的柔性事务.
* 
* @return 当前的柔性事务
*/
public static Optional<AbstractSoftTransaction> getCurrentTransaction() {
 Object transaction = ExecutorDataMap.getDataMap().get(TRANSACTION);
 return (null == transaction)
 ? Optional.<AbstractSoftTransaction>absent()
 : Optional.of((AbstractSoftTransaction) transaction);
}

// TransactionLogStorage.java
/**
* 存储事务日志.
* 
* @param transactionLog 事务日志
*/
void add(TransactionLog transactionLog);

// RdbTransactionLogStorage.java
@Override
public void add(final TransactionLog transactionLog) {
   String sql = "INSERT INTO `transaction_log` (`id`, `transaction_type`, `data_source`, `sql`, `parameters`, `creation_time`) VALUES (?, ?, ?, ?, ?, ?);";
   try (
    // ... 省略你熟悉的代码
   } catch (final SQLException ex) {
       throw new TransactionLogStorageException(ex);
   }
}

// TransactionLogStorage.java
/**
* 根据主键删除事务日志.
* 
* @param id 事务日志主键
*/
void remove(String id);

// RdbTransactionLogStorage.java    
@Override
public void remove(final String id) {
   String sql = "DELETE FROM `transaction_log` WHERE `id`=?;";
   try (
          // ... 省略你熟悉的代码
   } catch (final SQLException ex) {
       throw new TransactionLogStorageException(ex);
   }
}

// 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
public List<TransactionLog> findEligibleTransactionLogs(final int size, final int maxDeliveryTryTimes, final long maxDeliveryTryDelayMillis) {
   List<TransactionLog> result = new ArrayList<>(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 (final SQLException ex) {
       throw new TransactionLogStorageException(ex);
   }
   return result;
}

// TransactionLogStorage.java
/**
* 增加事务日志异步重试次数.
* 
* @param id 事务主键
*/
void increaseAsyncDeliveryTryTimes(String id);

// RdbTransactionLogStorage.java
@Override
public void increaseAsyncDeliveryTryTimes(final String id) {
   String sql = "UPDATE `transaction_log` SET `async_delivery_try_times`=`async_delivery_try_times`+1 WHERE `id`=?;";
   try (
       // ... 省略你熟悉的代码
   } catch (final SQLException ex) {
       throw new TransactionLogStorageException(ex);
   }
}

// TransactionLogStorage.java
/**
* 处理事务数据.
*
* @param connection 业务数据库连接
* @param transactionLog 事务日志
* @param maxDeliveryTryTimes 事务送达的最大尝试次数
*/
boolean processData(Connection connection, TransactionLog transactionLog, int maxDeliveryTryTimes);

// RdbTransactionLogStorage.java
@Override
public boolean processData(final Connection connection, final TransactionLog transactionLog, final int 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 (final SQLException ex) {
       // 重试失败，更新事务日志，增加已异步重试次数
       increaseAsyncDeliveryTryTimes(transactionLog.getId());
       throw new TransactionCompensationException(ex);
   }
   // 移除重试执行成功 SQL 对应的事务日志
   remove(transactionLog.getId());
   return true;
}

// BestEffortsDeliveryListener.java
@Subscribe
@AllowConcurrentEvents
public void listen(final DMLExecutionEvent event) {
   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(new TransactionLog(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 (final SQLException 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: 
           throw new UnsupportedOperationException(event.getEventExecutionType().toString());
   }
}

// BestEffortsDeliveryListener.java
/**
* 通过 SELECT 1 校验数据库连接是否有效
*
* @param conn 数据库连接
* @return 是否有效
*/
private boolean isValidConnection(final Connection conn) {
 try (PreparedStatement preparedStatement = conn.prepareStatement("SELECT 1")) {
 try (ResultSet rs = preparedStatement.executeQuery()) {
 return rs.next() && 1 == rs.getInt("1");
 }
 } catch (final SQLException ex) {
 return false;
 }
}

/**
* 关闭释放预编译SQL对象和数据库连接
*
* @param isNewConnection 是否新创建的数据库连接，是的情况下才释放
* @param conn 数据库连接
* @param preparedStatement 预编译SQL
*/
private void close(final boolean isNewConnection, final Connection conn, final PreparedStatement preparedStatement) {
 if (null != preparedStatement) {
 try {
           preparedStatement.close();
 } catch (final SQLException ex) {
           log.error("PreparedStatement closed error:", ex);
 }
 }
 if (isNewConnection && null != conn) {
 try {
           conn.close();
 } catch (final SQLException ex) {
           log.error("Connection closed error:", ex);
 }
 }
}

public class BestEffortsDeliveryJob extends AbstractIndividualThroughputDataFlowElasticJob<TransactionLog> {

    /**
     * 最大努力送达型异步作业配置对象
     */
    @Setter
    private BestEffortsDeliveryConfiguration bedConfig;
    /**
     * 事务日志存储器对象
     */
    @Setter
    private TransactionLogStorage transactionLogStorage;

    @Override
    public List<TransactionLog> fetchData(final JobExecutionMultipleShardingContext context) {
        return transactionLogStorage.findEligibleTransactionLogs(context.getFetchDataCount(),
            bedConfig.getJobConfig().getMaxDeliveryTryTimes(), bedConfig.getJobConfig().getMaxDeliveryTryDelayMillis());
    }

    @Override
    public boolean processData(final JobExecutionMultipleShardingContext context, final TransactionLog data) {
        try (
            Connection conn = bedConfig.getTargetDataSource(data.getDataSource()).getConnection()) {
            transactionLogStorage.processData(conn, data, bedConfig.getJobConfig().getMaxDeliveryTryTimes());
        } catch (final SQLException | 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()));
            return false;
        }
        return true;
    }

    @Override
    public boolean isStreamingProcess() {
        return false;
    }
}

public class AsyncSoftTransactionJobConfiguration {

    /**
     * 作业名称.
     */
    private String name = "bestEffortsDeliveryJob";

    /**
     * 触发作业的cron表达式.
     */
    private String cron = "0/5 * * * * ?";

    /**
     * 每次作业获取的事务日志最大数量.
     */
    private int transactionLogFetchDataCount = 100;

    /**
     * 事务送达的最大尝试次数.
     */
    private int maxDeliveryTryTimes = 3;

    /**
     * 执行事务的延迟毫秒数.
     *
     * <p>早于此间隔时间的入库事务才会被作业执行.</p>
     */
    private long maxDeliveryTryDelayMillis = 60  * 1000L;
}

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