如何将java.util.stream.Stream写入StreamingResponseBody输出流

Question

我正在构建一个REST，其中Oracle数据库中的大量数据可以通过流到客户端应用程序(如文件下载或直接流)以块形式发送。

我正在从JpaRepository获得流，如下所示-

@Query("select u from UsersEntity u")
Stream<UsersEntity> findAllByCustomQueryAndStream();

但是现在，将此流写入StreamingResponseBody输出流的挑战来了。

我试过很多方法，但都没有成功-

First Approach

Stream<UsersEntity> usersResultStream = usersRepository.findAllByCustomQueryAndStream();

        StreamingResponseBody stream = outputStream -> {
            Iterator<UsersEntity> iterator = usersResultStream.iterator();

            try (ObjectOutputStream oos = new ObjectOutputStream(outputStream)) {

                while (iterator.hasNext()) {
                    oos.write(iterator.next().toString().getBytes());
                }
            }
        };

得到了错误-

java.sql.SQLException: Closed Resultset: next
    at oracle.jdbc.driver.InsensitiveScrollableResultSet.next(InsensitiveScrollableResultSet.java:565) ~[ojdbc7-12.1.0.2.jar:12.1.0.2.0]

第二次逼近-

StreamingResponseBody stream = new StreamingResponseBody() {

            @Transactional(readOnly = true)
            @Override
            public void writeTo(OutputStream outputStream) throws IOException {

                Stream<UsersEntity> usersResultStream = usersRepository.findAllByCustomQueryAndStream();

                try (ObjectOutputStream oos = new ObjectOutputStream(outputStream)) {

                    usersResultStream.forEach(user->{
                        try {
                            oos.write(user.toString().getBytes());
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    });
                }
            }
        }; 

得到了错误-

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.

我已经将练习代码上传到下面的链接- 样例POC链路。

我没有任何与流媒体相关的工作经验，所以请帮助我这个。

如果我走错了方向，而不是建议在Spring框架中使用其他方法来做到这一点。如果有的话，请分享任何参考链接。

Answer 1

最后，我使用服务层解决了这个问题。最初，我在Controller中编写了创建问题的完整逻辑。

控制器类-

@RestController
@RequestMapping("/api")
public class UsersController {
    @Autowired
    private UserService service;

    @GetMapping(value = "/userstream")
    public ResponseEntity<StreamingResponseBody> fetchUsersStream() {

        StreamingResponseBody stream = this::writeTo;

        return new ResponseEntity<>(stream, HttpStatus.OK);
    }

    private void writeTo(OutputStream outputStream) {
        service.writeToOutputStream(outputStream);
    }
}

服务类-

@Service
public class UserService {

    @Autowired
    private UsersRepository usersRepository;

    @Transactional(readOnly = true)
    public void writeToOutputStream(final OutputStream outputStream) {
        try (Stream<UsersEntity> usersResultStream = usersRepository.findAllByCustomQueryAndStream()) {
            try (ObjectOutputStream oos = new ObjectOutputStream(outputStream)) {

                usersResultStream.forEach(emp -> {
                    try {
                        oos.write(emp.toString().getBytes());
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                });
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

完整的代码可在github - https://github.com/bagesh2050/HttpResponseStreamingDemo上获得

不过，我还是愿意听取有关Http流媒体的建议。如果你有更好的想法，请提供。

Answer 2

没有一个示例显示StreamingResponseBody的使用“如此复杂”，我担心这是“不可能的”(至少我无法使用StreamingResponseBody和Stream来管理/修复它)

...but，可能的是：

1. 在findAll()中使用StreamingResponseBody (正常的非流列表-回购方法)。 (但我理解异步执行web请求的“必要性”.和db请求“流”)
2. 使用Callable (异步web请求)和@Async CompletableFuture<..> (异步db请求)： @RestController @RequestMapping("/api")公共类UsersController { @Autowired私有UsersRepository usersRepository；@ return (value=“/异步/用户”)公共Callable fetchUsersAsync() {可调用可调用= () -> {返回usersRepository.readAllBy().get()；}；返回可调用；} ..and是一个仓库，类似： @Repository公共接口UsersRepository扩展JpaRepository {@异步CompletableFuture readAllBy()；} (见弹簧样)。不要忘记在您的@EnableAsync @SpringBootApplication公共类应用程序上使用{.}

对不起，这甚至不是一个答案，但我的发现-太长，不能发表评论。

异步web请求可以通过各种方式实现。(请参阅https://spring.io/blog/2012/05/10/spring-mvc-3-2-preview-making-a-controller-method-asynchronous/、https://niels.nu/blog/2016/spring-async-rest.html，甚至没有提到“反应性”api)