/*
|
* Copyright 1999-2019 Seata.io Group.
|
*
|
* Licensed under the Apache License, Version 2.0 (the "License");
|
* you may not use this file except in compliance with the License.
|
* You may obtain a copy of the License at
|
*
|
* http://www.apache.org/licenses/LICENSE-2.0
|
*
|
* Unless required by applicable law or agreed to in writing, software
|
* distributed under the License is distributed on an "AS IS" BASIS,
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
* See the License for the specific language governing permissions and
|
* limitations under the License.
|
*/
|
package io.seata.server.storage.db.lock;
|
|
import io.seata.common.exception.ShouldNeverHappenException;
|
import io.seata.common.loader.EnhancedServiceLoader;
|
import io.seata.common.loader.LoadLevel;
|
import io.seata.common.loader.Scope;
|
import io.seata.common.util.IOUtil;
|
import io.seata.common.util.StringUtils;
|
import io.seata.config.*;
|
import io.seata.core.constants.ConfigurationKeys;
|
import io.seata.core.constants.ServerTableColumnsName;
|
import io.seata.core.store.DistributedLockDO;
|
import io.seata.core.store.DistributedLocker;
|
import io.seata.core.store.db.DataSourceProvider;
|
import io.seata.core.store.db.sql.distributed.lock.DistributedLockSqlFactory;
|
import org.slf4j.Logger;
|
import org.slf4j.LoggerFactory;
|
|
import javax.sql.DataSource;
|
import java.sql.Connection;
|
import java.sql.PreparedStatement;
|
import java.sql.ResultSet;
|
import java.sql.SQLException;
|
import java.util.HashSet;
|
import java.util.Objects;
|
import java.util.Set;
|
|
import static io.seata.core.constants.ConfigurationKeys.DISTRIBUTED_LOCK_DB_TABLE;
|
|
/**
|
* @author chd
|
*/
|
@LoadLevel(name = "db", scope = Scope.SINGLETON)
|
public class DataBaseDistributedLocker implements DistributedLocker {
|
private static final Logger LOGGER = LoggerFactory.getLogger(DataBaseDistributedLocker.class);
|
|
private final String dbType;
|
|
private final String datasourceType;
|
|
private volatile String distributedLockTable;
|
|
private DataSource distributedLockDataSource;
|
|
private static final String LOCK_WAIT_TIMEOUT_MYSQL_MESSAGE = "try restarting transaction";
|
|
private static final int LOCK_WAIT_TIMEOUT_MYSQL_CODE = 1205;
|
|
private static final Set<Integer> IGNORE_MYSQL_CODE = new HashSet<>();
|
|
private static final Set<String> IGNORE_MYSQL_MESSAGE = new HashSet<>();
|
|
static {
|
IGNORE_MYSQL_CODE.add(LOCK_WAIT_TIMEOUT_MYSQL_CODE);
|
IGNORE_MYSQL_MESSAGE.add(LOCK_WAIT_TIMEOUT_MYSQL_MESSAGE);
|
}
|
|
/**
|
* whether the distribute lock demotion
|
* using for 1.5.0 only and will remove in 1.6.0
|
*/
|
@Deprecated
|
private volatile boolean demotion;
|
|
/**
|
* Instantiates a new Log store data base dao.
|
*/
|
public DataBaseDistributedLocker() {
|
Configuration configuration = ConfigurationFactory.getInstance();
|
|
distributedLockTable = configuration.getConfig(DISTRIBUTED_LOCK_DB_TABLE);
|
dbType = configuration.getConfig(ConfigurationKeys.STORE_DB_TYPE);
|
datasourceType = configuration.getConfig(ConfigurationKeys.STORE_DB_DATASOURCE_TYPE);
|
|
if (StringUtils.isBlank(distributedLockTable)) {
|
demotion = true;
|
ConfigurationCache.addConfigListener(DISTRIBUTED_LOCK_DB_TABLE, new ConfigurationChangeListener() {
|
@Override
|
public void onChangeEvent(ConfigurationChangeEvent event) {
|
String newValue = event.getNewValue();
|
if (StringUtils.isNotBlank(newValue)) {
|
distributedLockTable = newValue;
|
init();
|
demotion = false;
|
ConfigurationCache.removeConfigListener(DISTRIBUTED_LOCK_DB_TABLE, this);
|
}
|
}
|
});
|
|
LOGGER.error("The distribute lock table is not config, please create the target table and config it");
|
return;
|
}
|
|
init();
|
}
|
|
|
@Override
|
public boolean acquireLock(DistributedLockDO distributedLockDO) {
|
if (demotion) {
|
return true;
|
}
|
|
Connection connection = null;
|
boolean originalAutoCommit = false;
|
try {
|
connection = distributedLockDataSource.getConnection();
|
originalAutoCommit = connection.getAutoCommit();
|
connection.setAutoCommit(false);
|
|
DistributedLockDO lockFromDB = getDistributedLockDO(connection, distributedLockDO.getLockKey());
|
if (null == lockFromDB) {
|
boolean ret = insertDistribute(connection, distributedLockDO);
|
connection.commit();
|
return ret;
|
}
|
|
if (lockFromDB.getExpireTime() >= System.currentTimeMillis()) {
|
LOGGER.debug("the distribute lock for key :{} is holding by :{}, acquire lock failure.",
|
distributedLockDO.getLockKey(), lockFromDB.getLockValue());
|
connection.commit();
|
return false;
|
}
|
|
boolean ret = updateDistributedLock(connection, distributedLockDO);
|
connection.commit();
|
|
return ret;
|
} catch (SQLException ex) {
|
// ignore "Lock wait timeout exceeded; try restarting transaction"
|
// TODO: need nowait adaptation
|
if (!ignoreSQLException(ex)) {
|
LOGGER.error("execute acquire lock failure, key is: {}", distributedLockDO.getLockKey(), ex);
|
}
|
try {
|
if (connection != null) {
|
connection.rollback();
|
}
|
} catch (SQLException e) {
|
LOGGER.warn("rollback fail because of {}", e.getMessage(), e);
|
}
|
return false;
|
} finally {
|
try {
|
if (originalAutoCommit) {
|
connection.setAutoCommit(true);
|
}
|
IOUtil.close(connection);
|
} catch (SQLException ignore) { }
|
}
|
}
|
|
@Override
|
public boolean releaseLock(DistributedLockDO distributedLockDO) {
|
if (demotion) {
|
return true;
|
}
|
|
Connection connection = null;
|
boolean originalAutoCommit = false;
|
try {
|
connection = distributedLockDataSource.getConnection();
|
originalAutoCommit = connection.getAutoCommit();
|
connection.setAutoCommit(false);
|
|
DistributedLockDO distributedLockDOFromDB = getDistributedLockDO(connection, distributedLockDO.getLockKey());
|
if (null == distributedLockDOFromDB) {
|
throw new ShouldNeverHappenException("distributedLockDO would not be null when release distribute lock");
|
}
|
|
if (distributedLockDOFromDB.getExpireTime() >= System.currentTimeMillis()
|
&& !Objects.equals(distributedLockDOFromDB.getLockValue(), distributedLockDO.getLockValue())) {
|
if (LOGGER.isDebugEnabled()) {
|
LOGGER.debug("the distribute lock for key :{} is holding by :{}, skip the release lock.",
|
distributedLockDO.getLockKey(), distributedLockDOFromDB.getLockValue());
|
}
|
connection.commit();
|
return true;
|
}
|
|
distributedLockDO.setLockValue(StringUtils.SPACE);
|
distributedLockDO.setExpireTime(0L);
|
boolean ret = updateDistributedLock(connection, distributedLockDO);
|
|
connection.commit();
|
return ret;
|
} catch (SQLException ex) {
|
if (!ignoreSQLException(ex)) {
|
LOGGER.error("execute release lock failure, key is: {}", distributedLockDO.getLockKey(), ex);
|
}
|
|
try {
|
if (connection != null) {
|
connection.rollback();
|
}
|
} catch (SQLException e) {
|
LOGGER.warn("rollback fail because of {}", e.getMessage(), e);
|
}
|
return false;
|
} finally {
|
try {
|
if (originalAutoCommit) {
|
connection.setAutoCommit(true);
|
}
|
IOUtil.close(connection);
|
} catch (SQLException ignore) { }
|
}
|
}
|
|
protected DistributedLockDO getDistributedLockDO(Connection connection, String key) throws SQLException {
|
try (PreparedStatement pst = connection.prepareStatement(DistributedLockSqlFactory.getDistributedLogStoreSql(dbType)
|
.getSelectDistributeForUpdateSql(distributedLockTable))) {
|
|
pst.setString(1, key);
|
ResultSet resultSet = pst.executeQuery();
|
|
if (resultSet.next()) {
|
DistributedLockDO distributedLock = new DistributedLockDO();
|
distributedLock.setExpireTime(resultSet.getLong(ServerTableColumnsName.DISTRIBUTED_LOCK_EXPIRE));
|
distributedLock.setLockValue(resultSet.getString(ServerTableColumnsName.DISTRIBUTED_LOCK_VALUE));
|
distributedLock.setLockKey(key);
|
return distributedLock;
|
}
|
return null;
|
}
|
}
|
|
protected boolean insertDistribute(Connection connection, DistributedLockDO distributedLockDO) throws SQLException {
|
try (PreparedStatement insertPst = connection.prepareStatement(DistributedLockSqlFactory.getDistributedLogStoreSql(dbType)
|
.getInsertSql(distributedLockTable))) {
|
insertPst.setString(1, distributedLockDO.getLockKey());
|
insertPst.setString(2, distributedLockDO.getLockValue());
|
if (distributedLockDO.getExpireTime() > 0) {
|
distributedLockDO.setExpireTime(distributedLockDO.getExpireTime() + System.currentTimeMillis());
|
}
|
insertPst.setLong(3, distributedLockDO.getExpireTime());
|
return insertPst.executeUpdate() > 0;
|
}
|
}
|
|
protected boolean updateDistributedLock(Connection connection, DistributedLockDO distributedLockDO) throws SQLException {
|
try (PreparedStatement updatePst = connection.prepareStatement(DistributedLockSqlFactory.getDistributedLogStoreSql(dbType)
|
.getUpdateSql(distributedLockTable))) {
|
updatePst.setString(1, distributedLockDO.getLockValue());
|
if (distributedLockDO.getExpireTime() > 0) {
|
distributedLockDO.setExpireTime(distributedLockDO.getExpireTime() + System.currentTimeMillis());
|
}
|
updatePst.setLong(2, distributedLockDO.getExpireTime());
|
updatePst.setString(3, distributedLockDO.getLockKey());
|
return updatePst.executeUpdate() > 0;
|
}
|
}
|
|
private void init() {
|
this.distributedLockDataSource = EnhancedServiceLoader.load(DataSourceProvider.class, datasourceType).provide();
|
}
|
|
private boolean ignoreSQLException(SQLException exception) {
|
if (IGNORE_MYSQL_CODE.contains(exception.getErrorCode())) {
|
return true;
|
}
|
if (StringUtils.isNotBlank(exception.getMessage())) {
|
return IGNORE_MYSQL_MESSAGE.stream().anyMatch(message -> exception.getMessage().contains(message));
|
}
|
return false;
|
}
|
|
}
|
