使用 Spring MVC 流式传输可关闭资源

读完后本文 https://www.airpair.com/java/posts/spring-streams-memory-efficiency，我希望使用 Spring 将数据库查询结果直接流式传输到 JSON 响应，以确保恒定的内存使用量（不会贪婪地加载List在记忆中）。

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

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

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

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将被关闭？您能提供一个具体的示例解决方案吗？

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

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));
}

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