TransactionSynchronizationManager用法和含义

当我们有业务需要在事务提交过后进行某一项或者某一系列的业务操作时候我们就可以使用TransactionSynchronizationManager

 

通过spring的aop机制将需要进行后置业务处理的操作，提交给spring的处理机制，并且切入到事务处理的后面

 

TransactionSynchronizationManager这个类中由一系列的ThreadLocal ，我们需要关注的是synchronizations，在后面使用到的TransactionSynchronizationManager.isSynchronizationActive()、TransactionSynchronizationManager.registerSynchronization()和new TransactionSynchronizationAdapter()，都与它密切有关。

 

在Spring在开启数据库事务（无论是使用@Transactional注解，还是用xml配置）时，都会向其中写入一个实例，用于自动处理Connection的获取、提交或回滚等操作。

再看isSynchronizationActive()方法，判断了synchronizations中是否有数据（Set<TransactionSynchronization>非null即可，并不要求其中有TransactionSynchronization实例。

再看registerSynchronization()方法，首先调用isSynchronizationActive()做一个校验；然后将入参synchronization添加到synchronizations 中。入参synchronization中的方法不会在这里执行，而是要等到事务执行到特定阶段时才会被调用。

TransactionSynchronizationAdapter是一个适配器：它实现了TransactionSynchronization接口，并为每一个接口方法提供了一个空的实现。这类适配器的基本思想是：接口中定义了很多方法，然而业务代码往往只需要实现其中一小部分。利用这种“空实现”适配器，我们可以专注于业务上需要处理的回调方法，而不用在业务类中放大量而且重复的空方法。

 

结合TransactionSynchronizationManager和TransactionSynchronizationAdapter利用ThreadPoolExecutor实现一个事务后多线程处理功能。

import com.google.common.util.concurrent.ThreadFactoryBuilder; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.transaction.support.TransactionSynchronizationAdapter; import org.springframework.transaction.support.TransactionSynchronizationManager; import java.util.ArrayList; import java.util.List; import java.util.concurrent.*; /** * 事务提交异步线程 * * @author ly */ public class TransactionAfterCommitExecutor extends ThreadPoolExecutor { private static final Logger LOGGER = LoggerFactory.getLogger(TransactionAfterCommitExecutor.class); public TransactionAfterCommitExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) { super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue); } public TransactionAfterCommitExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) { super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory); } public TransactionAfterCommitExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) { super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler); } public TransactionAfterCommitExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) { super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler); } private ThreadLocal<List<Runnable>> currentRunables = new ThreadLocal<List<Runnable>>(){ @Override protected List<Runnable> initialValue() { return new ArrayList<>(5); } }; private ThreadLocal<Boolean> registed = new ThreadLocal<Boolean>(){ @Override protected Boolean initialValue() { return false; } }; /** * 默认策略丢弃最老的数据 */ public TransactionAfterCommitExecutor() { this( 50, 500, 500L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(1024), new ThreadFactoryBuilder().setNameFormat("transaction-after-commit-call-pool-%d").build(), new ThreadPoolExecutor.DiscardOldestPolicy()); } @Override public void execute(final Runnable runnable) { //如果事务同步未启用则认为事务已经提交，马上进行异步处理 if (!TransactionSynchronizationManager.isSynchronizationActive()) { super.execute(runnable); } else { //同一个事务的合并到一起处理 currentRunables.get().add(runnable); //如果存在事务则在事务结束后异步处理 if(!registed.get()){ TransactionSynchronizationManager.registerSynchronization(new AfterCommitTransactionSynchronizationAdapter()); registed.set(true); } } } @Override public Future<?> submit(final Runnable runnable) { //如果事务同步未启用则认为事务已经提交，马上进行异步处理 if (!TransactionSynchronizationManager.isSynchronizationActive()) { return super.submit(runnable); } else { final RunnableFuture<Void> ftask = newTaskFor(runnable, null); //如果存在事务则在事务结束后异步处理 TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() { @Override public void afterCommit() { TransactionAfterCommitExecutor.super.submit(ftask); } }); return ftask; } } private class AfterCommitTransactionSynchronizationAdapter extends TransactionSynchronizationAdapter{ @Override public void afterCompletion(int status) { final List<Runnable> txRunables = new ArrayList<>(currentRunables.get()); currentRunables.remove(); registed.remove(); if(status == STATUS_COMMITTED){ TransactionAfterCommitExecutor.super.execute(new Runnable() { @Override public void run() { for (Runnable runnable : txRunables) { try { runnable.run(); } catch (Exception e) { LOGGER.error("ex:",e); } } } }); } } } } import org.springframework.transaction.support.TransactionSynchronizationManager; import org.springframework.transaction.support.TransactionSynchronizationAdapter; @Transactional(readOnly = false,propagation=Propagation.REQUIRED)//开事物 public void save(String name,Integer age ,BigDecimal amount){ Zexample1Model zexample1Model = new Zexample1Model(); zexample1Model.setName(name+"_"); zexample1Model.setAge(age); zexample1Model.setAmount(amount); zexample1Model.setAddTime(new Date()); zexample1Model.setStatus(1); zexample1Dao.save(zexample1Model); System.out.println("id="+zexample1Model.getId()); TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() { @Override public void afterCommit() { System.out.println("send email after transaction commit..."); } }); System.out.println("this method complete...."); }

或者用于切面的事务处理

import java.lang.reflect.Field; import java.util.Objects; import com.my.data.multisource.redismanager.RedisBean; import com.my.data.utils.ThreadLocalUtil; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import org.aspectj.lang.JoinPoint; import org.aspectj.lang.ProceedingJoinPoint; import org.aspectj.lang.annotation.*; import org.springframework.context.annotation.EnableAspectJAutoProxy; import org.springframework.core.annotation.Order; import org.springframework.data.redis.core.RedisConnectionUtils; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.stereotype.Component; import org.springframework.transaction.support.TransactionSynchronizationManager; import javax.annotation.Resource; @Aspect @Component public class RedisAspect { private Logger logger = LogManager.getLogger(RedisAspect.class); /** * 定义切入点，切入点为com.example.aop下的所有函数 */ @Pointcut("execution(public * com.my.data.service..*.*(..))") public void redisPointcut() { } @Resource(name = RedisBean.defaultStringRedis) private StringRedisTemplate redis; /** * 前置通知：在连接点之前执行的通知 * * @param joinPoint * @throws Throwable */ @Before("redisPointcut()") public void doBefore(JoinPoint joinPoint) throws Throwable { try { Field field = joinPoint.getTarget().getClass().getDeclaredField("redis"); field.setAccessible(true); Object targetRedis = field.get(joinPoint.getTarget()); if (!Objects.isNull(targetRedis)) { this.redis = (StringRedisTemplate) targetRedis; logger.info("redis : {}", redis.hashCode()); } }catch (NoSuchFieldException e) { logger.info("not found redis"); }catch (Exception e) { logger.error("doAfterReturning error.", e); } } @AfterReturning(returning = "ret", pointcut = "redisPointcut()") public void doAfterReturning(Object ret) throws Throwable { try { if (!Objects.isNull(redis)) { logger.info("redis : {}", redis.hashCode()); Object bindResource = TransactionSynchronizationManager.getResource(redis.getConnectionFactory()); if(null == bindResource) { RedisConnectionUtils.unbindConnection(redis.getConnectionFactory()); } } } catch (Exception e) { logger.error("doAfterReturning error.", e); } } }

参考： https://www.cnblogs.com/muxi0407/p/11987409.html