Spring Data JPA 流查询方法导致有关事务的异常

Question

如果我使用返回Stream的Spring Data仓库方法，我总会得到以下异常：

org.springframework.dao.InvalidDataAccessApiUsageException: You're trying to execute a streaming query method without a surrounding transaction that keeps the connection open so that the Stream can actually be consumed. Make sure the code consuming the stream uses @Transactional or any other way of declaring a (read-only) transaction.
    org.springframework.data.jpa.repository.query.JpaQueryExecution$StreamExecution.doExecute(JpaQueryExecution.java:338)
    org.springframework.data.jpa.repository.query.JpaQueryExecution.execute(JpaQueryExecution.java:85)
    org.springframework.data.jpa.repository.query.AbstractJpaQuery.doExecute(AbstractJpaQuery.java:116)
    org.springframework.data.jpa.repository.query.AbstractJpaQuery.execute(AbstractJpaQuery.java:106)
    org.springframework.data.repository.core.support.RepositoryFactorySupport$QueryExecutorMethodInterceptor.doInvoke(RepositoryFactorySupport.java:483)
    org.springframework.data.repository.core.support.RepositoryFactorySupport$QueryExecutorMethodInterceptor.invoke(RepositoryFactorySupport.java:461)
    org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
    org.springframework.data.projection.DefaultMethodInvokingMethodInterceptor.invoke(DefaultMethodInvokingMethodInterceptor.java:61)
    org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
    org.springframework.transaction.interceptor.TransactionInterceptor$1.proceedWithInvocation(TransactionInterceptor.java:99)
    org.springframework.transaction.interceptor.TransactionAspectSupport.invokeWithinTransaction(TransactionAspectSupport.java:282)
    org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:96)
    org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
    org.springframework.dao.support.PersistenceExceptionTranslationInterceptor.invoke(PersistenceExceptionTranslationInterceptor.java:136)
    org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
    org.springframework.data.jpa.repository.support.CrudMethodMetadataPostProcessor$CrudMethodMetadataPopulatingMethodInterceptor.invoke(CrudMethodMetadataPostProcessor.java:133)
    org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
    org.springframework.aop.interceptor.ExposeInvocationInterceptor.invoke(ExposeInvocationInterceptor.java:92)
    org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
    org.springframework.data.repository.core.support.SurroundingTransactionDetectorMethodInterceptor.invoke(SurroundingTransactionDetectorMethodInterceptor.java:57)
    org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
    org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:213)
    com.sun.proxy.$Proxy201.findByPodcast(Unknown Source)
    <my controller class>$$Lambda$118/2013513791.apply(Unknown Source)

然而，该代码应该在事务中执行。我已经：

• 使用了OpenSessionManagerInViewFilter
• 启用了声明式事务管理（在我的根上下文配置中使用 @EnableTransactionManagement ），并在控制器类和请求方法中注释了 @Transactional

我还尝试在一个TransactionTemplate中包装代码，并将结果收集到 List 中，以避免事务超出范围，但这仍然没有起作用。

控制器方法：

返回:

代码应该在事务中执行。已经使用了OpenSessionManagerInViewFilter，启用了声明式事务管理，并在控制器类和请求方法中注释了 @Transactional。尝试过在 TransactionTemplate 中包装代码，并将结果收集到 List 中，以避免事务越界，但这仍然没用。

@RequestMapping ( "/pod/{id}" )
@Transactional
public Stream<RSSPodcastItem> podItems (@PathVariable("id") UUID id)
{
    return pods.get (id).map (items::findByPodcast).orElseThrow (() -> new RuntimeException ("failed"));
}
@RequestMapping ( "/podlist/{id}" )
@Transactional
public List<RSSPodcastItem> podItemsList (@PathVariable("id") UUID id)
{
    return tt.execute (ts -> 
        pods.get (id).map (items::findByPodcast).orElseThrow (() -> new RuntimeException ("failed"))
        .collect (Collectors.toList()));
}

上下文根配置类：

@Configuration
@ComponentScan ( ... my package names ...)
@EnableTransactionManagement
@EnableJpaRepositories( ... package with repositories ...)
public class SharedConfig
{
    @Bean
    public DataSource dataSource ()
    {
         // .... snipped
    }

    @Bean
    EntityManagerFactory entityManagerFactory ()
    {
        LocalContainerEntityManagerFactoryBean entityManagerFactoryBean = new LocalContainerEntityManagerFactoryBean ();
        entityManagerFactoryBean.setDataSource (dataSource());
        entityManagerFactoryBean.setJpaVendorAdapter (new HibernateJpaVendorAdapter ());
        entityManagerFactoryBean.setPackagesToScan ( ... package with entities ...);
        entityManagerFactoryBean.setJpaPropertyMap (hibernateProperties());
        entityManagerFactoryBean.afterPropertiesSet ();
        return entityManagerFactoryBean.getObject ();
    }

    @Bean
    JpaTransactionManager transactionManager ()
    {
        JpaTransactionManager transactionManager = new JpaTransactionManager ();
        transactionManager.setEntityManagerFactory (entityManagerFactory());

        return transactionManager;
    }

    @Bean
    TransactionTemplate transactionTemplate (JpaTransactionManager tm)
    {
        return new TransactionTemplate (tm);
    }

    @Bean
    Map<String, ?> hibernateProperties ()
    {
        Map<String, Object> m = new HashMap<> ();
        m.put ("hibernate.dialect", MySQL5Dialect.class);
        m.put ("hibernate.dialect.storage_engine", "innodb");
        boolean devSystem = isDevSystem ();
        m.put ("hibernate.hbm2ddl.auto", devSystem ? "update" : "create-only");  // will need to handle updates by hand on live system, but creation is OK.
        m.put ("hibernate.show_sql", "" + devSystem);
        m.put ("hibernate.cache.use_second_level_cache", "" + !devSystem);
        return m;
    }

有什么建议，这里出了什么问题吗？

- Jules

---

我的回答有帮到你吗？ - Cepr0

Answer 1

根据这篇文章所述：库客户端可以在try-with-resources块中使用方法调用的结果。

而Spring Data的参考文档表明：流可能会包装底层数据存储特定的资源，因此必须在使用后关闭。 您可以使用close()方法手动关闭流，也可以使用Java 7中的try-with-resources块。

因此，我认为您应该将流包装到try-with-resource块中，并像异常建议的那样设置只读事务，类似于以下内容：

@RequestMapping("/pod/{id}")
@Transactional(readOnly = true)
public Stream<RSSPodcastItem> podItems (@PathVariable("id") UUID id) {
    try (Stream<RSSPodcastItem> items = repository.findByPodcast(...)) {
        return items...;
    }
}

更多信息：Spring Data - Java 8示例

（此链接为相关的IT技术内容，本助手无法作出解释）

- Cepr0

---

如果在返回之前关闭流，MVC应该如何使用它？ - Jules

@Jules 我认为它是在Spring 5中实现的 - https://dev59.com/oVkS5IYBdhLWcg3wXFnc#44509954 - Cepr0

那个Spring Data - Java 8的示例链接已经失效。 - Kristijan Iliev

Answer 2

这对我有用：

// 存储库：

@Query("HQL Query here")
Stream<Object[]> findEntities();

// 服务：

@Autowired
private PlatformTransactionManager transactionManager;

@Autowired
private EntityRepository entityRepository;

.....

TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
transactionTemplate.execute(new TransactionCallbackWithoutResult() {
    @Override
    protected void doInTransactionWithoutResult(TransactionStatus transactionStatus) {
       try (Stream<Object[]> entityStream =entityRepository.findEntities()) {
        entityStream.forEach(entityObjs -> {
            // Here deal with each entity as entityObjs
        });
    }
  }
);

- user14952813