Stream resource resource với Spring MVC

Question

Sau khi đọc this article, tôi muốn sử dụng Spring để truyền kết quả truy vấn cơ sở dữ liệu trực tiếp tới phản hồi JSON để đảm bảo sử dụng bộ nhớ liên tục (không tải một bộ tham số List trong bộ nhớ).

Tương tự như những gì được thực hiện trong bài viết với Hibernate, tôi đã ráp một đối tượng greetingRepository trả về luồng nội dung cơ sở dữ liệu dựa trên JdbcTemplate. Trong việc thực hiện điều đó, tôi tạo ra một iterator trong truy vấn ResultSet, và tôi trả lại dòng như sau:

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

tức là với một phương pháp onClose() đảm bảo rằng cơ bản ResultSet sẽ đóng cửa nếu dòng được khai báo trong một cấu trúc try-with-resources :

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

Nhưng như trong bài viết này, tôi muốn dòng này để được tiêu thụ bởi một ánh xạ đối tượng tùy chỉnh (các tăng cường MappingJackson2HttpMessageConverter định nghĩa trong bài viết). Nếu chúng ta lấy try-with-resources cần sang một bên, đây là khả thi như sau:

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

Tuy nhiên như một đồng nghiệp nhận xét ở dưới cùng của bài viết đó, điều này không chăm sóc đóng nguồn lực tiềm ẩn.

Trong ngữ cảnh của Spring MVC, làm thế nào tôi có thể truyền từ cơ sở dữ liệu tất cả các cách vào một phản ứng JSON và vẫn đảm bảo rằng ResultSet sẽ bị đóng? Bạn có thể cung cấp một giải pháp ví dụ cụ thể không?

Answer 1

Bạn có thể tạo cấu trúc để hoãn thực thi truy vấn tại thời gian tuần tự hóa. Cấu trúc này sẽ bắt đầu và kết thúc chương trình giao dịch.

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

Sử dụng một serializer json chuyên dụng:

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

Mà có thể được sử dụng như thế này:

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

Tất cả các công việc sẽ được thực hiện khi jackson sẽ serialize đối tượng. Nó có thể là hoặc không phải là một vấn đề liên quan đến việc xử lý lỗi (ví dụ: sử dụng tư vấn bộ điều khiển).