中文字幕av专区_日韩电影在线播放_精品国产精品久久一区免费式_av在线免费观看网站

溫馨提示×

溫馨提示×

您好,登錄后才能下訂單哦!

密碼登錄×
登錄注冊×
其他方式登錄
點擊 登錄注冊 即表示同意《億速云用戶服務條款》

Java?@GlobalLock注解怎么使用

發布時間:2022-11-21 09:22:36 來源:億速云 閱讀:132 作者:iii 欄目:開發技術

本篇內容主要講解“Java @GlobalLock注解怎么使用”,感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷,實用性強。下面就讓小編來帶大家學習“Java @GlobalLock注解怎么使用”吧!

GlobalLock的作用

對于某條數據進行更新操作,如果全局事務正在進行,當某個本地事務需要更新該數據時,需要使用@GlobalLock確保其不會對全局事務正在操作的數據進行修改。防止的本地事務對全局事務的數據臟寫。如果和select for update組合使用,還可以起到防止臟讀的效果。

全局鎖

首先我們知道,seata的AT模式是二段提交的,而且AT模式能夠做到事務ACID四種特性中的A原子性和D持久性,默認情況下隔離級別也只能保證在讀未提交

那么為了保證原子性,在全局事務未提交之前,其中被修改的數據會被加上全局鎖,保證不再會被其他全局事務修改。

為什么要使用GlobalLock

但是全局鎖僅僅能防止全局事務對一個上鎖的數據再次進行修改,在很多業務場景中我們是沒有跨系統的rpc調用的,通常是不會加分布式事務的。

例如有分布式事務執行完畢A系統的業務邏輯,正在繼續執行B系統邏輯,并且A系統事務已經提交。此時A系統一個本地的spring事務去與分布式事務修改同一行數據,是可以正常修改的

由于本地的spring事務并不受seata的全局鎖控制容易導致臟寫,即全局事務修改數據后,還未提交,數據又被本地事務改掉了。這很容易發生數據出錯的問題,而且十分有可能導致全局事務回滾時發現 數據已經dirty(與uodoLog中的beforeImage不同)。那么就會回滾失敗,進而導致全局鎖無法釋放,后續的操作無法進行下去。也是比較嚴重的問題。

一種解決辦法就是,針對所有相關操作都加上AT全局事務,但這顯然是沒必要的,因為全局事務意味者需要與seata-server進行通信,創建全局事務,注冊分支事務,記錄undoLog,判斷鎖沖突,注冊鎖。

那么對于不需要跨系統,跨庫的的業務來說,使用GlobalTransactional實在是有點浪費了

那么更加輕量的GlobalLock就能夠發揮作用了,其只需要判斷本地的修改是否與全局鎖沖突就夠了

工作原理

加上@GlobalLock之后,會進入切面

io.seata.spring.annotation.GlobalTransactionalInterceptor#invoke

進而進入這個方法,處理全局鎖

    Object handleGlobalLock(final MethodInvocation methodInvocation,
        final GlobalLock globalLockAnno) throws Throwable {
        return globalLockTemplate.execute(new GlobalLockExecutor() {
            @Override
            public Object execute() throws Throwable {
                return methodInvocation.proceed();
            }
            @Override
            public GlobalLockConfig getGlobalLockConfig() {
                GlobalLockConfig config = new GlobalLockConfig();
                config.setLockRetryInternal(globalLockAnno.lockRetryInternal());
                config.setLockRetryTimes(globalLockAnno.lockRetryTimes());
                return config;
            }
        });
    }

進入execute方法

public Object execute(GlobalLockExecutor executor) throws Throwable {
        boolean alreadyInGlobalLock = RootContext.requireGlobalLock();
        if (!alreadyInGlobalLock) {
            RootContext.bindGlobalLockFlag();
        }
        // set my config to config holder so that it can be access in further execution
        // for example, LockRetryController can access it with config holder
        GlobalLockConfig myConfig = executor.getGlobalLockConfig();
        GlobalLockConfig previousConfig = GlobalLockConfigHolder.setAndReturnPrevious(myConfig);
        try {
            return executor.execute();
        } finally {
            // only unbind when this is the root caller.
            // otherwise, the outer caller would lose global lock flag
            if (!alreadyInGlobalLock) {
                RootContext.unbindGlobalLockFlag();
            }
            // if previous config is not null, we need to set it back
            // so that the outer logic can still use their config
            if (previousConfig != null) {
                GlobalLockConfigHolder.setAndReturnPrevious(previousConfig);
            } else {
                GlobalLockConfigHolder.remove();
            }
        }
    }
}

先判斷當前是否已經在globalLock范圍之內,如果已經在范圍之內,那么把上層的配置取出來,用新的配置替換,并在方法執行完畢時候,釋放鎖,或者將配置替換成之前的上層配置

如果開啟全局鎖,會在threadLocal put一個標記

    //just put something not null
CONTEXT_HOLDER.put(KEY_GLOBAL_LOCK_FLAG, VALUE_GLOBAL_LOCK_FLAG);

開始執行業務方法

那么加上相關GlobalLock標記的和普通方法的區別在哪里?

我們都知道,seata會對數據庫連接做代理,在生成PreparedStatement時會進入

io.seata.rm.datasource.AbstractConnectionProxy#prepareStatement(java.lang.String)

  @Override
    public PreparedStatement prepareStatement(String sql) throws SQLException {
        String dbType = getDbType();
        // support oracle 10.2+
        PreparedStatement targetPreparedStatement = null;
        if (BranchType.AT == RootContext.getBranchType()) {
            List<SQLRecognizer> sqlRecognizers = SQLVisitorFactory.get(sql, dbType);
            if (sqlRecognizers != null && sqlRecognizers.size() == 1) {
                SQLRecognizer sqlRecognizer = sqlRecognizers.get(0);
                if (sqlRecognizer != null && sqlRecognizer.getSQLType() == SQLType.INSERT) {
                    TableMeta tableMeta = TableMetaCacheFactory.getTableMetaCache(dbType).getTableMeta(getTargetConnection(),
                            sqlRecognizer.getTableName(), getDataSourceProxy().getResourceId());
                    String[] pkNameArray = new String[tableMeta.getPrimaryKeyOnlyName().size()];
                    tableMeta.getPrimaryKeyOnlyName().toArray(pkNameArray);
                    targetPreparedStatement = getTargetConnection().prepareStatement(sql,pkNameArray);
                }
            }
        }
        if (targetPreparedStatement == null) {
            targetPreparedStatement = getTargetConnection().prepareStatement(sql);
        }
        return new PreparedStatementProxy(this, targetPreparedStatement, sql);
    }

這里顯然不會進入AT模式的邏輯,那么直接通過真正的數據庫連接,生成PreparedStatement,再使用PreparedStatementProxy進行包裝,代理增強

在使用PreparedStatementProxy執行sql時,會進入seata定義的一些邏輯

 public boolean execute() throws SQLException {
        return ExecuteTemplate.execute(this, (statement, args) -> statement.execute());
    }

最終來到

io.seata.rm.datasource.exec.ExecuteTemplate#execute(java.util.List<io.seata.sqlparser.SQLRecognizer>, io.seata.rm.datasource.StatementProxy, io.seata.rm.datasource.exec.StatementCallback<T,S>, java.lang.Object&hellip;)

   public static <T, S extends Statement> T execute(List<SQLRecognizer> sqlRecognizers,
                                                     StatementProxy<S> statementProxy,
                                                     StatementCallback<T, S> statementCallback,
                                                     Object... args) throws SQLException {
        if (!RootContext.requireGlobalLock() && BranchType.AT != RootContext.getBranchType()) {
            // Just work as original statement
            return statementCallback.execute(statementProxy.getTargetStatement(), args);
        }
        String dbType = statementProxy.getConnectionProxy().getDbType();
        if (CollectionUtils.isEmpty(sqlRecognizers)) {
            sqlRecognizers = SQLVisitorFactory.get(
                    statementProxy.getTargetSQL(),
                    dbType);
        }
        Executor<T> executor;
        if (CollectionUtils.isEmpty(sqlRecognizers)) {
            executor = new PlainExecutor<>(statementProxy, statementCallback);
        } else {
            if (sqlRecognizers.size() == 1) {
                SQLRecognizer sqlRecognizer = sqlRecognizers.get(0);
                switch (sqlRecognizer.getSQLType()) {
                    case INSERT:
                        executor = EnhancedServiceLoader.load(InsertExecutor.class, dbType,
                                new Class[]{StatementProxy.class, StatementCallback.class, SQLRecognizer.class},
                                new Object[]{statementProxy, statementCallback, sqlRecognizer});
                        break;
                    case UPDATE:
                        executor = new UpdateExecutor<>(statementProxy, statementCallback, sqlRecognizer);
                        break;
                    case DELETE:
                        executor = new DeleteExecutor<>(statementProxy, statementCallback, sqlRecognizer);
                        break;
                    case SELECT_FOR_UPDATE:
                        executor = new SelectForUpdateExecutor<>(statementProxy, statementCallback, sqlRecognizer);
                        break;
                    default:
                        executor = new PlainExecutor<>(statementProxy, statementCallback);
                        break;
                }
            } else {
                executor = new MultiExecutor<>(statementProxy, statementCallback, sqlRecognizers);
            }
        }
        T rs;
        try {
            rs = executor.execute(args);
        } catch (Throwable ex) {
            if (!(ex instanceof SQLException)) {
                // Turn other exception into SQLException
                ex = new SQLException(ex);
            }
            throw (SQLException) ex;
        }
        return rs;
    }

如果當前線程不需要鎖并且不不在AT模式的分支事務下,直接使用原生的preparedStatement執行就好了

這里四種操作,通過不同的接口去執行,接口又有多種不同的數據庫類型實現

插入分為不同的數據庫類型,通過spi獲取

Java?@GlobalLock注解怎么使用

seata提供了三種數據庫的實現,

update,delete,select三種沒有多個實現類

他們在執行時都會執行父類的方法

io.seata.rm.datasource.exec.AbstractDMLBaseExecutor#executeAutoCommitTrue

   protected T executeAutoCommitTrue(Object[] args) throws Throwable {
        ConnectionProxy connectionProxy = statementProxy.getConnectionProxy();
        try {
            connectionProxy.changeAutoCommit();
            return new LockRetryPolicy(connectionProxy).execute(() -> {
                T result = executeAutoCommitFalse(args);
                connectionProxy.commit();
                return result;
            });
        } catch (Exception e) {
            // when exception occur in finally,this exception will lost, so just print it here
            LOGGER.error("execute executeAutoCommitTrue error:{}", e.getMessage(), e);
            if (!LockRetryPolicy.isLockRetryPolicyBranchRollbackOnConflict()) {
                connectionProxy.getTargetConnection().rollback();
            }
            throw e;
        } finally {
            connectionProxy.getContext().reset();
            connectionProxy.setAutoCommit(true);
        }
    }

全局鎖的策略, 是在一個while(true)循環里不斷執行

 protected <T> T doRetryOnLockConflict(Callable<T> callable) throws Exception {
            LockRetryController lockRetryController = new LockRetryController();
            while (true) {
                try {
                    return callable.call();
                } catch (LockConflictException lockConflict) {
                    onException(lockConflict);
                    lockRetryController.sleep(lockConflict);
                } catch (Exception e) {
                    onException(e);
                    throw e;
                }
            }
        }

如果出現異常是LockConflictException,進入sleep

public void sleep(Exception e) throws LockWaitTimeoutException {
        if (--lockRetryTimes < 0) {
            throw new LockWaitTimeoutException("Global lock wait timeout", e);
        }
        try {
            Thread.sleep(lockRetryInternal);
        } catch (InterruptedException ignore) {
        }
    }

這兩個變量就是@GlobalLock注解的兩個配置,一個是重試次數,一個重試之間的間隔時間。

繼續就是執行數據庫更新操作

io.seata.rm.datasource.exec.AbstractDMLBaseExecutor#executeAutoCommitFalse

發現這里也會生成,undoLog,beforeImage和afterImage,其實想想,在GlobalLock下,是沒必要生成undoLog的。但是現有邏輯確實要生成,這個seata后續應該會優化。

protected T executeAutoCommitFalse(Object[] args) throws Exception {
        if (!JdbcConstants.MYSQL.equalsIgnoreCase(getDbType()) && isMultiPk()) {
            throw new NotSupportYetException("multi pk only support mysql!");
        }
        TableRecords beforeImage = beforeImage();
        T result = statementCallback.execute(statementProxy.getTargetStatement(), args);
        TableRecords afterImage = afterImage(beforeImage);
        prepareUndoLog(beforeImage, afterImage);
        return result;
    }

生成beforeImage和aferImage的邏輯也比較簡單。分別在執行更新前,查詢數據庫,和更新后查詢數據庫

可見記錄undoLog是十分影響性能的,查詢就多了兩次,如果undoLog入庫還要再多一次入庫操作。

再看prepareUndoLog

 protected void prepareUndoLog(TableRecords beforeImage, TableRecords afterImage) throws SQLException {
        if (beforeImage.getRows().isEmpty() && afterImage.getRows().isEmpty()) {
            return;
        }
        if (SQLType.UPDATE == sqlRecognizer.getSQLType()) {
            if (beforeImage.getRows().size() != afterImage.getRows().size()) {
                throw new ShouldNeverHappenException("Before image size is not equaled to after image size, probably because you updated the primary keys.");
            }
        }
        ConnectionProxy connectionProxy = statementProxy.getConnectionProxy();
        TableRecords lockKeyRecords = sqlRecognizer.getSQLType() == SQLType.DELETE ? beforeImage : afterImage;
        String lockKeys = buildLockKey(lockKeyRecords);
        if (null != lockKeys) {
            connectionProxy.appendLockKey(lockKeys);

            SQLUndoLog sqlUndoLog = buildUndoItem(beforeImage, afterImage);
            connectionProxy.appendUndoLog(sqlUndoLog);
        }
    }

將lockKeys,和undoLog,暫時記錄在connectionProxy中,也就是說至此還沒有將uodoLog記錄到數據庫,也沒有判斷全局鎖,這些事情都留到了事務提交

io.seata.rm.datasource.ConnectionProxy#doCommit

 private void doCommit() throws SQLException {
        if (context.inGlobalTransaction()) {
            processGlobalTransactionCommit();
        } else if (context.isGlobalLockRequire()) {
            processLocalCommitWithGlobalLocks();
        } else {
            targetConnection.commit();
        }
    }

進入io.seata.rm.datasource.ConnectionProxy#processLocalCommitWithGlobalLocks

這個 方法很簡單就是首先進行鎖的檢查,并沒有我想象中的加索全局事務。

 private void processLocalCommitWithGlobalLocks() throws SQLException {
        checkLock(context.buildLockKeys());
        try {
            targetConnection.commit();
        } catch (Throwable ex) {
            throw new SQLException(ex);
        }
        context.reset();
    }

也就是說,使用GlobalLock會對全局鎖檢測,但是并不會對記錄加全局鎖。但是配合全局事務這樣已經能夠保證全局事務的原子性了。可見GlobalLock還是要和本地事務組合一起使用的,這樣才能保證,GlobalLock執行完畢本地事務未提交的數據不會被別的本地事務/分布式事務修改掉。

到此,相信大家對“Java @GlobalLock注解怎么使用”有了更深的了解,不妨來實際操作一番吧!這里是億速云網站,更多相關內容可以進入相關頻道進行查詢,關注我們,繼續學習!

向AI問一下細節

免責聲明:本站發布的內容(圖片、視頻和文字)以原創、轉載和分享為主,文章觀點不代表本網站立場,如果涉及侵權請聯系站長郵箱:is@yisu.com進行舉報,并提供相關證據,一經查實,將立刻刪除涉嫌侵權內容。

AI

长岭县| 高唐县| 安溪县| 囊谦县| 福泉市| 通许县| 繁峙县| 霍城县| 武陟县| 保定市| 宁陵县| 西宁市| 沅陵县| 和田县| 富阳市| 雅安市| 襄汾县| 郸城县| 钦州市| 革吉县| 静乐县| 巴彦淖尔市| 玉山县| 神农架林区| 青川县| 保山市| 云南省| 凌云县| 嘉兴市| 惠安县| 延吉市| 阳高县| 北海市| 乌拉特前旗| 临江市| 伽师县| 平阳县| 剑阁县| 栾城县| 九江县| 永川市|