通过 RestController 从 Spring CrudRepository 流式传输答案

【问题标题】：Streaming from Spring CrudRepository via RestController通过 RestController 从 Spring CrudRepository 流式传输
【发布时间】：2018-03-07 23:05:51
【问题描述】：

我正在尝试通过 RestController 从 CrudRepository 流式传输大型数据集。我正在使用 HSQLDB 进行测试。但是，无论我做什么，我都无法让它发挥作用。首先我尝试了流：

串流尝试

存储库

public interface PersonRepository extends CrudRepository<Person, String> {

    @Query("select p from Person p")
    Stream<Person> streamAll();
}

休息控制器

@RestController
public class PersonRestController {

    @Autowired
    private PersonRepository personRepository;

    @Transactional(readOnly = true, propagation = Propagation.REQUIRED)
    @GetMapping("stream")
    public StreamingResponseBody getPersonsStream() {

        return new StreamingResponseBody() {
            @Transactional(readOnly = true, propagation = Propagation.REQUIRED)
            @Override
            public void writeTo(OutputStream outputStream) throws IOException {
                try (
                        PrintWriter writer = new PrintWriter(outputStream);
                        Stream<Person> stream = personRepository.streamAll()
                ) {
                    stream
                            .map(person -> person.getFirstName() + " " + person.getLastName())
                            .forEach(name -> {
                                writer.println(name);
                                writer.flush();
                            });
                }
            }
        };
    }
}

但它不起作用。无论我做什么，Spring 都会抱怨缺少只读事务。

然后我尝试使用 WebFlux：

WebFlux 尝试

存储库

public interface PersonRepository extends CrudRepository<Person, String> {

    @Query("select p from Person p")
    Flux<Person> fluxAll();

}

休息控制器

@RestController
public class PersonRestController {

    @Autowired
    private PersonRepository personRepository;

    @Transactional(readOnly = true)
    @GetMapping(value = "stream", produces = MediaType.APPLICATION_STREAM_JSON_VALUE)
    public Flux<Person> getPersonsStream() {

        return personRepository.fluxAll();
    }
}

我来了

org.springframework.core.convert.ConverterNotFoundException: No converter found capable of converting from type [java.util.ArrayList<?>] to type [reactor.core.publisher.Flux<?>]

有没有一种简单的方法可以通过 RestController 从存储库中流式传输数据？

【问题讨论】：

不管怎样，第一个错误看起来像是一个线程问题（两个 @Transactional 注释可能在不同的地方查看），这个用例可能应该在文档中明确解决。
我试着不放，只放第一个，只放第二个，两者都放，但都没有。
外层Transactional没用，因为不同的线程正在执行writeTo方法。内部 Transactional 被忽略，因为它注释了一个不是 Spring bean 的对象的方法。将 writeTo 方法的内容移动到另一个 Spring bean 的方法中，使用 Transactional 进行注释，并从 StreamngResponseBody 的 writeTo 方法中调用该 Spring bean 方法。

标签： java spring spring-data-jpa spring-restcontroller spring-webflux

【解决方案1】：

在您的 WebFlux 尝试中，PersonRepository 需要扩展 ReactiveCrudRepository。

【讨论】：

这导致org.springframework.data.mapping.PropertyReferenceException: No property saveAll found for type Person!
那是因为它需要一个Publisher;查看github.com/spring-projects/spring-data-commons/blob/master/src/…

【解决方案2】：

这就是我做的。

存储库

@Query("SELECT c FROM Customer c")
@QueryHints(value = @QueryHint(name = HINT_FETCH_SIZE, value = "0"))
public Stream<Customer> streamAll();

服务

@Transactional(readOnly = true)
public void streamAll(final OutputStream outputStream) {
  try {
    final Stream<Customer> stream = customerRepository.streamAll();
    objectMapper.writeValue(outputStream, stream);
  } catch (Exception e) {
    log.error(e.getMessage(), e);
  }
}

控制器

@GetMapping("/stream")
@ResponseBody
public StreamingResponseBody streamAll() {
  return outputStream -> customerService.streamAll(outputStream);
}

【讨论】：