9. sharding-jdbc源碼之最大努力型事務

阿飛Javaer，轉(zhuǎn)載請注明原創(chuàng)出處，謝謝！

BASE Transaction

  • Best efforts delivery transaction (已經(jīng)實現(xiàn)).
  • Try confirm cancel transaction (待定).

Sharding-JDBC由于性能方面的考量，決定不支持強一致性分布式事務。

最大努力送達型事務

在分布式數(shù)據(jù)庫的場景下，相信對于該數(shù)據(jù)庫的操作最終一定可以成功，所以通過最大努力反復嘗試送達操作。

最大努力送達型事務的架構(gòu)圖

最大努力送達型事務的架構(gòu)圖

摘自sharding-jdbc使用指南?事務支持

執(zhí)行過程有以下幾種情況：

  1. 執(zhí)行成功--如圖所示，執(zhí)行結(jié)果事件->監(jiān)聽執(zhí)行事件->執(zhí)行成功->清理事務日志
  2. 執(zhí)行失敗，同步重試成功--如圖所示，執(zhí)行結(jié)果事件->監(jiān)聽執(zhí)行事件->執(zhí)行失敗->重試執(zhí)行->執(zhí)行成功->清理事務日志
  3. 執(zhí)行失敗，同步重試失敗，異步重試成功--如圖所示，執(zhí)行結(jié)果事件->監(jiān)聽執(zhí)行事件->執(zhí)行失敗->重試執(zhí)行->執(zhí)行失敗->"異步送達作業(yè)"重試執(zhí)行->執(zhí)行成功->清理事務日志
  4. 執(zhí)行失敗，同步重試失敗，異步重試失敗，事務日志保留----如圖所示，執(zhí)行結(jié)果事件->監(jiān)聽執(zhí)行事件->執(zhí)行失敗->重試執(zhí)行->執(zhí)行失敗->"異步送達作業(yè)"重試執(zhí)行->執(zhí)行失敗->... ...

說明：不管執(zhí)行結(jié)果如何，執(zhí)行前事件都會記錄事務日志；執(zhí)行事件類型包括3種：BEFORE_EXECUTE，EXECUTE_FAILURE和EXECUTE_SUCCESS；另外，這里的"同步"不是絕對的同步執(zhí)行，而是通過google-guava的EventBus發(fā)布事件后，在監(jiān)聽端判斷是EXECUTE_FAILURE事件，最多重試syncMaxDeliveryTryTimes次；后面對BestEffortsDeliveryListener的源碼分析有介紹；這里的"異步"通過外掛實現(xiàn)，在后面的文章10. sharding-jdbc源碼之異步送達JOB會有分析；

適用場景

  • 根據(jù)主鍵刪除數(shù)據(jù)。
  • 更新記錄永久狀態(tài)，如更新通知送達狀態(tài)。

使用限制

  • 使用最大努力送達型柔性事務的SQL需要滿足冪等性。
  • INSERT語句要求必須包含主鍵，且不能是自增主鍵。
  • UPDATE語句要求冪等，不能是UPDATE xxx SET x=x+1
  • DELETE語句無要求。

開發(fā)示例

// 1. 配置SoftTransactionConfiguration
SoftTransactionConfiguration transactionConfig = new SoftTransactionConfiguration(dataSource);
// 配置相關(guān)請看后面的備注
transactionConfig.setXXX();

// 2. 初始化SoftTransactionManager
SoftTransactionManager transactionManager = new SoftTransactionManager(transactionConfig);
transactionManager.init();

// 3. 獲取BEDSoftTransaction
BEDSoftTransaction transaction = (BEDSoftTransaction) transactionManager.getTransaction(SoftTransactionType.BestEffortsDelivery);

// 4. 開啟事務
transaction.begin(connection);

// 5. 執(zhí)行JDBC
/* 
    code here
*/
* 
// 6.關(guān)閉事務
transaction.end();

備注：SoftTransactionConfiguration支持的配置以及含義請參考sharding-jdbc使用指南?事務支持，這段開發(fā)示例的代碼也摘自這里；也可參考sharding-jdbc-transaction模塊中com.dangdang.ddframe.rdb.transaction.soft.integrate.SoftTransactionTest如何使用柔性事務，但是這里的代碼需要稍作修改，否則只是普通的執(zhí)行邏輯，不是sharding-jdbc的執(zhí)行邏輯：

@Test
public void bedSoftTransactionTest() throws SQLException {
    SoftTransactionManager transactionManagerFactory = new SoftTransactionManager(getSoftTransactionConfiguration(getShardingDataSource()));
    // 初始化柔性事務管理器
    transactionManagerFactory.init();
    BEDSoftTransaction transactionManager = (BEDSoftTransaction) transactionManagerFactory.getTransaction(SoftTransactionType.BestEffortsDelivery);
    transactionManager.begin(getShardingDataSource().getConnection());
    // 執(zhí)行INSERT SQL（DML類型），如果執(zhí)行過程中異常，會在`BestEffortsDeliveryListener`中重試
    insert();
    transactionManager.end();
}

private void insert() {
    String dbSchema = "insert into transaction_test(id, remark) values (2, ?)";
    try (
            // 將.getConnection("db_trans", SQLType.DML)移除，這樣的話，得到的才是ShardingConnection 
            Connection conn = getShardingDataSource().getConnection();
            PreparedStatement preparedStatement = conn.prepareStatement(dbSchema)) {
        preparedStatement.setString(1, "JUST TEST IT .");
        preparedStatement.executeUpdate();
    } catch (final SQLException e) {
        e.printStackTrace();
    }
}

核心源碼分析

通過3. sharding-jdbc源碼之路由&執(zhí)行中對ExecutorEngine的分析可知，sharding-jdbc在執(zhí)行SQL前后，分別調(diào)用EventBusInstance.getInstance().post()提交了事件，那么調(diào)用EventBusInstance.getInstance().register()的地方，就是柔性事務處理的地方，通過查看源碼的調(diào)用關(guān)系可知，只有SoftTransactionManager.init()調(diào)用了EventBusInstance.getInstance().register()，所以柔性事務實現(xiàn)的核心在SoftTransactionManager這里；

柔性事務管理器

柔性事務實現(xiàn)在SoftTransactionManager中，核心源碼如下：

public final class SoftTransactionManager {

    // 柔性事務配置對象   
    @Getter
    private final SoftTransactionConfiguration transactionConfig;
    
    /**
     * Initialize B.A.S.E transaction manager.
     * @throws SQLException SQL exception
     */
    public void init() throws SQLException {
        // 初始化注冊最大努力送達型柔性事務監(jiān)聽器；
        EventBusInstance.getInstance().register(new BestEffortsDeliveryListener());
        if (TransactionLogDataSourceType.RDB == transactionConfig.getStorageType()) {
            // 如果事務日志數(shù)據(jù)源類型是關(guān)系型數(shù)據(jù)庫，則創(chuàng)建事務日志表transaction_log
            createTable();
        }
        // 內(nèi)嵌的最大努力送達型異步JOB任務，依賴當當開源的elastic-job
        if (transactionConfig.getBestEffortsDeliveryJobConfiguration().isPresent()) {
            new NestedBestEffortsDeliveryJobFactory(transactionConfig).init();
        }
    }
    
    // 從這里可知創(chuàng)建的事務日志表表名是transaction_log（所以需要保證每個庫中用戶沒有自定義創(chuàng)建transaction_log表）
    private void createTable() throws SQLException {
        String dbSchema = "CREATE TABLE IF NOT EXISTS `transaction_log` ("
                + "`id` VARCHAR(40) NOT NULL, "
                + "`transaction_type` VARCHAR(30) NOT NULL, "
                + "`data_source` VARCHAR(255) NOT NULL, "
                + "`sql` TEXT NOT NULL, "
                + "`parameters` TEXT NOT NULL, "
                + "`creation_time` LONG NOT NULL, "
                + "`async_delivery_try_times` INT NOT NULL DEFAULT 0, "
                + "PRIMARY KEY (`id`));";
        try (
                Connection conn = transactionConfig.getTransactionLogDataSource().getConnection();
                PreparedStatement preparedStatement = conn.prepareStatement(dbSchema)) {
            preparedStatement.executeUpdate();
        }
    }

從這段源碼可知，柔性事務的幾個重點如下，接下來一一根據(jù)源碼進行分析；

  • 事務日志存儲器；
  • 最大努力送達型事務監(jiān)聽器；
  • 異步送達JOB任務；

1.事務日志存儲器

柔性事務日志接口類為TransactionLogStorage.java，有兩個實現(xiàn)類：

  1. RdbTransactionLogStorage：關(guān)系型數(shù)據(jù)庫存儲柔性事務日志；
  2. MemoryTransactionLogStorage：內(nèi)存存儲柔性事務日志；

1.1.1事務日志核心接口

TransactionLogStorage中幾個重要接口在兩個實現(xiàn)類中的實現(xiàn)：

  • void add(TransactionLog)：Rdb實現(xiàn)就是把事務日志TransactionLog 插入到transaction_log表中，Memory實現(xiàn)就是把事務日志保存到ConcurrentHashMap中；
  • void remove(String id)：Rdb實現(xiàn)就是從transaction_log表中刪除事務日志，Memory實現(xiàn)從ConcurrentHashMap中刪除事務日志；
  • void increaseAsyncDeliveryTryTimes(String id)：異步增加送達重試次數(shù)，即TransactionLog中的asyncDeliveryTryTimes+1；Rdb實現(xiàn)就是update transaction_log表中async_delivery_try_times字段加1；Memory實現(xiàn)就是TransactionLog中重新給asyncDeliveryTryTimes賦值new AtomicInteger(transactionLog.getAsyncDeliveryTryTimes()).incrementAndGet()；
  • findEligibleTransactionLogs(): 查詢需要處理的事務日志，條件是：①異步處理次數(shù)async_delivery_try_times小于參數(shù)最大處里次數(shù)maxDeliveryTryTimes，②transaction_type是BestEffortsDelivery，③系統(tǒng)當前時間與事務日志的創(chuàng)建時間差要超過參數(shù)maxDeliveryTryDelayMillis，每次最多查詢參數(shù)size條；Rdb實現(xiàn)通過sql從transaction_log表中查詢，Memory實現(xiàn)遍歷ConcurrentHashMap匹配符合條件的TransactionLog；
  • boolean processData()：Rdb實現(xiàn)執(zhí)行TransactionLog中的sql，如果執(zhí)行過程中拋出異常，那么調(diào)用increaseAsyncDeliveryTryTimes()增加送達重試次數(shù)并拋出異常，如果執(zhí)行成功，刪除事務日志，并返回true；Memory實現(xiàn)直接返回false（因為processData()的目的是執(zhí)行TransactionLog中的sql，而Memory類型無法觸及數(shù)據(jù)庫，所以返回false）

1.1.2事務日志存儲核心源碼

RdbTransactionLogStorage.java實現(xiàn)源碼：

public final class RdbTransactionLogStorage implements TransactionLogStorage {
    
    private final DataSource dataSource;
    
    @Override
    public void add(final TransactionLog transactionLog) {
        // 保存事務日志到rdb中的sql
        String sql = "INSERT INTO `transaction_log` (`id`, `transaction_type`, `data_source`, `sql`, `parameters`, `creation_time`) VALUES (?, ?, ?, ?, ?, ?);";
        try (
            Connection conn = dataSource.getConnection();
            PreparedStatement preparedStatement = conn.prepareStatement(sql)) {
            ... ...
            preparedStatement.executeUpdate();
        } catch (final SQLException ex) {
            throw new TransactionLogStorageException(ex);
        }
    }
    
    @Override
    public void remove(final String id) {
        // 根據(jù)id刪除事務日志的sql
        String sql = "DELETE FROM `transaction_log` WHERE `id`=?;";
        try (
            Connection conn = dataSource.getConnection();
            PreparedStatement preparedStatement = conn.prepareStatement(sql)) {
            preparedStatement.setString(1, id);
            preparedStatement.executeUpdate();
        } catch (final SQLException ex) {
            throw new TransactionLogStorageException(ex);
        }
    }
    
    @Override
    public List<TransactionLog> findEligibleTransactionLogs(final int size, final int maxDeliveryTryTimes, final long maxDeliveryTryDelayMillis) {
        List<TransactionLog> result = new ArrayList<>(size);
        // 執(zhí)行該sql查詢需要處理的事務日志，最多取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()) {
            try (PreparedStatement preparedStatement = conn.prepareStatement(sql)) {
                ... ...
                preparedStatement.setLong(3, System.currentTimeMillis() - maxDeliveryTryDelayMillis);
                ... ...
            }
        } catch (final SQLException ex) {
            throw new TransactionLogStorageException(ex);
        }
        return result;
    }
    
    @Override
    public void increaseAsyncDeliveryTryTimes(final String id) {
        // 更新處理次數(shù)+1
        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);
        }
    }
    
    @Override
    public boolean processData(final Connection connection, final TransactionLog transactionLog, final int maxDeliveryTryTimes) {
        try (
            Connection conn = connection;
            // 執(zhí)行TransactionLog中的sql
            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) {
            如果拋出異常，表示執(zhí)行sql失敗，那么把增加處理次數(shù)并把異常拋出去；
            increaseAsyncDeliveryTryTimes(transactionLog.getId());
            throw new TransactionCompensationException(ex);
        }
        // 如果沒有拋出異常，表示執(zhí)行sql成功，那么刪除該事務日志；
        remove(transactionLog.getId());
        return true;
    }
}

1.1.3事務日志存儲樣例

id transction_type data_source sql parameters creation_time async_delivery_try_times
85c141c4-1b8f-4e54-9010-0cc661bb1864 BestEffortsDelivery db_trans insert into transaction_test(id, remark) values (3, ?) ["TEST BY AFEI."] 1517899200989 0

transaction_log中存儲的事務日志樣例：

id transction_type data_source sql parameters creation_time async_delivery_try_times
85c141c4-1b8f-4e54-9010-0cc661bb1864 BestEffortsDelivery db_trans insert into transaction_test(id, remark) values (3, ?) ["TEST BY AFEI."] 1517899200989 0

1.2最大努力送達型事務監(jiān)聽器

核心源碼如下：

/**
 * Best efforts delivery B.A.S.E transaction listener.
 * 
 * @author zhangliang
 */
@Slf4j
public final class BestEffortsDeliveryListener {
    
    @Subscribe
    @AllowConcurrentEvents
    // 從方法可知，只監(jiān)聽DML執(zhí)行事件（DML即數(shù)據(jù)維護語言，包括INSERT, UPDATE, DELETE）
    public void listen(final DMLExecutionEvent event) {
        // 判斷是否需要繼續(xù)，判斷邏輯為：事務存在，并且是BestEffortsDelivery類型事務
        if (!isProcessContinuously()) {
            return;
        }
        // 從柔性事務管理器中得到柔性事務配置
        SoftTransactionConfiguration transactionConfig = SoftTransactionManager.getCurrentTransactionConfiguration().get();
        // 得到配置的柔性事務存儲器
        TransactionLogStorage transactionLogStorage = TransactionLogStorageFactory.createTransactionLogStorage(transactionConfig.buildTransactionLogDataSource());
        // 這里肯定是最大努力送達型事務（如果不是BEDSoftTransaction，isProcessContinuously()就是false）
        BEDSoftTransaction bedSoftTransaction = (BEDSoftTransaction) SoftTransactionManager.getCurrentTransaction().get();
        // 根據(jù)事件類型做不同處理
        switch (event.getEventExecutionType()) {
            case BEFORE_EXECUTE:
                // 如果執(zhí)行前事件，那么先保存事務日志；
                //TODO for batch SQL need split to 2-level records
                transactionLogStorage.add(new TransactionLog(event.getId(), bedSoftTransaction.getTransactionId(), bedSoftTransaction.getTransactionType(), 
                        event.getDataSource(), event.getSql(), event.getParameters(), System.currentTimeMillis(), 0));
                return;
            case EXECUTE_SUCCESS: 
                // 如果執(zhí)行成功事件，那么刪除事務日志；
                transactionLogStorage.remove(event.getId());
                return;
            case EXECUTE_FAILURE: 
                boolean deliverySuccess = false;
                // 如果執(zhí)行成功事件，最大努力送達型最多嘗試3次（可配置，SoftTransactionConfiguration中的參數(shù)syncMaxDeliveryTryTimes）；
                for (int i = 0; i < transactionConfig.getSyncMaxDeliveryTryTimes(); i++) {
                    // 如果在該Listener中執(zhí)行成功，那么返回，不需要再嘗試
                    if (deliverySuccess) {
                        return;
                    }
                    boolean isNewConnection = false;
                    Connection conn = null;
                    PreparedStatement preparedStatement = null;
                    try {
                        conn = bedSoftTransaction.getConnection().getConnection(event.getDataSource(), SQLType.DML);
                        // 通過執(zhí)行"select 1"判斷conn是否是有效的數(shù)據(jù)庫連接；如果不是有效的數(shù)據(jù)庫連接，釋放掉并重新獲取一個數(shù)據(jù)庫連接；
                        if (!isValidConnection(conn)) {
                            bedSoftTransaction.getConnection().release(conn);
                            conn = bedSoftTransaction.getConnection().getConnection(event.getDataSource(), SQLType.DML);
                            isNewConnection = true;
                        }
                        preparedStatement = conn.prepareStatement(event.getSql());
                        //TODO for batch event need split to 2-level records
                        for (int parameterIndex = 0; parameterIndex < event.getParameters().size(); parameterIndex++) {
                            preparedStatement.setObject(parameterIndex + 1, event.getParameters().get(parameterIndex));
                        }
                        // 因為只監(jiān)控DML，所以調(diào)用executeUpdate()
                        preparedStatement.executeUpdate();
                        // executeUpdate()后能執(zhí)行到這里，說明執(zhí)行成功；根據(jù)id刪除事務日志；
                        deliverySuccess = true;
                        transactionLogStorage.remove(event.getId());
                    } catch (final SQLException ex) {
                        // 如果sql執(zhí)行有異常，那么輸出error日志
                        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源碼總結(jié)：

  • 執(zhí)行前，插入事務日志；
  • 執(zhí)行成功，則刪除事務日志；
  • 執(zhí)行失敗，則最大努力嘗試syncMaxDeliveryTryTimes次；

1.3 異步送達JOB任務

  • 部署用于存儲事務日志的數(shù)據(jù)庫。
  • 部署用于異步作業(yè)使用的zookeeper。
  • 配置YAML文件,參照示例文件config.yaml。
  • 下載并解壓文件sharding-jdbc-transaction-async-job-$VERSION.tar，通過start.sh腳本啟動異步作業(yè)。

異步送達JOB任務基于elastic-job，所以需要部署zookeeper；

異步送達JOB任務將在下一張詳細講解10. sharding-jdbc源碼之異步送達JOB；

最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容