如何在 Quarkus 中通过 REST 从数据库流式传输大数据

Question

我在 Quarkus 中实现了一个 GET 方法，它应该向客户端发送大量数据。使用 JPA/Hibernate 从数据库中读取数据，序列化为 JSON，然后发送到客户端。如果不将整个数据都保存在内存中，如何有效地做到这一点？我尝试了以下三种可能性都没有成功：

1. 使用来自 JPA 的 getResultList 并返回一个 Response 并将列表作为正文。 MessageBodyWriter 将负责将列表序列化为 JSON。但是，这会将所有数据拉入内存，这对于大量记录是不可行的。
2. 使用来自 JPA 的 getResultStream 并返回一个 Response 以流作为主体。 MessageBodyWriter 将负责将流序列化为 JSON。不幸的是，这不起作用，因为在执行 JAX-RS 方法之后和调用 MessageBodyWriter 之前，EntityManager 似乎已关闭。这意味着底层的ResultSet 也已关闭，写入器无法再从流中读取。
3. 使用StreamingOutput 作为Response 正文。出现与 2. 中相同的问题。

所以我的问题是：使用 Quarkus 发送通过 JPA 读取的大数据的技巧是什么？

Answer 1

您的结果必须全部在一个响应中吗？如何让客户端请求下一个结果页面，直到没有下一个 - 典型的 REST API 分页练习？此外，JPA 后端只会从数据库中获取该页面，因此不会有所有内容都位于内存中的时刻。

Answer 2

根据您的要求，您有两种选择：

选项 1： 采取 HATEOAS 方法 (https://restfulapi.net/hateoas/)。通过 REST 标准交换大型数据集的标准模式之一。因此，在这种方法中，服务器将在第一响应中快速响应一组 HATEOAS URI。其中每个 HATEOAS URI 代表一组元素。因此，您需要根据数据大小生成这些 URI，并让客户端代码负责单独调用这些 URI 作为 REST API 以获取实际数据。但同样在此选项中，您也可以考虑 Reactive 样式，以便在内存占用少的情况下获得流处理的更多优势。

选项 2： 正如上面@Serkan 所建议的，将数据库中的结果集作为 REST 响应连续流式传输到客户端。这里需要确保客户端和Service之间的网关进行超时设置。如果没有网关你很好。因此，您可以利用所有层的反应式编程来实现连续流式传输。 “DAO/数据访问层”-->“服务层”--> REST 控制器--> 客户端。 Spring reactor 也符合 JAX-RS。 https://quarkus.io/guides/getting-started-reactive。这是处理大数据处理时最好的架构风格。

Answer 3

这里有一些资源可以帮助您：

• 使用反应式休眠：https://quarkusio.zulipchat.com/#narrow/stream/187030-users/topic/Large.20datasets.20using.20reactive.20SQL.20clients
• 分页与仅转发结果集：https://knes1.github.io/blog/2015/2015-10-19-streaming-mysql-results-using-java8-streams-and-spring-data.html

上一篇是针对SpringBoot的，不过思路也可以用Quarkus来实现。

------------编辑：

好的，我已经制定了一个批量选择的示例。我用 Panache 做到了，但没有它你也可以轻松做到。

我正在返回一个 ScrollableResult，然后在 Rest 资源中使用它通过 SSE（服务器发送事件）将其流式传输到客户端。

------------编辑2：

我已将 setFetchSize 添加到查询中。您应该使用这个数字并将其设置在 1-50 之间。如果 value = 1，则 db 行将被 1 比 1 获取，这最模仿流式传输。并且它会使用最少的内存，但 db & app 之间的 I/O 会更频繁。

在执行此类批量操作时，强烈建议使用 StatelessSession。

@Entity
public class Fruit extends PanacheEntity {

    public String name;
    
    // I've removed the logic from here to the Rest resource, 
     // otherwise you cannot close the session
    
}

@Path("/fruits")
public class FruitResource {

    @GET
    @Produces(SERVER_SENT_EVENTS)
    public void fruitsStream(@Context Sse sse, @Context SseEventSink sink) {
        var sf = Fruit.getEntityManager().getEntityManagerFactory().unwrap(SessionFactory.class);

        try (var session = sf.openStatelessSession();
             var scrollableResults = session.createQuery("select f from Fruit f")
                                    .scroll(ScrollMode.FORWARD_ONLY) 
                                    .setFetchSize(1) {
            while (scrollableResults.next()) {
                sink.send(sse.newEventBuilder().data(scrollableResults.get(0)).mediaType(APPLICATION_JSON_TYPE).build());
            }
            sink.close();
        }
    }
}

然后我这样调用这个 Rest 端点（通过 httpie）：

> http :8080/fruits --stream

data: {"id":9996,"name":"applecfcdd592-1934-4f0e-a6a8-2f88fae5d14c"}

data: {"id":9997,"name":"apple7f5045a8-03bd-4bf5-9809-03b22069d9f3"}

data: {"id":9998,"name":"apple0982b65a-bc74-408f-a6e7-a165ec3250a1"}

data: {"id":9999,"name":"apple2f347c25-d0a1-46b7-bcb6-1f1fd5098402"}

data: {"id":10000,"name":"apple65d456b8-fb04-41da-bf07-73c962930629"}

希望对你有所帮助。