23如何在CompletableFuture异步线程中正确使用JPA？

你好，欢迎学习第 23 讲。通过前几讲对 Session 核心原理的学习，相信你已经可以解决实际工作中的一些疑难杂症了。这一讲我再给你举一个复杂点的例子，继续深度剖析如何利用 Session 原理解决复杂问题。那么，都有哪些问题呢？我们看一个例子。

CompletableFuture 的使用实际案例

在我们的实际开发过程中，难免会用到异步方法，我在这里列举一个异步方法的例子，经典地还原一些在异步方法里面经常会犯的错误。

我们模拟一个 Service 方法，通过异步操作，更新 UserInfo 信息，并且可能一个方法里面有不同的业务逻辑，会多次更新 UserInfo 信息，模拟的代码如下。

java

@RestController
public class UserInfoController {
   //异步操作必须要建立线程池，这个不多说了，因为不是本讲的重点，有兴趣的话你可以了解一下线程池的原理，我的Demo采用的是Spring异步框架字段的异步线程池
   @Autowired
   private  Executor executor;
   /**
    * 模拟一个业务service方法，里面有一些异步操作，一些业务方法里面可能修改了两次用户信息
    * @param name
    * @return
    */
   @PostMapping("test/async/user")
   @Transactional // 模拟一个service方法，期待是一个事务
   public String testSaveUser(String name) {
   CompletableFuture<Void> cf = CompletableFuture.runAsync(() -> {
      UserInfo user = userInfoRepository.findById(1L).get();
      //..... 此处模拟一些业务操作，第一次改变UserInfo里面的值；
      try {
         Thread.sleep(200L);// 加上复杂业务耗时200毫秒
      } catch (InterruptedException e) {
         e.printStackTrace();
      }
      user.setName(RandomUtils.nextInt(1,100000)+ "_first"+name); //模拟一些业务操作，改变了UserInfo里面的值
      userInfoRepository.save(user);
      //..... 此处模拟一些业务操作，第二次改变UserInfo里面的值；
      try {
         Thread.sleep(300L);// 加上复杂业务耗时300毫秒
      } catch (InterruptedException e) {
         e.printStackTrace();
      }
      user.setName(RandomUtils.nextInt(1,100000)+ "_second"+name);//模拟一些业务操作，改变了UserInfo里面的值
      userInfoRepository.save(user);
   }, executor).exceptionally(throwable -> {
      throwable.printStackTrace();
      return null;
   });
   //... 实际业务中，可能还有会其他异步方法，我们举这一个例子已经可以说明问题了
   cf.isDone();
   return "Success";
}
}

为了便于测试，我们在 UserInfoController 里面模拟了一个复杂点的 Service 方法，上面的代码很多是为了方便给你演示和做测试，实际工作中可能代码会不一样、会演变，但是你通过实质分析，就会发现解决思路是一样的。

我们在 testSaveUser 方法里面开了一个异步线程，异步线程采用 CompletableFuture 方法，在里面执行了两次 UserInfo 的 Save 操作，实际工作中可能不会有像我的 Demo 那么简单的 Save，因为我把中间的业务计算省去了，这不影响我们分析问题。

那么上面的代码问题的表象是什么呢？

表现出来的问题现状是什么样的？

那么实际工作中，我们如果写出来类似的代码，会发生什么样的问题呢？

整个请求非常正常，永远都是 200；也没有任何报错信息，但是发现数据库里面第二次的 save(user) 永远不生效，永远不会出现 name 包含 "_second" 的记录，这个是必现的；
整个请求非常正常，永远都是 200；也没有任何报错信息，有的时候会发现数据库里面没有任何变化，甚至第一次 save(user) 都没有生效，但是这个是偶发的。

实际工作中我们肯定会通过 QA 或者自己多次测试，发现以上现象就会感觉非常奇怪，那么我们来分步拆解一下，看看怎么解决？

步骤拆解

有一定经验的开发者，遇到类似问题，第一步应该想到是不是发生什么异常了？日志信息去哪里了？那么我们需要先看一下 CompletableFuture 的用法，是不是发生异常的时候我们漏掉了什么环节？

CompletableFuture 使用最佳实践

CompletableFuture 主要的功能是实现了 Future 和 CompletionStage 的接口，主要的方法如下述代码所示。

java

//通过给定的线程池，异步执行 Runnable方法，不带返回结果
public static CompletableFuture<Void> 	runAsync(Runnable runnable, Executor executor)
//通过给定的线程池，异步执行Runnable方法，带返回结果
public static <U> CompletableFuture<U> 	supplyAsync(Supplier<U> supplier, Executor executor)
//当上面的异步方法执行完之后需要执行的回调方法
public CompletableFuture<Void> 	thenAccept(Consumer<? super T> action)
//阻塞等待 future执行完结果
boolean isDone();
//阻塞获取结果
V get();
//当异步操作发生异常的时候执行的方法
public CompletionStage<T> exceptionally(Function<Throwable, ? extends T> fn);

以上我只是列举了一些和我们案例相关的关键方法，而 CompletableFuture 还有更多的方法，其功能也非常强大，所以一般开发过程中用此类的场景还非常多。

其实上面的 Demo 只是利用 runAsync 做了异步操作，并利用 isDone 做了阻塞等待的动作，而没有使用 Exceptionally 处理异常信息。

所以如果我们想打印异常信息，基本上可以利用 Exceptionally。我们改进一下 Demo 代码，把异常信息打印一下，看看是否发生了异常。变动的代码如下所示。

java

CompletableFuture<Void> cf = CompletableFuture.runAsync(() -> {
   ......这里的代码不变，我们不做copy了
}, executor).exceptionally(e -> {
   log.error(e);//把异常信息打印出来
   return null;
});

那么我们再请求上面的 Controller 方法的时候，发现控制台就会打印出如下所示的 Error 信息。

java

java.util.concurrent.CompletionException: org.springframework.orm.ObjectOptimisticLockingFailureException: Object of class [com.example.jpa.demo.db.UserInfo] with identifier [1]: optimistic locking failed; nested exception is org.hibernate.StaleObjectStateException: Row was updated or deleted by another transaction (or unsaved-value mapping was incorrect) : [com.example.jpa.demo.db.UserInfo#1]
	at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:314)
	at java.base/java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:319)
	at java.base/java.util.concurrent.CompletableFuture$AsyncRun.run$$$capture(CompletableFuture.java:1739)
	at java.base/java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1167)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:641)
	at java.base/java.lang.Thread.run(Thread.java:844)
Caused by: org.springframework.orm.ObjectOptimisticLockingFailureException: Object of class [com.example.jpa.demo.db.UserInfo] with identifier [1]: optimistic locking failed; nested exception is org.hibernate.StaleObjectStateException: Row was updated or deleted by another transaction (or unsaved-value mapping was incorrect) : [com.example.jpa.demo.db.UserInfo#1]
	at org.springframework.orm.jpa.vendor.HibernateJpaDialect.convertHibernateAccessException(HibernateJpaDialect.java:337)
	at org.springframework.orm.jpa.vendor.HibernateJpaDialect.translateExceptionIfPossible(HibernateJpaDialect.java:255)
	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
	at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:212)
	at com.sun.proxy.$Proxy116.save(Unknown Source)
	at com.example.jpa.demo.web.UserInfoController.lambda$testSaveUser$0(UserInfoController.java:57)
	at java.base/java.util.concurrent.CompletableFuture$AsyncRun.run$$$capture(CompletableFuture.java:1736)
	... 4 more

通过报错信息，可以发现其实就是发生了乐观锁异常，导致上面实例中的第二次 save(user) 必然失败；而第一次 save(user) 的失败，主要是因为在并发的情况下有其他请求线程改变了 UserInfo 的值，也就是改变了 Version。

我们来看一下完整的 UserInfo 对象实体。

java

@Entity
@Data
@SuperBuilder
@AllArgsConstructor
@NoArgsConstructor
@ToString(callSuper = true)
@Table
@EntityListeners({AuditingEntityListener.class})
public class UserInfo{
   @Id
   @GeneratedValue(strategy= GenerationType.AUTO)
   private Long id;
   @Version
   private Integer version;
   @CreatedBy
   private Integer createUserId;
   @CreatedDate
   private Instant createTime;
   @LastModifiedBy
   private Integer lastModifiedUserId;
   @LastModifiedDate
   private Instant lastModifiedTime;
   private String name;
   private Integer ages;
   private String lastName;
   private String emailAddress;
   private String telephone;
}

看过前面课时的同学应该知道，我们通过 @Version 乐关锁机制就是防止数据被覆盖；而实际生产过程中其实很难发现类似问题。

所以当我们使用任何的异步线程处理框架的时候，一定要想好异常情况下怎么打印日志，否则就像黑洞一样，完全不知道发生了什么。

那么既然知道发生了乐观锁异常，这里就有个疑问了：我们不是在 UserInfoController 的 testSaveUser 方法上面加了 @Transaction 的注解了吗？为什么事务没有回滚？

通过日志查看事务的执行过程

我们看看异步请求的情况下，事务应该怎么做呢？先打开事务的日志，看看上面方法的事务执行过程是什么样的。

java

## 我们在db的连接中开启logger=Slf4JLogger&profileSQL=true看一下每个事务里执行的sql有哪些
spring.datasource.url=jdbc:mysql://localhost:3306/test?logger=Slf4JLogger&profileSQL=true
## 打开下面这些类的日志级别，观察一下事务的开启和关闭时机
logging.level.org.springframework.orm.jpa=DEBUG
logging.level.org.springframework.transaction=DEBUG
logging.level.org.springframework.orm.jpa.JpaTransactionManager=trace
logging.level.org.hibernate.engine.transaction.internal.TransactionImpl=DEBUG

再请求一下刚才的测试接口：POSThttp://127.0.0.1:8087/test/async/user?name=jack就会产生下图所示的日志。

先看一下上半部分，通过日志我们可以看到，首先执行这个方法的时候开启了两个事务，分别做如下解释。

线程 1：[nio-8087-exec-1] 开启了 UserInfoController.testSaveUser 方法上面的事务，也就是 http 的请求线程，开启了一个 Controller 请求事务。这是因为我们在 testSaveUser 的方法上面加了 @Transaction 的注解，所以开启了一个事务。

而通过日志我们也可以发现，事务 1 里面什么都没有做，随后就进行了 Commit 操作，所以我们可以看得出来，默认不做任何处理的情况下，事务是不能跨线程的。每个线程里面的事务相互隔离、互不影响。

线程 2：[ task-1]，通过异步线程池开启了 SimpleJpaRepository.findById 方法上面的只读事务。这是因为默认的 SimpleJpaRepository 类上面加了 @Transaction(readOnly=true) 产生的结果。而我们通过 MySQL 的日志也可以看得出来，此次事务里面只做了和我们代码相关的 select user_info 的操作。

我们再看一下后半部分的日志，如图所示。

通过后半部分日志，我们可以看到两次 save(user) 方法，也分别开启了各自的事务，这是因为 SimpleJpaRepository.save 方法上面有 @Transaction 注解起了作用，而第二次事务因为 JPA 的实现方法判断了数据库这条数据的 Version 和我们 UserInfo 的对象中的 Version 不一致，从而第二次进行了回滚操作。

两次 save(user) 的操作里面分别有一次 Select 和 Update 语义，正是我们之前所说的 Save 方法的原理。两次事务，分别开启了两个 Session，所以对象对于这两次 Session 来说分别是从游离态（Detached）转成持久态（Persistent）的过程，所以两个独立的事务里面，一次 Select，一次 Update。

通过日志可以看到，上面一个简单的方法中一共发生了四次事务，都是采用的默认隔离级别和传播机制。那么如果我们想让异步方法里面只有一个事务应该怎么办呢？

异步事务的正确使用方法

既然我们知道异步方法里面的事务是独立的，那么直接把异步的代码块用独立的事务包装起来即可，做法有如下几种。

第一种处理方法：把其中的异步代码块，移到一个外部类里面。我们这里放到 UserInfoService 中，同时方法中加上 @Transaction 注解用来开启事务，加上 @Retryable 注解进行乐观锁重试，代码如下。

java

//加上事务，这样可以做到原子性，解决事务加到异常方法之外没有任何作用的问题
@Transactional
//加上重试机制，这样当我们发生乐观锁异常的时候，重新尝试下面的逻辑，减少请求的失败次数
@Retryable(value = ObjectOptimisticLockingFailureException.class,backoff = @Backoff(multiplier = 1.5,random = true))
public void businessUserMethod(String name) {
   UserInfo user = userInfoRepository.findById(1L).get();
   //..... 此处模拟一些业务操作，第一次改变UserInfo里面的值；
   try {
      Thread.sleep(200L);// 加上复杂业务耗时200毫秒
   } catch (InterruptedException e) {
      e.printStackTrace();
   }
   user.setName(RandomUtils.nextInt(1,100000)+ "_first"+name); //模拟一些业务操作，改变了UserInfo里面的值
   userInfoRepository.save(user);
   //..... 此处模拟一些业务操作，第二次改变UserInfo里面的值；
   try {
      Thread.sleep(300L);// 加上复杂业务耗时300毫秒
   } catch (InterruptedException e) {
      e.printStackTrace();
   }
   user.setName(RandomUtils.nextInt(1,100000)+ "_second"+name);//模拟一些业务操作，改变了UserInfo里面的值
   userInfoRepository.save(user);
}

那么 Controller 里面只需要变成如下写法即可。

java

/**
 * 模拟一个业务service方法，里面有一些异步操作，一些业务方法里面可能修改了两次用户信息
 * @param name
 * @return
 */
@PostMapping("test/async/user")
@Transactional // 模拟一个service方法，期待是一个事务
public String testSaveUser(String name) {
   CompletableFuture<Void> cf = CompletableFuture.runAsync(() -> {
      userInfoService.businessUserMethod(name);
   }, executor).exceptionally(e -> {
      log.error(e);//把异常信息打印出来
      return null;
   });
   //... 实际业务中，可能还有会其他异步方法，我们举这个例子已经可以说明问题了
   cf.isDone();
   return "Success";
}

我们再次发起一下请求，看一下日志。

通过上图的日志，我们可以知道两个重要信息：

这个时候只有 UserInfoServiceImpl.businessUserMethod 开启了一个事务，这是因为 findById 和 Save 方法中，事务的传播机制都是"如果存在事务就利用当前事务"的原理，所以就不会像我们上面一样创建四次事务了；
而此时两次 save(user) 只产生了一个 update 的 sql 语句，并且也很难出现乐观锁异常了，因为这是 Session 的机制，将两次对 UserInfo 实体的操作进行了合并；所以当我们使用 JPA 的时候某种程度上也会降低 db 的压力，增加代码的执行性能。

而另外一个侧论，就是当事务的生命周期执行越快的时候，发生异常的概率就会越低，因为可以减少并发处理的机会。

第二种处理方法：可以利用"19 | 如何搞清楚事务、连接池的关系？正确配置是怎样的？"讲过的 TransactionTemplate 方法开启事务，这里不再重复讲述了。

第三种处理方法：我们可以建一个自己的 TransanctionHelper，并带上重试机制，代码如下：

java

/**
 * 利用spring进行管理
 */
@Component
public class TransactionHelper {
    /**
     * 利用spring 机制和jdk8的Consumer机制实现只消费的事务
     */
    @Transactional(rollbackFor = Exception.class) //可以根据实际业务情况，指定明确的回滚异常
    @Retryable(value = ObjectOptimisticLockingFailureException.class,backoff = @Backoff(multiplier = 1.5,random = true))
    public void transactional(Consumer consumer,Object o) {
        consumer.accept(o);
    }
}

那么 Controller 里面的写法可以变成如下方式，也可以达到同样效果。

java

@PostMapping("test/async/user")
public String testSaveUser(String name) {
   CompletableFuture<Void> cf = CompletableFuture.runAsync(() -> {
      transactionHelper.transactional((param)->{ // 通过lambda实现事务管理
         UserInfo user = userInfoRepository.findById(1L).get();
         //..... 此处模拟一些业务操作，第一次改变UserInfo里面的值；
         try {
            Thread.sleep(200L);// 加上复杂业务耗时200毫秒
         } catch (InterruptedException e) {
            e.printStackTrace();
         }
         user.setName(RandomUtils.nextInt(1,100000)+ "_first"+name); //模拟一些业务操作，改变了UserInfo里面的值
         userInfoRepository.save(user);
         //..... 此处模拟一些业务操作，第二次改变UserInfo里面的值；
         try {
            Thread.sleep(300L);// 加上复杂业务耗时300毫秒
         } catch (InterruptedException e) {
            e.printStackTrace();
         }
         user.setName(RandomUtils.nextInt(1,100000)+ "_second"+name);//模拟一些业务操作，改变了UserInfo里面的值
         userInfoRepository.save(user);
      },name);
   }, executor).exceptionally(e -> {
      log.error(e);//把异常信息打印出来
      return null;
   });
   //... 实际业务中，可能还有会其他异步方法，我们举一个例子已经可以说明问题了
   cf.isDone();
   return "Success";
}

这种方式主要是通过 Lambda 表达式解决事务问题。

总之，不管是以上哪种方法，都可以解决我们所说的异步事务的问题。所以搞清楚事务的背后实现逻辑，就很容易解决类似问题了。

还有一个问题就是，为什么当异步方法中是同一个事务的时候，第二次 save(user) 就成功了？而异步代码块里面的两个 save(user) 分别在两个事务里面，第二次就不成功了呢？我们利用前两个课时讲过的 Persistence Context 和实体的状态来分析一下。

Session 的机制与 Repository.save(entity) 是什么关系？

我们在学习 Persistence Context 的时候，知道 Entity 有不同的状态。

在一个 Session 里面，如果我们通过 findById(id) 得到一个 Entity，它就会变成 Manager（persist）持久态。那么同一个 Session 里面，同一个 Entity 多次操作 Hibernate 就会进行 Merge 操作。

所以上面的实例中，当我们在 businessUserMethod 方法上面加 @Transaction 的时候，会造成异步代码的整块逻辑处于同一个事务里面，而按照我们上一讲介绍的 Session 原理，同一个事务就会共享同一个 Session，所以同一个事务里面的 findById、save、save 的多次操作都是同一个实例。

什么意思呢？我们可以通过设置 Debug 断点，查看一下对象的内存对象地址是否一样，就可以看得出来。如下图所示，findById 之后和两次 save 之后都是同一个对象。

而如果我们跨 Session 传递实体对象，那么在一个 Session 里面持久态的对象，对于另外一个 Session 来说就是一个Detached（游离态）的对象。

而根据 Session 里面的 Persistenc Context 的原理，一旦这个游离态的对象进行 db 操作，Session 会 Copy 一个新的实体对象。也就是说，当我们不在异步代码中加事务的时候，即去掉异步代码块businessUserMethod 方法中的@Transaction 注解，findById 之后就会产生一个新的事务、新的 Session，那么返回的就是对象 1；第一次 Save 之后，由于又是一个新的事务、新的 Session，那么返回的实体 u2 就是对象 2。

我们知道这个原理之后，对代码做如下改动。

java

//  @Transactional 去掉事务
   public void businessUserMethod(String name) {
      UserInfo user = userInfoRepository.findById(1L).get();
      user.setName(RandomUtils.nextInt(1,100000)+ "_first"+name); //模拟一些业务操作，改变了UserInfo里面的值
      UserInfo u2 = userInfoRepository.save(user);
      user.setName(RandomUtils.nextInt(1,100000)+ "_second"+name); //模拟一些业务操作，改变了UserInfo里面的值
      UserInfo u3 = userInfoRepository.save(u2);// 第二次save采用第一次save的返回结果，这样里面带有了最新的version的值，所以也就会保存成功
}

异步里面调用这个方法也是成功的，因为乐观锁的原理是 Version 变了，我们用最新的对象，也就是最新的 Version 就可以了。

我们设置一个断点看一下 user、u2、u3 在不同的 Session 作用域之后，就变成不同的实例了，如下所示。

问题分析完了，那么这些内容带给我们哪些思考呢？

思考

在上面 Demo 中的异步场景下设置 open-in-view 等于 true / false，会对上面的测试结果有影响吗？

答案是肯定没有影响的，spring.jpa.open-in-view 的本质还是开启 Session，而保持住 Session 的本质还是利用 ThreadLocal，也就是必须为同一个线程的情况下才适用。所以异步场景不受 spring.jpa.open-in-view 控制。

如果是大量的异步操作 db connection 的持有模式，应该配置成哪一种比较合适？

答案是DELAYED_ACQUISITION_AND_RELEASE_AFTER_TRANSACTION，因为这样可以做到对 db 连接最大的利用率。用的时候就获取，事务提交完就释放，这样就不用关心业务逻辑执行多长时间了。

总结

上面的例子折射出来的是一些 JPA 初学者最容易犯的错误，我们通过前几讲对原理知识的学习，解决了工作中最常见、最容易犯错的，如异步问题和事务问题。其中关键的几个问题你一定要好好思考，尤其是在开发业务代码的时候。

我们的一个请求，开启了几次事务？在什么时机开始的？
我们的一个请求，开启了几次 Session？在什么时机开启的？
事务和 Session 分别会对实体的状态有什么影响？

上面的几个问题是对一个高级 Java 工程师最基础的要求，如果你想晋级资深开发工程师，还需要知道：

我们的一个请求，对 db 连接池里面的连接持有时间是多久？
我们的一个请求，性能指标都有哪些决定因素？

针对以上问题，你可以回过头去文中找答案，并且希望你深入钻研，遇到问题做到心中有数。

本讲内容就到这里了，下一讲我们来聊聊 Lazy 的核心原理和问题。欢迎你对本讲内容提出问题和建议，如果本专栏对你有帮助，就动动手指分享吧。我们下一讲再见。

点击下方链接查看源码（不定时更新）
https://github.com/zhangzhenhuajack/spring-boot-guide/tree/master/spring-data/spring-data-jpa

23如何在CompletableFuture异步线程中正确使用JPA？ ​

CompletableFuture 的使用实际案例 ​

表现出来的问题现状是什么样的？ ​

步骤拆解 ​

CompletableFuture 使用最佳实践 ​

通过日志查看事务的执行过程 ​

异步事务的正确使用方法 ​

Session 的机制与 Repository.save(entity) 是什么关系？ ​

思考 ​

总结 ​

23如何在CompletableFuture异步线程中正确使用JPA？

CompletableFuture 的使用实际案例

表现出来的问题现状是什么样的？

步骤拆解

CompletableFuture 使用最佳实践

通过日志查看事务的执行过程

异步事务的正确使用方法

Session 的机制与 Repository.save(entity) 是什么关系？

思考

总结