快捷導(dǎo)航

sharding-jdbc中的事務(wù)詳細(xì)解讀

更新時(shí)間：2023年12月14日 10:41:50 作者：那個(gè)天真的人

這篇文章主要介紹了sharding-jdbc中的事務(wù)詳細(xì)解讀,sharding-jdbc在分庫分表方面提供了很大的便利性,在使用DB的時(shí)候,通常都會涉及到事務(wù)這個(gè)概念,而在分庫分表的環(huán)境上再加上事務(wù),就會使事情變得復(fù)雜起來,需要的朋友可以參考下

序言

sharding-jdbc在分庫分表方面提供了很大的便利性

在使用DB的時(shí)候，通常都會涉及到事務(wù)這個(gè)概念，而在分庫分表的環(huán)境上再加上事務(wù)，就會使事情變得復(fù)雜起來。

本章試圖剖析sharding-jdbc在事務(wù)方面的解決思路。

傳統(tǒng)事務(wù)回顧

傳統(tǒng)的事務(wù)模型如下：

Connection conn = getConnection();
try{
    Statement stmt1 = conn.parpareStatement(sql1);
    stmt1.executeUpdate();
    Statement stmt2 = conn.parpareStatement(sql2);
    stmt2.executeUpdate();
    conn.commit();
}catch(Exception e){
    conn.rollback();
}

對于同一個(gè)連接，可以執(zhí)行多條sql語句，任何一條語句出現(xiàn)錯(cuò)誤的時(shí)候，整個(gè)操作流程都可以回滾，從而達(dá)到事務(wù)的原子操作。

再來看最基本的spring事務(wù)操作：

class ServiceA(){
   public void updateA(){...}
}
class ServiceB(){
    public void updateB(){...}
}
@Transactional
class ServiceC(){
    public void updateC(){
        serviceA.updateA();
        serviceB.updateB();
    }
}

我們知道，當(dāng)updateC執(zhí)行的時(shí)候，不管是updateA還是updateB出現(xiàn)了異常，updateC都可以整體回滾，達(dá)到原子操作的效果，其主要原因是updateA和updateB共享了同一個(gè)Connection，這是spring底層通過ThreadLocal緩存了Connection實(shí)現(xiàn)的。

以上介紹的這兩種情況都只是針對單庫單表的原子操作，事務(wù)的實(shí)現(xiàn)并不難理解，那么在跨庫的情況下，sharding-jdbc又是如何解決事務(wù)問題的呢？

shrading-jdbc之弱事務(wù)

在官方文檔中，針對弱事務(wù)有如下三點(diǎn)說明：

完全支持非跨庫事務(wù)，例如：僅分表，或分庫但是路由的結(jié)果在單庫中。
完全支持因邏輯異常導(dǎo)致的跨庫事務(wù)。例如：同一事務(wù)中，跨兩個(gè)庫更新。更新完畢后，拋出空指針，則兩個(gè)庫的內(nèi)容都能回滾。
不支持因網(wǎng)絡(luò)、硬件異常導(dǎo)致的跨庫事務(wù)。例如：同一事務(wù)中，跨兩個(gè)庫更新，更新完畢后、未提交之前，第一個(gè)庫死機(jī)，則只有第二個(gè)庫數(shù)據(jù)提交。

為了理解以上幾點(diǎn)，我們來看看sharding-jdbc默認(rèn)是如何處理事務(wù)的。

這里寫圖片描述

這是一個(gè)非常常見的處理模式，一個(gè)總連接處理了多條sql語句，最后一次性提交整個(gè)事務(wù)，每一條sql語句可能會分為多條子sql分庫分表去執(zhí)行，這意味著底層可能會關(guān)聯(lián)多個(gè)真正的數(shù)據(jù)庫連接，我們先來看看如果一切正常，commit會如何去處理。

public abstract class AbstractConnectionAdapter extends AbstractUnsupportedOperationConnection {
    @Override
    public final void commit() throws SQLException {
        Collection<SQLException> exceptions = new LinkedList<>();
        for (Connection each : cachedConnections.values()) {
            try {
                each.commit();
            } catch (final SQLException ex) {
                exceptions.add(ex);
            }
        }
        throwSQLExceptionIfNecessary(exceptions);
    }
}

引擎會遍歷底層所有真正的數(shù)據(jù)庫連接，一個(gè)個(gè)進(jìn)行commit操作，如果任何一個(gè)出現(xiàn)了異常，直接捕獲異常，但是也只是捕獲而已，然后接著下一個(gè)連接的commit，這也就很好的說明了，如果在執(zhí)行任何一條sql語句出現(xiàn)了異常，整個(gè)操作是可以原子性回滾的，因?yàn)榇藭r(shí)所有連接都不會執(zhí)行commit，但如果已經(jīng)到了commit這一步的話，如果有連接commit失敗了，是不會影響到其他連接的。

sharding-jdbc之柔性事務(wù)

sharding-jdbc的弱事務(wù)并不是完美的，有時(shí)可能會導(dǎo)致數(shù)據(jù)的一致性問題，所以針對某些特定的場景，又提出了柔性事務(wù)的概念。先來看一張官方的說明圖：

這里寫圖片描述

這里想表達(dá)兩個(gè)意思：

1. 對于sql的執(zhí)行，在執(zhí)行前記錄日志，如果執(zhí)行成功，把日志刪除，如果執(zhí)行失敗，重試一定次數(shù)（如果未達(dá)到最大嘗試次數(shù)便執(zhí)行成功了，一樣刪除日志）。

2. 異步任務(wù)不斷掃描執(zhí)行日志，如果重試次數(shù)未達(dá)到最大上限，嘗試重新執(zhí)行，如果執(zhí)行成功，刪除日志。

從上面兩點(diǎn)分析可以看出，由于采用的是重試的模式，也就是說同一條語句，是有可能被多次執(zhí)行的，所以官方提到了柔性事務(wù)的適用場景：

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

而且它還有一定的限制： SQL需要滿足冪等性，具體為：

INSERT語句要求必須包含主鍵，且不能是自增主鍵。
UPDATE語句要求冪等，不能是UPDATE xxx SET x=x+1
DELETE語句無要求。

在有了一個(gè)大概的了解之后，我們來更加深入的了解。

sharding-jdbc使用了google的EventBus事件模型，注冊了一個(gè)Listener，監(jiān)聽器對三種事件進(jìn)行了處理，如下代碼所示：

switch (event.getEventExecutionType()) {
            case BEFORE_EXECUTE:
                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 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));
                        }
                        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());
        }

以上代碼可以抽取為如下圖的描述：

這里寫圖片描述

監(jiān)聽器根據(jù)三種不同的事件類型對事務(wù)日志進(jìn)行不同的操作。有監(jiān)聽，必然就有事件的投遞，那么引擎是什么時(shí)候產(chǎn)生這些事件的呢？我們知道每一條sql語句拆分后有可能對應(yīng)多條子sql語句，而每一條子sql語句是單獨(dú)執(zhí)行的，執(zhí)行是封裝在一個(gè)內(nèi)部方法的：

private <T> T executeInternal(final SQLType sqlType, final BaseStatementUnit baseStatementUnit, final List<List<Object>> parameterSets, final ExecuteCallback<T> executeCallback, 
                          final boolean isExceptionThrown, final Map<String, Object> dataMap) throws Exception {
        synchronized (baseStatementUnit.getStatement().getConnection()) {
            T result;
            ExecutorExceptionHandler.setExceptionThrown(isExceptionThrown);
            ExecutorDataMap.setDataMap(dataMap);
            List<AbstractExecutionEvent> events = new LinkedList<>();
            if (parameterSets.isEmpty()) {
                events.add(getExecutionEvent(sqlType, baseStatementUnit, Collections.emptyList()));
            }
            for (List<Object> each : parameterSets) {
                events.add(getExecutionEvent(sqlType, baseStatementUnit, each));
            }
            for (AbstractExecutionEvent event : events) {
                EventBusInstance.getInstance().post(event);
            }
            try {
                result = executeCallback.execute(baseStatementUnit);
            } catch (final SQLException ex) {
                for (AbstractExecutionEvent each : events) {
                    each.setEventExecutionType(EventExecutionType.EXECUTE_FAILURE);
                    each.setException(Optional.of(ex));
                    EventBusInstance.getInstance().post(each);
                    ExecutorExceptionHandler.handleException(ex);
                }
                return null;
            }
            for (AbstractExecutionEvent each : events) {
                each.setEventExecutionType(EventExecutionType.EXECUTE_SUCCESS);
                EventBusInstance.getInstance().post(each);
            }
            return result;
        }
    }

以上代碼可以簡化為如下流程：

這里寫圖片描述

其實(shí)執(zhí)行流程比較簡單，但還有兩個(gè)重要的細(xì)節(jié)這里沒有體現(xiàn)：

當(dāng)使用柔性事務(wù)的時(shí)候，需要?jiǎng)?chuàng)建事務(wù)管理器，并獲取事務(wù)對象，調(diào)用事務(wù)對象的begin開始一個(gè)事務(wù)，在這一步，會強(qiáng)制設(shè)置連接的autoCommit=true，這會導(dǎo)致所有的sql語句執(zhí)時(shí)后立即提交，想想如果能回滾，那柔性事務(wù)也就失去了意義。
當(dāng)事務(wù)執(zhí)行begin時(shí)，會標(biāo)記當(dāng)前不拋出異常，這樣當(dāng)執(zhí)行sql語句有異常時(shí)，會生成相應(yīng)的EXECUTE_FAILURE事件，從而進(jìn)行事務(wù)日志處理，而不是往外拋出異常，當(dāng)事務(wù)結(jié)束時(shí)，調(diào)用事務(wù)對象的end方法，恢復(fù)異常的捕獲。

一個(gè)常見的代碼編寫模式如下（來自官方的demo)

private static void updateFailure(final DataSource dataSource) throws SQLException {
        String sql1 = "UPDATE t_order SET status='UPDATE_1' WHERE user_id=10 AND order_id=1000";
        String sql2 = "UPDATE t_order SET not_existed_column=1 WHERE user_id=1 AND order_id=?";
        String sql3 = "UPDATE t_order SET status='UPDATE_2' WHERE user_id=10 AND order_id=1000";
        SoftTransactionManager transactionManager = new SoftTransactionManager(getSoftTransactionConfiguration(dataSource));
        transactionManager.init();
        BEDSoftTransaction transaction = (BEDSoftTransaction) transactionManager.getTransaction(SoftTransactionType.BestEffortsDelivery);
        Connection conn = null;
        try {
            conn = dataSource.getConnection();
            transaction.begin(conn);
            PreparedStatement preparedStatement1 = conn.prepareStatement(sql1);
            PreparedStatement preparedStatement2 = conn.prepareStatement(sql2);
            preparedStatement2.setObject(1, 1000);
            PreparedStatement preparedStatement3 = conn.prepareStatement(sql3);
            preparedStatement1.executeUpdate();
            preparedStatement2.executeUpdate();
            preparedStatement3.executeUpdate();
        } finally {
            transaction.end();
            if (conn != null) {
                conn.close();
            }
        }
    }

看到這個(gè)編寫模式，你一定會想，如果我使用MyBatis和spring，這一切能否整合起來，這個(gè)話題有興趣大家可以去嘗試。