溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
本篇內容主要講解“Java @GlobalLock注解怎么使用”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實用性強。下面就讓小編來帶大家學習“Java @GlobalLock注解怎么使用”吧!
對于某條數據進行更新操作，如果全局事務正在進行，當某個本地事務需要更新該數據時，需要使用@GlobalLock確保其不會對全局事務正在操作的數據進行修改。防止的本地事務對全局事務的數據臟寫。如果和select for update組合使用，還可以起到防止臟讀的效果。
首先我們知道，seata的AT模式是二段提交的，而且AT模式能夠做到事務ACID四種特性中的A原子性和D持久性，默認情況下隔離級別也只能保證在讀未提交
那么為了保證原子性，在全局事務未提交之前，其中被修改的數據會被加上全局鎖，保證不再會被其他全局事務修改。
但是全局鎖僅僅能防止全局事務對一個上鎖的數據再次進行修改，在很多業務場景中我們是沒有跨系統的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…)
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獲取
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注解怎么使用”有了更深的了解，不妨來實際操作一番吧！這里是億速云網站，更多相關內容可以進入相關頻道進行查詢，關注我們，繼續學習！
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
