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

• 事務日誌儲存器；

• 最大努力送達型事務監聽器；

• 非同步送達JOB任務；

1.事務日誌儲存器

柔性事務日誌介面類為 TransactionLogStorage.java，有兩個實現類：

1. RdbTransactionLogStorage：關係型資料庫儲存柔性事務日誌；

2. MemoryTransactionLogStorage：記憶體儲存柔性事務日誌；

1.1.1事務日誌核心介面

TransactionLogStorage中幾個重要介面在兩個實現類中的實現：

• void add(TransactionLog)：Rdb實現就是把事務日誌TransactionLog 插入到 transaction_log表中，Memory實現就是把事務日誌儲存到 ConcurrentHashMap中；

• void remove(String id)：Rdb實現就是從 transaction_log表中刪除事務日誌，Memory實現從 ConcurrentHashMap中刪除事務日誌；

• void increaseAsyncDeliveryTryTimes(String id)：非同步增加送達重試次數，即TransactionLog中的asyncDeliveryTryTimes+1；Rdb實現就是update transaction_log表中 async_delivery_try_times欄位加1；Memory實現就是TransactionLog中重新給asyncDeliveryTryTimes賦值 newAtomicInteger(transactionLog.getAsyncDeliveryTryTimes()).incrementAndGet()；

• findEligibleTransactionLogs(): 查詢需要處理的事務日誌，條件是：① 非同步處理次數async_delivery_try_times小於引數最大處裡次數maxDeliveryTryTimes，② transaction_type是BestEffortsDelivery，③ 系統當前時間與事務日誌的建立時間差要超過引數maxDeliveryTryDelayMillis，每次最多查詢引數size條；Rdb實現透過sql從transaction_log表中查詢，Memory實現遍歷ConcurrentHashMap匹配符合條件的TransactionLog；

• boolean processData()：Rdb實現執行TransactionLog中的sql，如果執行過程中丟擲異常，那麼呼叫increaseAsyncDeliveryTryTimes()增加送達重試次數並丟擲異常，如果執行成功，刪除事務日誌，並傳回true；Memory實現直接傳回false（因為processData()的目的是執行TransactionLog中的sql，而Memory型別無法觸及資料庫，所以傳回false）

1.1.2事務日誌儲存核心原始碼

RdbTransactionLogStorage.java實現原始碼：

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) {
        // 根據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);
        // 執行該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) {
        // 更新處理次數+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;
            // 執行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) {
            如果丟擲異常，表示執行sql失敗，那麼把增加處理次數並把異常丟擲去；
            increaseAsyncDeliveryTryTimes(transactionLog.getId());
            throw new TransactionCompensationException(ex);
        }
        // 如果沒有丟擲異常，表示執行sql成功，那麼刪除該事務日誌；
        remove(transactionLog.getId());
        return true;
    }
}

1.1.3事務日誌儲存樣例

id	transction_type	data_source	sql	parameters	creation_time	asyncdeliverytry_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	asyncdeliverytry_times
85c141c4-1b8f-4e54-9010-0cc661bb1864	BestEffortsDelivery	db_trans	insert into transaction_test(id, remark) values (3, ?)	["TEST BY AFEI."]	1517899200989	0

1.2最大努力送達型事務監聽器

核心原始碼如下：

/**
 * Best efforts delivery B.A.S.E transaction listener.
 * 
 * @author zhangliang
 */
@Slf4j
public final class BestEffortsDeliveryListener {
    @Subscribe
    @AllowConcurrentEvents
    // 從方法可知，只監聽DML執行事件（DML即資料維護語言，包括INSERT, UPDATE, DELETE）
    public void listen(final DMLExecutionEvent event) {
        // 判斷是否需要繼續，判斷邏輯為：事務存在，並且是BestEffortsDelivery型別事務
        if (!isProcessContinuously()) {
            return;
        }
        // 從柔性事務管理器中得到柔性事務配置
        SoftTransactionConfiguration transactionConfig = SoftTransactionManager.getCurrentTransactionConfiguration().get();
        // 得到配置的柔性事務儲存器
        TransactionLogStorage transactionLogStorage = TransactionLogStorageFactory.createTransactionLogStorage(transactionConfig.buildTransactionLogDataSource());
        // 這裡肯定是最大努力送達型事務（如果不是BEDSoftTransaction，isProcessContinuously()就是false）
        BEDSoftTransaction bedSoftTransaction = (BEDSoftTransaction) SoftTransactionManager.getCurrentTransaction().get();
        // 根據事件型別做不同處理
        switch (event.getEventExecutionType()) {
            case BEFORE_EXECUTE:
                // 如果執行前事件，那麼先儲存事務日誌；
                //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: 
                // 如果執行成功事件，那麼刪除事務日誌；
                transactionLogStorage.remove(event.getId());
                return;
            case EXECUTE_FAILURE: 
                boolean deliverySuccess = false;
                // 如果執行成功事件，最大努力送達型最多嘗試3次（可配置，SoftTransactionConfiguration中的引數syncMaxDeliveryTryTimes）；
                for (int i = 0; i < transactionConfig.getSyncMaxDeliveryTryTimes(); i++) {
                    // 如果在該Listener中執行成功，那麼傳回，不需要再嘗試
                    if (deliverySuccess) {
                        return;
                    }
                    boolean isNewConnection = false;
                    Connection conn = null;
                    PreparedStatement preparedStatement = null;
                    try {
                        conn = bedSoftTransaction.getConnection().getConnection(event.getDataSource(), SQLType.DML);
                        // 透過執行"select 1"判斷conn是否是有效的資料庫連線；如果不是有效的資料庫連線，釋放掉並重新獲取一個資料庫連線；
                        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));
                        }
                        // 因為只監控DML，所以呼叫executeUpdate()
                        preparedStatement.executeUpdate();
                        // executeUpdate()後能執行到這裡，說明執行成功；根據id刪除事務日誌；
                        deliverySuccess = true;
                        transactionLogStorage.remove(event.getId());
                    } catch (final SQLException ex) {
                        // 如果sql執行有異常，那麼輸出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原始碼總結：

• 執行前，插入事務日誌；

• 執行成功，則刪除事務日誌；

• 執行失敗，則最大努力嘗試 syncMaxDeliveryTryTimes次；

1.3 非同步送達JOB任務

• 部署用於儲存事務日誌的資料庫。

• 部署用於非同步作業使用的zookeeper。

• 配置YAML檔案,參照示例檔案config.yaml。

• 下載並解壓檔案sharding-jdbc-transaction-async-job-$VERSION.tar，透過start.sh指令碼啟動非同步作業。

非同步送達JOB任務基於elastic-job，所以需要部署zookeeper；

END

阿飛哥

贊賞

長按二維碼向我轉賬

受蘋果公司新規定影響，微信 iOS 版的贊賞功能被關閉，可透過二維碼轉賬支援公眾號。

閱讀原文


微信掃一掃
使用小程式
即將開啟""小程式
取消
開啟