使用 Spring MVC 流式关闭资源答案

【问题标题】：Stream closeable resource with Spring MVC使用 Spring MVC 流式关闭资源
【发布时间】：2015-05-04 00:04:57
【问题描述】：

在阅读了this article 之后，我希望使用 Spring 将数据库查询结果直接流式传输到 JSON 响应，以确保恒定的内存使用（不会在内存中贪婪地加载 List）。

与文章中使用 Hibernate 所做的类似，我组装了一个 greetingRepository 对象，该对象基于 JdbcTemplate 返回数据库内容流。在该实现中，我在查询的ResultSet 上创建了一个迭代器，并按如下方式返回流：

return StreamSupport.stream(spliterator(), false).onClose(() -> {
  log.info("Closing ResultSetIterator stream");
  JdbcUtils.closeResultSet(resultSet);
});

即使用onClose() 方法保证如果在try-with-resources 构造中声明流，则底层ResultSet 将被关闭：

try(Stream<Greeting> stream = greetingRepository.stream()) {
  // operate on the stream
} // ResultSet underlying the stream will be guaranteed to be closed

但与文章中一样，我希望此流由自定义对象映射器（文章中定义的增强型MappingJackson2HttpMessageConverter）使用。如果我们把try-with-resources的需求放在一边，这是可行的：

@RequestMapping(method = GET)
Stream<GreetingResource> stream() {
  return greetingRepository.stream().map(GreetingResource::new);
}

但是正如那篇文章底部的一位同事评论的那样，这并没有考虑关闭底层资源。

在 Spring MVC 的上下文中，我如何才能从数据库一直流式传输到 JSON 响应并仍然保证 ResultSet 将被关闭？您能提供一个具体的示例解决方案吗？

【问题讨论】：

我不认为您的问题是资源泄漏：Spring 肯定会提交事务并释放连接，从而传递关闭您的结果集。但我预计会出现相反的问题：您如何管理连接在视图层中仍然存在？为此，我依赖 OpenSessionInViewInterceptor，这是特定于 Hibernate 的。
对，在我的测试场景中，我忘了使用事务，使用它们我无法再流式传输到视图层（另外我使用 MySQL，所以我不走运）。我的结论是，流式传输到视图层具有潜在的高效性和相当优雅的吸引力，但遗憾的是在实践中仍然难以使用。
不幸的是，对此的支持仍然不足。这是荒野的边界。不过，我确实希望它能流行起来，因为到目前为止，这个部门非常缺乏 Java 架构。
@MarkoTopolnik - 虽然OpenSessionInViewInterceptor 和...Filter 是特定于Hibernate 的，但OpenEntityManagerInViewInterceptor 和...Filter 支持所有提供JPA 实现的ORM。

标签： java spring-mvc java-8 spring-jdbc java-stream

【解决方案1】：

您可以创建一个构造以在序列化时推迟查询执行。此构造将以编程方式开始和结束事务。

public class TransactionalStreamable<T> {

    private final PlatformTransactionManager platformTransactionManager;

    private final Callable<Stream<T>> callable;

    public TransactionalStreamable(PlatformTransactionManager platformTransactionManager, Callable<Stream<T>> callable) {
        this.platformTransactionManager = platformTransactionManager;
        this.callable = callable;
    }

    public Stream stream() {
        TransactionTemplate txTemplate = new TransactionTemplate(platformTransactionManager);
        txTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
        txTemplate.setReadOnly(true);

        TransactionStatus transaction = platformTransactionManager.getTransaction(txTemplate);

        try {
            return callable.call().onClose(() -> {
                platformTransactionManager.commit(transaction);
            });
        } catch (Exception e) {
            platformTransactionManager.rollback(transaction);
            throw new RuntimeException(e);
        }
    }

    public void forEach(Consumer<T> c) {
        try (Stream<T> s = stream()){
            s.forEach(c);
        }
    }
}

使用专用的 json 序列化器：

JsonSerializer<?> transactionalStreamableSer = new StdSerializer<TransactionalStreamable<?>>(TransactionalStreamable.class, true) {
    @Override
    public void serialize(TransactionalStreamable<?> streamable, JsonGenerator jgen, SerializerProvider provider) throws IOException {
        jgen.writeStartArray();
        streamable.forEach((CheckedConsumer) e -> {
            provider.findValueSerializer(e.getClass(), null).serialize(e, jgen, provider);
        });

        jgen.writeEndArray();
    }
};

可以这样使用：

@RequestMapping(method = GET)
TransactionalStreamable<GreetingResource> stream() {
  return new TransactionalStreamable(platformTransactionManager , () -> greetingRepository.stream().map(GreetingResource::new));
}

当杰克逊将序列化对象时，所有工作都将完成。关于错误处理（例如使用控制器建议）可能是也可能不是问题。

【讨论】：