锁超时发现parallelStream并行流线程上下文坑解决

detached entity passed to persist问题

就我之前因为在处理jpa持久化对象上下文时，spring jpa关于线程池异步执行导致detached entity passed to persist问题排查和解决

我这边有个批量插入用户OpenUser和应用OpenApp关联关系数据的操作，由于耗时较长时间，所以准备用线程池异步执行操作，然而却遇到了一个jpa的detached entity passed to persist问题，我这边的操作是批量保存一个OpenAppUser关联关系表，所以需要先获得对应OpenUser和OpenApp的引用，再设置到关联对象OpenAppUser里，然后在保存，我这边是先通过userRepository.findById(userId)获取到OpenUser，然后openAppUser.setOpenUser(openUser)，在执行appUserRepository.save(openAppUser);时发生了如标题上的错误，说是OpenUser对象处于游离态，无法保存。

经过排查，我这边是因为OpenAppUser类里设置了@ManyToOne(cascade = CascadeType.ALL)级联OpenUser，所以在保存OpenAppUser的时候会级联操作OpenUser，本来在没有开线程异步的情况下，因为OpenUser之前通过findById查出来了，所以在jpa的PersistenceContext里是有该OpenUser的脱管对象的，这时候就不会报错，而在线程异步的情况下context里确没有该脱管对象了

（这里说明一下，为啥不开线程有，开了线程没有？）因为spring-boot默认jpa.open-in-view=true,会使用ThreadLocal在当前线程里保存EntityManager上下文信息，所以在整个controller里都是使用的同一个context

PersistenceContext持久性上下文有两种类型

• 事务范围的持久性上下文；当我们在事务中执行任何操作时，EntityManager 会检查持久性上下文。 如果存在，则将使用它。否则，它将创建一个持久性上下文
• 扩展范围的持久性上下文；扩展持久性上下文可以跨越多个事务。我们可以在没有事务的情况下持久化实体，但不能在没有事务的情况下刷新它。

在@PersistenceContext注解里type可以指定范围：PersistenceContextType.TRANSACTION；PersistenceContextType.EXTENDED

而当我们用线程池异步的时候，拿不到之前的EntityManager的配置信息，而spring jpa repository默认的方法上都会自带一个事务，所以在执行完userRepository.findById(userId)获取到OpenUser之后，会commit，而commit操作会clear掉EntityManager里保存的脱管对象OpenUser，等到appUserRepository.save(openAppUser);保存的时候，由于引用的OpenUser已经没有在PersistenceContext上下文里了，不是脱管对象了（具体可以看EntityState entityState = getEntityState( entity, entityName, entityEntry, source );里面的实现，有几种判断条件，是不是脱管对象，有没有id、version等等属性），就会报detached entity passed to persist这个异常

所以根据实际情况，我们只要参考open-in-view=true产生对应的OpenEntityManagerInViewInterceptor拦截器改造一下自己线程里的PersistenceContext上下文生效范围，就可以解决该异常了

parallelStream并行流

parallelStream并行流给我的印象就是会读不到父线程的上下文的，所以应该在父线程里的事务和在parallelStream里的事务应该是区分的，而不是共用同一个事务的，然而今天因为一个锁超时的问题，发现并没有那么简单，下面我们一步一步来验证。

锁超时场景

具体的业务我不讲了，就说下伪代码

@PostMapping("/saveUser")
@Transactional
public void saveUser(@RequestBody List<Complex> list) {
    list.parallelStream().forEach(complex->{
        Integer appId = complex.getAppId();
        Integer userId = complex.getUserId();
        GeneratedKeyHolder keyHolder = new GeneratedKeyHolder();
        String sql = "insert ignore into open_app_user (app_id, open_id, user_status, creator, modifier, create_time, modify_time, status, version) values ("+appId+","+userId+",0,1,1,now(),now(),1,1)";
        int id = jdbcTemplate.update(con -> con.prepareStatement(sql, Statement.RETURN_GENERATED_KEYS), keyHolder);  
    });
    //todo 业务逻辑...
}

这里我有个批量保存的逻辑，需要先保存一个中间表open_app_user表（该表app_id和open_id是联合唯一键）获得id，拿到用户的open_app_user_id后再进行其他业务逻辑，这里按我原来的理解是虽然我在controller的方法上加了@Transactional注解，但是parallelStream里的事务应该都是独立的，不会是同一个事务，所以即使有数据重复，第一个线程插入后，第二个线程也只会插入失败（不会报错，因为我加了ignore），所以即使并行也不会有问题的，然而却发生了锁超时的问题。

查看锁超时以及定位的操作可以看我前面的文章，通过查找mysql的 https://www.jb51.net/article/259480.htm

select * from information_schema.INNODB_TRX;
select * from performance_schema.data_lock_waits;
select * from performance_schema.data_locks;

定位到了这里，然而我也百思不得其解，为啥会锁超时呢，这里应该都是马上执行就马上释放了啊，难道是其中的事务没有提交？

因为现在都是spring的声明式事务管理，spring是在有@Transactional注解的情况下，执行完了才提交事务，在没有@Transactional注解的情况下，每个方法都差不多可以理解成原子，比如我上面的jdbcTemplate.update()这个方法就是一个事务，执行完了就直接提交事务了。

验证

因为spring是把事务上下文放在ThreadLocal里了，主要是用TransactionSynchronizationManager这个类来管理，所以我写了一个demo来进行验证

@GetMapping("/get")
@Transactional
public String get() {
    List<Complex> list = new ArrayList<>();
    for (int i = 0; i < 10; i++) {
        list.add(new Complex(1, 1));
    }
    list.parallelStream().forEach(complex->{
        Map<Object, Object> resourceMap = TransactionSynchronizationManager.getResourceMap();
        System.err.println("count:"+resourceMap.size());
        Integer appId = complex.getAppId();
        Integer userId = complex.getUserId();
        String sql = "insert ignore into open_app_user (app_id, open_id, user_status, creator, modifier, create_time, modify_time, status, version) values ("+appId+","+userId+",0,1,1,now(),now(),1,1)";
        int update = jdbcTemplate.update(sql);
    });
    return "hello, world! ";
}

有趣的事情发生了，我在注释掉@Transactional注解时，代码里resourceMap.size()返回的内容是竟然不一样，因为我的list有10条记录，差不多就是10个并行，然而我的输出却是：

count:1
count:0
count:0
count:0
count:0
count:0
count:0
count:0
count:0
count:0

没有注释掉@Transactional注解时，输出是：

count:2
count:0
count:0
count:0
count:0
count:0
count:0
count:0
count:0
count:0

并且还会出现锁超时的现象，奇怪的地方就是为啥我用的parallelStream会有线程上下文里的值，我并没有做什么操作，而且10个并行里只有一个（这里并不是说明固定只有一次，下面会说明）获得了线程上下文的信息

测试

我又进一步测试，伪代码改成：

@GetMapping("/get")
public void get() {
    List<Complex> list = new ArrayList<>();
    for (int i = 0; i < 10; i++) {
        list.add(new Complex(1, 1));
    }
    ThreadLocal local = new ThreadLocal();
    local.set("parent_set_value");
    list.parallelStream().forEach(complex->{
        System.err.println(local.get());
    });
}

结果如我所料，输出为：

parent_set_value
null
null
null
null
null
null
null
null
null

使用parallelStream并不完全都是另开了线程，其中有一个是属于主线程的，可以使用System.err.println(Thread.currentThread().getName());查看当前线程的名称，我发现parallelStream会把当前主线程也作为一个执行线程去执行任务

后面我再去了解了一下parallelStream的实现，在这个方法上的注解里第一句话有个单词是possibly，是“可能”返回并行流，原来参与并行处理的线程有主线程以及ForkJoinPool中的worker线程，所以parallelStream是有两种情况的，一是可能只一个线程并发执行，二是多个线程并行执行，而我这里导致锁超时，就是因为用到了主线程，所以在并行插入的时候，有个处理有事务上下文，导致一直没有提交事务（@Transactional注释方法的方法没有跑完，这里也不可能跑完），所以其他线程的插入就一直等待这个，产生了锁超时报错

以上就是锁超时发现parallelStream并行流线程上下文坑解决的详细内容，更多关于parallelStream并行流线程坑的资料请关注脚本之家其它相关文章！

最新评论