我要评论一、原生jdbc的父子线程共用事务
初始化数据库中的数据
1、单线程事务提交
private static void t1() throws exception {
connection conn = getconn();
preparedstatement stmt1 = null;
preparedstatement stmt2 = null;
try {
// 1. 禁用自动提交，开启事务
conn.setautocommit(false);
// 2. 执行多个数据库操作
string sql1 = "update student set age = age - 1 where id = ?";
stmt1 = conn.preparestatement(sql1);
stmt1.setint(1, 1);
stmt1.executeupdate();
string sql2 = "update student set age = age + 1 where id = ?";
stmt2 = conn.preparestatement(sql2);
stmt2.setint(1, 2);
stmt2.executeupdate();
// 3. 事务回滚, 将会回滚两个执行语句的内容
conn.rollback();
// 4. 提交事务
conn.commit();
system.out.println("transaction committed successfully.");
} catch (sqlexception e) {
// 发生异常时回滚事务
if (conn != null) {
try {
conn.rollback();
system.out.println("transaction rolled back.");
} catch (sqlexception ex) {
ex.printstacktrace();
}
}
e.printstacktrace();
} finally {
close(stmt1, stmt2);
}
}
获取数据库连接和释放连接
private static connection getconn() throws exception {
string url = "jdbc:mysql://localhost:3306/test";
string username = "root";
string password = "123456";
// 1. 获取数据库连接
return drivermanager.getconnection(url, username, password);
}
private static void close(autocloseable... closeables) {
for (autocloseable closeable : closeables) {
if (closeable != null) {
try {
closeable.close();
} catch (exception e) {
// ... do something
e.printstacktrace();
}
}
}
}
执行完后, 数据库中数据如下, 可以看到两条语句都执行成功
2、父子线程提交事务
还原数据库中的数据如下
private static void t2() throws exception {
connection conn = getconn();
preparedstatement stmt1 = null;
try {
// 1. 禁用自动提交，开启事务
conn.setautocommit(false);
// 2. 父线程执行
string sql1 = "update student set age = age - 1 where id = ?";
stmt1 = conn.preparestatement(sql1);
stmt1.setint(1, 1);
stmt1.executeupdate();
// 3. 子线程执行
thread t1 = new thread(() -> {
string sql2 = "update student set age = age + 1 where id = ?";
try {
preparedstatement stmt2 = conn.preparestatement(sql2);
stmt2.setint(1, 2);
stmt2.executeupdate();
// 子线程回滚
conn.rollback();
} catch (sqlexception e) {
e.printstacktrace();
}
});
t1.start();
// 等t1执行完
t1.join();
// 4. 提交事务
conn.commit();
system.out.println("transaction committed successfully.");
} catch (sqlexception e) {
// 发生异常时回滚事务
if (conn != null) {
try {
conn.rollback();
system.out.println("transaction rolled back.");
} catch (sqlexception ex) {
ex.printstacktrace();
}
}
e.printstacktrace();
} finally {
close(stmt1);
}
}
执行之后看结果
age没有发生变化, 也就是说子线程回滚了, 主线程中的提交不会生效, 达到了异步任务中事务共享的效果。
二、jdbctemplate实现父子共用事务
以jdbctemplate的update方法为例, 看看内部实现
// ... 只看核心update方法
protected int update(final preparedstatementcreator psc, @nullable final preparedstatementsetter pss)
throws dataaccessexception {
return updatecount(execute(psc, ps -> {
try {
// 赋值
if (pss != null) {
pss.setvalues(ps);
}
// 使用preparedstatement执行
int rows = ps.executeupdate();
if (logger.istraceenabled()) {
logger.trace("sql update affected " + rows + " rows");
}
return rows;
}
finally {
// ...
}
}, true));
}
继续看execute
private <t> t execute(preparedstatementcreator psc, preparedstatementcallback<t> action, boolean closeresources)
throws dataaccessexception {
// 从数据源中获取一个链接
connection con = datasourceutils.getconnection(obtaindatasource());
preparedstatement ps = null;
try {
// ...省略部分代码
// 执行
t result = action.doinpreparedstatement(ps);
return result;
}
catch (sqlexception ex) {
//...
}
finally {
// 关闭资源
if (closeresources) {
if (psc instanceof parameterdisposer) {
((parameterdisposer) psc).cleanupparameters();
}
// 关闭statement
jdbcutils.closestatement(ps);
// 回收链接(连接池的场景下是对连接的引用数-1, 并不会直接释放连接)
datasourceutils.releaseconnection(con, getdatasource());
}
}
}
这里是从连接池中获取一个链接, 然后执行对应的db操作, 这里重点看一下 connection con = datasourceutils.getconnection(obtaindatasource());
datasourceutils.getconnection的核心逻辑如下
public static connection dogetconnection(datasource datasource) throws sqlexception {
assert.notnull(datasource, "no datasource specified");
// 上下文中获取当前数据源中的连接句柄
connectionholder conholder = (connectionholder) transactionsynchronizationmanager.getresource(datasource);
// 1.连接句柄存在 并且 2(句柄中有链接 或者 是事务同步)
if (conholder != null && (conholder.hasconnection() || conholder.issynchronizedwithtransaction())) {
// 引用数+1
conholder.requested();
// 直接获取连接
return conholder.getconnection();
}
// 从数据源中获取连接
connection con = fetchconnection(datasource);
// ... 省略事务同步处理逻辑
return con;
}
从这个方法可以知道两点
- 如果当前上下文中存在连接就用该连接
- 不存在上下文中的连接就从数据源中获取
清楚了这个逻辑后, 我们来构建多线性下使用事务的场景
@autowired
private datasource datasource;
@autowired
private jdbctemplate jdbctemplate;
private void jdbctemplatetest() throws exception {
connection con = datasourceutils.getconnection(datasource);
con.setautocommit(false);
connectionholder holdertouse = new connectionholder(con);
// 在jdbctemplate执行update之前将数据库连接放到上下文中
transactionsynchronizationmanager.bindresource(datasource, holdertouse);
string sql1 = "update student set age = age - 1 where id = ?";
jdbctemplate.update(sql1, "1");
thread t1 = new thread(() -> {
transactionsynchronizationmanager.bindresource(datasource, holdertouse);
string sql2 = "update student set age = age + 1 where id = ?";
jdbctemplate.update(sql2, "2");
try {
con.rollback();
} catch (sqlexception e) {
throw new runtimeexception(e);
}
});
t1.start();
t1.join();
con.commit();
con.setautocommit(true);
}
注意这里需要在jdbctemplate执行update之前调用bindresource操作, 因为上面介绍的update里面会先从上下文中获取数据库连接, 在子线程中也需要执行bindresource将连接绑定到子线程中, 这样父子线程就用的同一个连接了。
注意这里是使用connection直接调用的setautocommit控制手动提交, 所以用完后记得设置为自动提交
三、platformtransactionmanager 管理事务下的父子线程共享事务
简单demo
先看下一般的单线程下的手动事务
@autowired
private platformtransactionmanager transactionmanager;
@autowired
private transactiondefinition definition;
private void manualtx() {
string sql1 = "update student set age = age - 1 where id = ?";
// 开启手动事务
transactionstatus status = this.transactionmanager.gettransaction(definition);
try {
jdbctemplate.update(sql1, "1");
throw new runtimeexception("错误了");
} catch (exception e) {
this.transactionmanager.rollback(status);
}
if (!status.iscompleted()) {
this.transactionmanager.commit(status);
}
}
这里主要是使用gettransaction方法开启手动提交, 然后rollback方法回滚事务, commit方法提交事务
原理剖析
这里我们看看手动开启方法gettransaction
public final transactionstatus gettransaction(@nullable transactiondefinition definition)
throws transactionexception {
// use defaults if no transaction definition given.
transactiondefinition def = (definition != null ? definition : transactiondefinition.withdefaults());
// 上下文中获取事务
object transaction = dogettransaction();
// 判断是否获取到了连接和并且事务由jdbc管理
if (isexistingtransaction(transaction)) {
// existing transaction found -> check propagation behavior to find out how to behave.
return handleexistingtransaction(def, transaction, debugenabled);
}
// 以下是事务不存在的情况
// 判断传播机制; 这里是必须有事务的传播机制
if (def.getpropagationbehavior() == transactiondefinition.propagation_mandatory) {
throw new illegaltransactionstateexception(
"no existing transaction found for transaction marked with propagation 'mandatory'");
}
// 需要事务的传播机制
else if (def.getpropagationbehavior() == transactiondefinition.propagation_required ||
def.getpropagationbehavior() == transactiondefinition.propagation_requires_new ||
def.getpropagationbehavior() == transactiondefinition.propagation_nested) {
suspendedresourcesholder suspendedresources = suspend(null);
// 创建新的事物
try {
return starttransaction(def, transaction, debugenabled, suspendedresources);
}
catch (runtimeexception | error ex) {
resume(null, suspendedresources);
throw ex;
}
}
else {
// 创建空事物
// ....
}
}
核心逻辑
- 如果存在事务, 则使用当前事务
- 如果不存在事务, 并且隔离界别需要开启事务, 那么创建新事务
再跟一下dogettransaction方法
protected object dogettransaction() {
datasourcetransactionobject txobject = new datasourcetransactionobject();
txobject.setsavepointallowed(isnestedtransactionallowed());
connectionholder conholder =
(connectionholder) transactionsynchronizationmanager.getresource(obtaindatasource());
txobject.setconnectionholder(conholder, false);
return txobject;
}
我们看到了上面也介绍过的从上下文中获取连接句柄的方法transactionsynchronizationmanager.getresource
继续看starttransaction方法, 创建新的事物
private transactionstatus starttransaction(transactiondefinition definition, object transaction,
boolean debugenabled, @nullable suspendedresourcesholder suspendedresources) {
boolean newsynchronization = (gettransactionsynchronization() != synchronization_never);
defaulttransactionstatus status = newtransactionstatus(
definition, transaction, true, newsynchronization, debugenabled, suspendedresources);
// 开启事务
dobegin(transaction, definition);
// 将事物相关信息放入到上下文中
preparesynchronization(status, definition);
return status;
}
这里做了两件事: 1. 开启事务 2. 将事物相关信息放入到上下文中
先看一下开启事务dobegin方法
protected void dobegin(object transaction, transactiondefinition definition) {
datasourcetransactionobject txobject = (datasourcetransactionobject) transaction;
connection con = null;
try {
// 1.上下文中没有连接 2.或者需要将事务资源绑定到上下文中
if (!txobject.hasconnectionholder() ||
txobject.getconnectionholder().issynchronizedwithtransaction()) {
// 创建新的数据库连接
connection newcon = obtaindatasource().getconnection();
// 标记为新创建的数据库连接
txobject.setconnectionholder(new connectionholder(newcon), true);
}
// 设置为绑定资源到上下文
txobject.getconnectionholder().setsynchronizedwithtransaction(true);
con = txobject.getconnectionholder().getconnection();
// 隔离级别
integer previousisolationlevel = datasourceutils.prepareconnectionfortransaction(con, definition);
txobject.setpreviousisolationlevel(previousisolationlevel);
txobject.setreadonly(definition.isreadonly());
// 设置为手动提交
if (con.getautocommit()) {
txobject.setmustrestoreautocommit(true);
con.setautocommit(false);
}
// 设置事务只读
preparetransactionalconnection(con, definition);
// 标记事务激活状态
txobject.getconnectionholder().settransactionactive(true);
int timeout = determinetimeout(definition);
if (timeout != transactiondefinition.timeout_default) {
txobject.getconnectionholder().settimeoutinseconds(timeout);
}
// 新的数据库连接, 就将连接绑定的上下文中
if (txobject.isnewconnectionholder()) {
transactionsynchronizationmanager.bindresource(obtaindatasource(), txobject.getconnectionholder());
}
}
}
方法小结
- 没事务就开启事务
- 设置事务为手动提交
- 将数据库连接放到上下文中
transactionsynchronizationmanager.bindresource
咋们再来看看preparesynchronization方法, 将事务信息放到上下文
protected void preparesynchronization(defaulttransactionstatus status, transactiondefinition definition) {
// 新事务打开了事务同步
if (status.isnewsynchronization()) {
// 事务激活状态
transactionsynchronizationmanager.setactualtransactionactive(status.hastransaction());
// 隔离级别
transactionsynchronizationmanager.setcurrenttransactionisolationlevel(
definition.getisolationlevel() != transactiondefinition.isolation_default ?
definition.getisolationlevel() : null);
transactionsynchronizationmanager.setcurrenttransactionreadonly(definition.isreadonly());
transactionsynchronizationmanager.setcurrenttransactionname(definition.getname());
transactionsynchronizationmanager.initsynchronization();
}
}
在传播机制是propagation_required(有事务就用当前事务, 没有就新建)的场景下我们看一下handleexistingtransaction方法, 用来获取当前事务
// ... 省略传播机制判断的代码
if (isvalidateexistingtransaction()) {
// 非默认隔离级别, 校验隔离级别
if (definition.getisolationlevel() != transactiondefinition.isolation_default) {
integer currentisolationlevel = transactionsynchronizationmanager.getcurrenttransactionisolationlevel();
if (currentisolationlevel == null || currentisolationlevel != definition.getisolationlevel()) {
// ...抛异常
}
}
// 校验readonly
if (!definition.isreadonly()) {
if (transactionsynchronizationmanager.iscurrenttransactionreadonly()) {
// ...抛异常
}
}
}
// 将事务信息绑定到线程上下文
boolean newsynchronization = (gettransactionsynchronization() != synchronization_never);
return preparetransactionstatus(definition, transaction, false, newsynchronization, debugenabled, null);
总结一下transactionmanager.gettransaction方法
- 从当前上下文获取数据库连接句柄
transactionsynchronizationmanager.getresource - 如果存在连接和事务
- 直接使用当前事务
- 校验事务的传播机制, 是否只读
- 将事务信息放到上下文
- 如果当前不存在事务(
propagation_required场景下)
- 创建新的数据库连接
- 设置为手动提交
- 绑定数据库连接句柄到上下文
transactionsynchronizationmanager.bindresource - 将事务信息放到上下文
关于transactionstatus的commit方法和rollback方法我们就不看了, 它里面都校验了事务是否已经激活事务同步, 可以使用transactionsynchronizationmanager.initsynchronization方法来激活
有了以上信息支撑后, 我们再来看platformtransactionmanager下的父子线程共享事务
我们只需要将父线程中的数据库句柄放到子线程, 并且设置传播机制, 是否只读, 激活事务同步就可以了;
platformtransactionmanager下的父子线程使用同一个事务的demo
private void manualmultitx() throws exception {
string sql1 = "update student set age = age - 1 where id = ?";
// 开启手动事务
transactionstatus status = this.transactionmanager.gettransaction(definition);
// 获取父线程中绑定的数据库连接句柄
connectionholder connectionholder = (connectionholder) transactionsynchronizationmanager.getresource(datasource);
// 因为gettransaction已经将connectionholder绑定到当前线程(transactionsynchronizationmanager.bindresource)，所以这里会直接拿这个连接
jdbctemplate.update(sql1, "1");
thread t1 = new thread(() -> {
/******************** 核心步骤start ***************/
// 将connectionholder绑定到当前线程
transactionsynchronizationmanager.bindresource(datasource, connectionholder);
// 将事务相关信息绑定到现成上下文
// 设置隔离级别
transactionsynchronizationmanager.setcurrenttransactionisolationlevel(definition.getisolationlevel());
// 设置是否只读
transactionsynchronizationmanager.setcurrenttransactionreadonly(definition.isreadonly());
// 激活事务同步
transactionsynchronizationmanager.initsynchronization();
/******************** 核心步骤end ***************/
system.out.println("线程1执行");
jdbctemplate.update(sql1, "2");
if (!status.iscompleted()) {
this.transactionmanager.rollback(status);
}
});
t1.start();
t1.join();
system.out.println("run线程执行");
if (!status.iscompleted()) {
this.transactionmanager.commit(status);
}
}
其实这里的将事务相关信息绑定到现成上下文直接copy上面的preparesynchronization方法的内容即可
改成这样更完整
/******************** 核心步骤start ***************/ transactionsynchronizationmanager.bindresource(datasource, connectionholder); defaulttransactionstatus s = (defaulttransactionstatus) status; // 事务激活状态 transactionsynchronizationmanager.setactualtransactionactive(s.hastransaction()); // 隔离级别, 这里稍微不一样 transactionsynchronizationmanager.setcurrenttransactionisolationlevel(definition.getisolationlevel()); transactionsynchronizationmanager.setcurrenttransactionreadonly(definition.isreadonly()); transactionsynchronizationmanager.setcurrenttransactionname(definition.getname()); transactionsynchronizationmanager.initsynchronization(); /******************** 核心步骤end ***************/
四、@transactional注解下的父子线程事务共享
我们得看到@transactional注解的增强类transactioninterceptor中去;
在invoke中调用了invokewithintransaction方法, 我们关注一下
protected object invokewithintransaction(method method, @nullable class<?> targetclass,
final invocationcallback invocation) throws throwable {
// 事务属性
transactionattributesource tas = gettransactionattributesource();
final transactionattribute txattr = (tas != null ? tas.gettransactionattribute(method, targetclass) : null);
// 事务管理器
final transactionmanager tm = determinetransactionmanager(txattr);
// ... 省略一些其它代码
platformtransactionmanager ptm = asplatformtransactionmanager(tm);
// 切点方法
final string joinpointidentification = methodidentification(method, targetclass, txattr);
// 这里是jdbctransactionmanager, 也是platformtransactionmanager实现类
if (txattr == null || !(ptm instanceof callbackpreferringplatformtransactionmanager)) {
// 创建事务
transactioninfo txinfo = createtransactionifnecessary(ptm, txattr, joinpointidentification);
object retval;
try {
// 调用目标方法
retval = invocation.proceedwithinvocation();
}
catch (throwable ex) {
// 这里会rollback
completetransactionafterthrowing(txinfo, ex);
throw ex;
}
// 这里会commit
committransactionafterreturning(txinfo);
return retval;
}
// ... 省略部分代码
}
这里我们只需要关注用来创建事务的createtransactionifnecessary方法即可
protected transactioninfo createtransactionifnecessary(@nullable platformtransactionmanager tm,
@nullable transactionattribute txattr, final string joinpointidentification) {
// ... 省略部分代码
transactionstatus status = null;
if (txattr != null) {
if (tm != null) {
status = tm.gettransaction(txattr);
}
}
// 包装一下返回
return preparetransactioninfo(tm, txattr, joinpointidentification, status);
}
回到了上面原理剖析中介绍的platformtransactionmanager#gettransaction方法了, 也就是说它也会先从线程上下文中获取数据库连接句柄, 有的话直接用了, 没有就去创建然后放到当前上下文, 和上面介绍的第3节中的一样, 这里就不赘述了。
demo和上面的也基本没有区别
private void transactionaltest() throws exception {
// 开启手动事务
transactionstatus status = this.transactionmanager.gettransaction(definition);
connectionholder connectionholder = (connectionholder) transactionsynchronizationmanager.getresource(datasource);
// 因为gettransaction已经将connectionholder绑定到当前线程(transactionsynchronizationmanager.bindresource)，所以这里会直接拿这个连接
// 主线程中service调用
myservice.trantest();
thread t1 = new thread(() -> {
// 将connectionholder绑定到当前线程
transactionsynchronizationmanager.bindresource(datasource, connectionholder);
defaulttransactionstatus s = (defaulttransactionstatus) status;
// 标记当前事务已存在
transactionsynchronizationmanager.setactualtransactionactive(s.hastransaction());
// 设置隔离级别
transactionsynchronizationmanager.setcurrenttransactionisolationlevel(definition.getisolationlevel());
// 设置是否只读
transactionsynchronizationmanager.setcurrenttransactionreadonly(definition.isreadonly());
transactionsynchronizationmanager.setcurrenttransactionname(definition.getname());
transactionsynchronizationmanager.initsynchronization();
system.out.println("线程1执行");
// 子线程中service调用
myservice.trantest();
if (!status.iscompleted()) {
this.transactionmanager.rollback(status);
}
});
t1.start();
t1.join();
system.out.println("run线程执行");
if (!status.iscompleted()) {
this.transactionmanager.commit(status);
}
}
service
@component
public class myservice {
@autowired
private jdbctemplate jdbctemplate;
@transactional
public void trantest() {
string sql1 = "update student set age = age - 1 where id = ?";
jdbctemplate.update(sql1, "2");
}
}
五、总结
- 父子线程中事务共享的原理就是要让一个数据库连接在父子线程中同时使用
- 原生jdbc中, 将数据库连接直接从父线程传递到子线程即可
- jdbctemplate的使用中, 只需将同一个数据库连接同时绑定到父子线程中接口, 使用
transactionsynchronizationmanager.bindresource - 在spring中使用
platformtransactionmanager开启手动事务, 除了第三点的需要将数据库连接同时在父线程中绑定transactionmanager.gettransaction(definition)和在子线程中绑定transactionsynchronizationmanager.bindresource之外, 还需要将事务的相关信息绑定到子线程(详细见文章内部) - 在spring中使用
platformtransactionmanager不管是配合jdbctemplate还是调用@transactional使用声明式原理都是一样的, 都是得从在子线程中绑定父线程中的相同的事务信息和连接信息
到此这篇关于spring中父子线程共享事务的实现的文章就介绍到这了,更多相关spring 父子线程共享事务内容请搜索代码网以前的文章或继续浏览下面的相关文章希望大家以后多多支持代码网！
发表评论