欢迎您访问 最编程 本站为您分享编程语言代码，编程技术文章！

热门搜索/Hot Search

您现在的位置是： 首页

SpringBoot + 交易钩子功能，打造高效支付系统！

最编程 2024-04-30 20:40:52
...

92b6a44289626dd2f94b7389e1eb1b3c.jpeg来源：juejin.cn/post/6984574787511123999

???? 欢迎加入小哈的星球 ，你将获得: 专属的项目实战 / Java 学习路线 / 一对一提问 / 学习打卡 /  赠书福利

全栈前后端分离博客项目 2.0 版本完结啦， 演示链接：http://116.62.199.48/ ，新项目正在酝酿中。全程手摸手，后端 + 前端全栈开发，从 0 到 1 讲解每个功能点开发步骤，1v1 答疑，直到项目上线。目前已更新了239小节，累计38w+字，讲解图：1645张，还在持续爆肝中.. 后续还会上新更多项目，目标是将Java领域典型的项目都整一波，如秒杀系统, 在线商城, IM即时通讯，Spring Cloud Alibaba 等等，戳我加入学习，已有1300+小伙伴加入(早鸟价超低)

a95ecf05da715ad39bc6800cd066ea50.gif

前言

经过前面对Spring AOP、事务的总结，我们已经对它们有了一个比较感性的认知了。今天，我继续安利一个独门绝技：Spring 事务的钩子函数。 单纯的讲技术可能比较枯燥乏味。

接下来，我将以一个实际的案例来描述Spring事务钩子函数的正确使用姿势。

一、案例背景

拿支付系统相关的业务来举例。在支付系统中，我们需要记录每个账户的资金流水（记录用户A因为哪个操作扣了钱，因为哪个操作加了钱），这样我们才能对每个账户的账做到心中有数，对于支付系统而言，资金流水的数据可谓是最重要的。

因此，为了防止支付系统的老大徇私舞弊，CTO提了一个流水存档的需求：要求支付系统对每个账户的资金流水做一份存档，要求支付系统在写流水的时候，把流水相关的信息以消息的形式推送到kafka，由存档系统消费这个消息并落地到库里（这个库只有存档系统拥有写权限）。

整个需求的流程如下所示：

78fe4bc0e494ca4db34bc565e19d1d1f.jpeg

图片

整个需求的流程还是比较简单的，考虑到后续会有其他事业部也要进行数据存档操作，CTO建议支付系统团队内部开发一个二方库，这个二方库的主要功能就是发送消息到kafka中去。

二、确定方案

既然要求开发一个二方库，因此，我们需要考虑如下几件事情：

1、技术栈使用的springboot，因此，这里最好以starter的方式提供

2、二方库需要发送消息给kafka，最好是二方库内部基于kafka生产者的api创建生产者，不要使用Spring自带的kafkaTemplate，因为集成方有可能已经使用了kafkaTemplate。不能与集成方造成冲突。

3、减少对接方的集成难度、学习成本，最好是提供一个简单实用的api，业务侧能简单上手。

4、发送消息这个操作需要支持事务，尽量不影响主业务

在上述的几件事情中，最需要注意的应该就是第4点：发送消息这个操作需要支持事务，尽量不影响主业务。 这是什么意思呢？首先，尽量不影响主业务，这个最简单的方式就是使用异步机制。其次，需要支持事务是指：假设我们的api是在事务方法内部调用的，那么我们需要保证事务提交后再执行这个api。 那么，我们的流水落地api应该要有这样的功能：

26347e97b5b8157ac3704f7921c2dfb7.jpeg

图片

内部可以判断当前是否存在事务，如果存在事务，则需要等事务提交后再异步发送消息给kafka。如果不存在事务则直接异步发送消息给kafka。而且这样的判断逻辑得放在二方库内部才行。那现在摆在我们面前的问题就是：我要如何判断当前是否存在事务，以及如何在事务提交后再触发我们自定义的逻辑呢？

三、TransactionSynchronizationManager显神威

这个类内部所有的变量、方法都是static修饰的，也就是说它其实是一个工具类。是一个事务同步器。下述是流水落地API的伪代码，这段代码就解决了我们上述提到的疑问：

private final ExecutorService executor = Executors.newSingleThreadExecutor();

public void sendLog() {
    // 判断当前是否存在事务
    if (!TransactionSynchronizationManager.isSynchronizationActive()) {
        // 无事务，异步发送消息给kafka
        
        executor.submit(() -> {
            // 发送消息给kafka
            try {
                // 发送消息给kafka
            } catch (Exception e) {
                // 记录异常信息，发邮件或者进入待处理列表，让开发人员感知异常
            }
        });
        return;
    }

    // 有事务，则添加一个事务同步器，并重写afterCompletion方法（此方法在事务提交后会做回调）
    TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {

        @Override
        public void afterCompletion(int status) {
            if (status == TransactionSynchronization.STATUS_COMMITTED) {
                // 事务提交后，再异步发送消息给kafka
                executor.submit(() -> {
                    try {
                     // 发送消息给kafka
                    } catch (Exception e) {
                     // 记录异常信息，发邮件或者进入待处理列表，让开发人员感知异常
                    }
                });
            }
        }

    });

}

代码比较简单，其主要是TransactionSynchronizationManager的使用。

3.1、判断是否存在事务？TransactionSynchronizationManager.isSynchronizationActive() 方法显神威

我们先看下这个方法的源码：

// TransactionSynchronizationManager.java类内部的部分代码

private static final ThreadLocal<Set<TransactionSynchronization>> synchronizations =
   new NamedThreadLocal<>("Transaction synchronizations");

public static boolean isSynchronizationActive() {
    return (synchronizations.get() != null);
}

很明显，synchronizations是一个线程变量（ThreadLocal）。那它是在什么时候set进去的呢？

这里的话，可以参考下这个方法：org.springframework.transaction.support.TransactionSynchronizationManager#initSynchronization，其源码如下所示：

/**
  * Activate transaction synchronization for the current thread.
  * Called by a transaction manager on transaction begin.
  * @throws IllegalStateException if synchronization is already active
  */
public static void initSynchronization() throws IllegalStateException {
    if (isSynchronizationActive()) {
        throw new IllegalStateException("Cannot activate transaction synchronization - already active");
    }
    logger.trace("Initializing transaction synchronization");
    synchronizations.set(new LinkedHashSet<>());
}

由源码中的注释也可以知道，它是在事务管理器开启事务时调用的。换句话说，只要我们的程序执行到带有事务特性的方法时，就会在线程变量中放入一个LinkedHashSet，用来标识当前存在事务。只要isSynchronizationActive返回true，则代表当前有事务。

因此，结合这两个方法我们是指能解决我们最开始提出的疑问：要如何判断当前是否存在事务

3.2、如何在事务提交后触发自定义逻辑？TransactionSynchronizationManager.registerSynchronization()方法显神威

我们来看下这个方法的源代码：

/**
  * Register a new transaction synchronization for the current thread.
  * Typically called by resource management code.
  * <p>Note that synchronizations can implement the
  * {@link org.springframework.core.Ordered} interface.
  * They will be executed in an order according to their order value (if any).
  * @param synchronization the synchronization object to register
  * @throws IllegalStateException if transaction synchronization is not active
  * @see org.springframework.core.Ordered
  */
public static void registerSynchronization(TransactionSynchronization synchronization)
    throws IllegalStateException {

    Assert.notNull(synchronization, "TransactionSynchronization must not be null");
    if (!isSynchronizationActive()) {
        throw new IllegalStateException("Transaction synchronization is not active");
    }
    synchronizations.get().add(synchronization);
}

这里又使用到了synchronizations线程变量，我们在判断是否存在事务时，就是判断这个线程变量内部是否有值。那我们现在想在事务提交后触发自定义逻辑和这个有什么关系呢？

我们在上面构建流水落地api的伪代码中有向synchronizations内部添加了一个TransactionSynchronizationAdapter，内部并重写了afterCompletion方法，其代码如下所示：

TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {

    @Override
    public void afterCompletion(int status) {
        if (status == TransactionSynchronization.STATUS_COMMITTED) {
            // 事务提交后，再异步发送消息给kafka
            executor.submit(() -> {
                    try {
                     // 发送消息给kafka
                    } catch (Exception e) {
                     // 记录异常信息，发邮件或者进入待处理列表，让开发人员感知异常
                    }
            });
        }
    }

});

我们结合registerSynchronization的源码来看，其实这段代码主要就是向线程变量内部的LinkedHashSet添加了一个对象而已，但就是这么一个操作，让Spring在事务执行的过程中变得“有事情可做”。这是什么意思呢？

是因为Spring在执行事务方法时，对于操作事务的每一个阶段都有一个回调操作，比如：trigger系列的回调

9ed5074e843d1f320e84bfd11348929b.jpeg

图片

invoke系列的回调

cb7e61c2e4c8614dab33149f1f84ff6b.jpeg

图片

而我们现在的需求就是在事务提交后触发自定义的函数，那就是在invokeAfterCommit和invokeAfterCompletion这两个方法来选了。首先，这两个方法都会拿到所有TransactionSynchronization的集合（其中会包括我们上述添加的TransactionSynchronizationAdapter）。

但是要注意一点：invokeAfterCommit只能拿到集合，invokeAfterCompletion除了集合还有一个int类型的参数，而这个int类型的参数其实是当前事务的一种状态。也就是说，如果我们重写了invokeAfterCompletion方法，我们除了能拿到集合外，还能拿到当前事务的状态。

因此，此时我们可以根据这个状态来做不同的事情，比如：可以在事务提交时做自定义处理，也可以在事务回滚时做自定义处理等等。

四、总结

上面有说到，我们判断当前是否存在事务、添加钩子函数都是依赖线程变量的。因此，我们在使用过程中，一定要避免切换线程。否则会出现不生效的情况。

好书推荐

58d576aead2d53ab7e1c6b58042d0e92.jpeg

玩转“搜索绝技”，成为“解决问题”的高手

80种搜索工具+90个搜索技巧+100个典型场景应用

讲解基于信息解决问题的方法和思维

也是工作、学习、生活的底层能力

更是信息化时代生存思维进阶的核心素养

d90d96906b49c4bc6681bc94108ddd04.jpeg

解锁MATLAB科研绘图魅力

让数据可视化成为你的科研利器！

专业级MATLAB科研绘图绘制指南

以“软件底层原理+详解代码逻辑+案例实际操作”为讲解思路

轻松掌握科技绘图基础、变量图形

极坐标图形、3D图形、地理信息可视化等

???? 欢迎加入小哈的星球 ，你将获得: 专属的项目实战 / Java 学习路线 / 一对一提问 / 学习打卡 /  赠书福利

全栈前后端分离博客项目 2.0 版本完结啦， 演示链接：http://116.62.199.48/ ，新项目正在酝酿中。全程手摸手，后端 + 前端全栈开发，从 0 到 1 讲解每个功能点开发步骤，1v1 答疑，直到项目上线。目前已更新了239小节，累计38w+字，讲解图：1645张，还在持续爆肝中.. 后续还会上新更多项目，目标是将Java领域典型的项目都整一波，如秒杀系统, 在线商城, IM即时通讯，Spring Cloud Alibaba 等等，戳我加入学习，已有1300+小伙伴加入(早鸟价超低)

a44f76a7f448bb583e8a2c7a14b3ecbd.gif

e2e94816cc12731a6a7ea52cc2b9a342.jpeg

 
 
1968b89c37a2211e141f8c122f01207f.gif
 
 
 
 
 
 
1. 我的私密学习小圈子~
2. 如何用 Java 几分钟处理完 30 亿个数据?
3. Nginx 的 5 大应用场景，太实用了！
4. 炫技Groovy！SpringBoot中的动态编程实战
 
 
 
 
最近面试BAT，整理一份面试资料《Java面试BATJ通关手册》，覆盖了Java核心技术、JVM、Java并发、SSM、微服务、数据库、数据结构等等。
获取方式：点“在看”，关注公众号并回复 Java 领取，更多内容陆续奉上。
 
 
PS：因公众号平台更改了推送规则，如果不想错过内容，记得读完点一下“在看”，加个“星标”，这样每次新文章推送才会第一时间出现在你的订阅列表里。
点“在看”支持小哈呀，谢谢啦
					

																				

						

							上一篇：							
								ResponseVo							
						

						

							下一篇：							
								陈婉芬 太原 康军							
						

					

															

						

							推荐阅读