baomidou
新增多数据源事务传播机制
1.主要提取DynamicLocalTransactionInterceptor中的方法到TransactionalTemplate TransactionalInfo transactionInfo = transactionalExecutor.getTransactionInfo(); if (!StringUtils.isEmpty(TransactionContext.getXID())) { return transactionalExecutor.execute(); } boolean state = true; Object o; String xid = LocalTxUtil.startTransaction(); try { o = transactionalExecutor.execute(); } catch (Exception e) { state = !isRollback(e, transactionInfo); throw e; } finally { if (state) { LocalTxUtil.commit(xid); } else { LocalTxUtil.rollback(xid); } } return o; 原代理方法改为获取注解属性对象的封装 Method method = methodInvocation.getMethod(); final DSTransactional dsTransactional = method.getAnnotation(DSTransactional.class);
TransactionalExecutor transactionalExecutor = new TransactionalExecutor() {
@Override
public Object execute() throws Throwable {
return methodInvocation.proceed();
}
@Override
public TransactionalInfo getTransactionInfo() {
TransactionalInfo transactionInfo = new TransactionalInfo();
transactionInfo.setPropagation(dsTransactional.propagation());
transactionInfo.setNoRollbackFor(dsTransactional.noRollbackFor());
transactionInfo.setRollbackFor(dsTransactional.rollbackFor());
return transactionInfo;
}
}
2.AbstractRoutingDataSource获取数据源中获取事务的数据源的方法加入xid ConnectionProxy connection = ConnectionFactory.getConnection(ds); return connection == null ? getConnectionProxy(ds, determineDataSource().getConnection()) : connection; ConnectionProxy connection = ConnectionFactory.getConnection(xid, ds); return connection == null ? getConnectionProxy(xid,ds, determineDataSource().getConnection()) : connection; 3,。增加事务异常类TransactionException
public class TransactionException extends RuntimeException { public TransactionException(String message) { super(message); }
public TransactionException(String message, Throwable cause) {
super(message, cause);
}
} 4.ConnectionFactory 对于事务数据源的处理通过xid保存 通过xid获取指定事务 public class ConnectionFactory {
private static final ThreadLocal<Map<String, ConnectionProxy>> CONNECTION_HOLDER =
new ThreadLocal<Map<String, ConnectionProxy>>() {
private static final ThreadLocal<Map<String, Map<String, ConnectionProxy>>> CONNECTION_HOLDER =
new ThreadLocal<Map<String, Map<String, ConnectionProxy>>>() {
@Override
protected Map<String, ConnectionProxy> initialValue() {
return new ConcurrentHashMap<>(8);
protected Map<String, Map<String, ConnectionProxy>> initialValue() {
return new ConcurrentHashMap<>();
}
};
public static void putConnection(String ds, ConnectionProxy connection) {
Map<String, ConnectionProxy> concurrentHashMap = CONNECTION_HOLDER.get();
if (!concurrentHashMap.containsKey(ds)) {
public static void putConnection(String xid, String ds, ConnectionProxy connection) {
Map<String, Map<String, ConnectionProxy>> concurrentHashMap = CONNECTION_HOLDER.get();
Map<String, ConnectionProxy> connectionProxyMap = concurrentHashMap.get(xid);
if (connectionProxyMap == null) {
connectionProxyMap = new ConcurrentHashMap<>();
concurrentHashMap.put(xid, connectionProxyMap);
}
if (!connectionProxyMap.containsKey(ds)) {
try {
connection.setAutoCommit(false);
} catch (SQLException e) {
e.printStackTrace();
}
concurrentHashMap.put(ds, connection);
connectionProxyMap.put(ds, connection);
}
}
public static ConnectionProxy getConnection(String ds) {
return CONNECTION_HOLDER.get().get(ds);
public static ConnectionProxy getConnection(String xid, String ds) {
Map<String, Map<String, ConnectionProxy>> concurrentHashMap = CONNECTION_HOLDER.get();
Map<String, ConnectionProxy> connectionProxyMap = concurrentHashMap.get(xid);
if (CollectionUtils.isEmpty(connectionProxyMap)) {
return null;
}
return connectionProxyMap.get(ds);
}
public static void notify(Boolean state) {
public static void notify(String xid, Boolean state) {
Map<String, Map<String, ConnectionProxy>> concurrentHashMap = CONNECTION_HOLDER.get();
try {
Map<String, ConnectionProxy> concurrentHashMap = CONNECTION_HOLDER.get();
for (ConnectionProxy connectionProxy : concurrentHashMap.values()) {
connectionProxy.notify(state);
if (CollectionUtils.isEmpty(concurrentHashMap)) {
return;
}
Map<String, ConnectionProxy> connectionProxyMap = concurrentHashMap.get(xid);
for (ConnectionProxy connectionProxy : connectionProxyMap.values()) {
if (connectionProxy != null) {
connectionProxy.notify(state);
}
}
} finally {
CONNECTION_HOLDER.remove();
concurrentHashMap.remove(xid);
}
}
5.DsPropagation 事务传播机制枚举 6.src/main/java/com/baomidou/dynamic/datasource/tx/LocalTxUtil.java 事务工具提交关闭事务加入xid进行获取 public static void startTransaction() { if (!StringUtils.isEmpty(TransactionContext.getXID())) { log.debug("dynamic-datasource exist local tx [{}]", TransactionContext.getXID()); public static String startTransaction() { String xid = TransactionContext.getXID(); if (!StringUtils.isEmpty(xid)) { log.debug("dynamic-datasource exist local tx [{}]", xid); } else { String xid = UUID.randomUUID().toString(); xid = UUID.randomUUID().toString(); TransactionContext.bind(xid); log.debug("dynamic-datasource start local tx [{}]", xid); } return xid; }
/**
* 手动提交事务
*/
public static void commit() {
ConnectionFactory.notify(true);
public static void commit(String xid) {
ConnectionFactory.notify(xid, true);
log.debug("dynamic-datasource commit local tx [{}]", TransactionContext.getXID());
TransactionContext.remove();
}
/**
* 手动回滚事务
*/
public static void rollback() {
ConnectionFactory.notify(false);
public static void rollback(String xid) {
ConnectionFactory.notify(xid, false);
log.debug("dynamic-datasource rollback local tx [{}]", TransactionContext.getXID());
TransactionContext.remove();
}
7.事务传播机制使用的事务暂停xid对象 用于事务暂停和恢复xid package com.baomidou.dynamic.datasource.tx;
import javax.annotation.Nonnull;
public class SuspendedResourcesHolder { /** * The xid */ private String xid;
public SuspendedResourcesHolder(String xid) {
if (xid == null) {
throw new IllegalArgumentException("xid must be not null");
}
this.xid = xid;
}
@Nonnull
public String getXid() {
return xid;
}
} 8。TransactionalExecutor 事务执行器接口 9.TransactionalInfo 对事务注解中属性的封装 10.TransactionalTemplate 事务模板方法 对事务的操作抽取到模板方法中实现。后续可自定义扩展模板方法 处理异常和事务传播机制 package com.baomidou.dynamic.datasource.tx;
import com.baomidou.dynamic.datasource.exception.TransactionException; import com.baomidou.mybatisplus.core.toolkit.ArrayUtils; import lombok.extern.slf4j.Slf4j; import org.springframework.util.StringUtils;
import java.util.Objects;
@Slf4j public class TransactionalTemplate {
public Object execute(TransactionalExecutor transactionalExecutor) throws Throwable {
TransactionalInfo transactionInfo = transactionalExecutor.getTransactionInfo();
DsPropagation propagation = transactionInfo.propagation;
SuspendedResourcesHolder suspendedResourcesHolder = null;
try {
switch (propagation) {
case NOT_SUPPORTED:
// If transaction is existing, suspend it.
if (existingTransaction()) {
suspendedResourcesHolder = suspend();
}
return transactionalExecutor.execute();
case REQUIRES_NEW:
// If transaction is existing, suspend it, and then begin new transaction.
if (existingTransaction()) {
suspendedResourcesHolder = suspend();
}
// Continue and execute with new transaction
break;
case SUPPORTS:
// If transaction is not existing, execute without transaction.
if (!existingTransaction()) {
return transactionalExecutor.execute();
}
// Continue and execute with new transaction
break;
case REQUIRED:
// default
break;
case NEVER:
// If transaction is existing, throw exception.
if (existingTransaction()) {
throw new TransactionException("Existing transaction found for transaction marked with propagation never");
} else {
// Execute without transaction and return.
return transactionalExecutor.execute();
}
case MANDATORY:
// If transaction is not existing, throw exception.
if (!existingTransaction()) {
throw new TransactionException("No existing transaction found for transaction marked with propagation 'mandatory'");
}
// Continue and execute with current transaction.
break;
default:
throw new TransactionException("Not Supported Propagation:" + propagation);
}
return doExecute(transactionalExecutor);
} finally {
resume(suspendedResourcesHolder);
}
}
private Object doExecute(TransactionalExecutor transactionalExecutor) throws Throwable {
TransactionalInfo transactionInfo = transactionalExecutor.getTransactionInfo();
if (!StringUtils.isEmpty(TransactionContext.getXID())) {
return transactionalExecutor.execute();
}
boolean state = true;
Object o;
String xid = LocalTxUtil.startTransaction();
try {
o = transactionalExecutor.execute();
} catch (Exception e) {
state = !isRollback(e, transactionInfo);
throw e;
} finally {
if (state) {
LocalTxUtil.commit(xid);
} else {
LocalTxUtil.rollback(xid);
}
}
return o;
}
private boolean isRollback(Throwable e, TransactionalInfo transactionInfo) {
boolean isRollback = true;
Class<? extends Throwable>[] rollbacks = transactionInfo.rollbackFor;
Class<? extends Throwable>[] noRollbackFor = transactionInfo.noRollbackFor;
if (ArrayUtils.isNotEmpty(noRollbackFor)) {
for (Class<? extends Throwable> noRollBack : noRollbackFor) {
int depth = getDepth(e.getClass(), noRollBack);
if (depth >= 0) {
return false;
}
}
}
if (ArrayUtils.isNotEmpty(rollbacks)) {
for (Class<? extends Throwable> rollback : rollbacks) {
int depth = getDepth(e.getClass(), rollback);
if (depth >= 0) {
return isRollback;
}
}
}
return false;
}
private int getDepth(Class<?> exceptionClass, Class<? extends Throwable> rollback) {
if (rollback == Throwable.class || rollback == Exception.class) {
return 0;
}
// If we've gone as far as we can go and haven't found it...
if (exceptionClass == Throwable.class) {
return -1;
}
if (Objects.equals(exceptionClass, rollback)) {
return 0;
}
return getDepth(exceptionClass.getSuperclass(), rollback);
}
private void resume(SuspendedResourcesHolder suspendedResourcesHolder) {
if (suspendedResourcesHolder != null) {
String xid = suspendedResourcesHolder.getXid();
TransactionContext.bind(xid);
}
}
public SuspendedResourcesHolder suspend() {
String xid = TransactionContext.getXID();
if (xid != null) {
if (log.isInfoEnabled()) {
log.info("Suspending current transaction, xid = {}", xid);
}
TransactionContext.unbind(xid);
return new SuspendedResourcesHolder(xid);
} else {
return null;
}
}
public boolean existingTransaction() {
return !StringUtils.isEmpty(TransactionContext.getXID());
}
}
