在 Micronaut 控制器中流式传输大响应而不会耗尽内存

Question

我们将 Micronaut 与 Mongo 结合使用，通过一些控制器公开数据。由于响应实体的大小正在增长，我们的应用程序有时会出现内存不足。因此，我们正在研究切换到异步 mongo 驱动程序并使用响应式响应将数据流式传输到客户端。很遗憾，我们无法更改 API 响应结构或内容类型（全部为 application/json）

我们的一个 API 返回的实体结构如下：

[
  { "field": "value" },
  { "field": "value" },
  ...
  { "field": "value" }
]

我们使用这个控制器来工作，dataStore 返回一个Publisher<Example>：

    @Get("all")
    Flowable<Example> getAllExamples() {
        return Flowable.fromPublisher(dataStore.find()).map(SomeMapper::toPublic);
    }

这很好用，庞大的示例列表在将其流式传输到客户端之前不必完全加载到内存中。

其他 API 返回（更明智的）结构：

{
  "list": [
    { "field": "value" },
    { "field": "value" },
    ...
    { "field": "value" }
  ],
  "meta": {
    ...
  }
}

我们可以为这样的实体应用类似的发布者/可流动模式，还是在将此类响应发送出去之前将数据加载到内存中？

我们尝试了以下签名：

    @Get("all/dev")
    Single<ExamplesWrapper> getAllDev() {
        Publisher<Example> dev = dataStore.find();
        return Flowable.fromPublisher(dev)
                .map(mapper::map)
                .collect((Callable<ArrayList<Example>>) ArrayList::new, ArrayList::add)
                .map(ExampleWrapper::new);
    }

包装器将添加一些元数据的地方。但这再次将其全部加载到内存中，然后再发送出去，导致应用程序崩溃。

将 Flowable 添加到响应包装器中：

public class ExamplesWrapper {

    private final Flowable<Example> examples;

    @ConstructorProperties({"examples"})
    public ExamplesWrapper(Flowable<Example> examples) {
        this.examples = examples;
    }

    public Flowable<Example> getExamples() {
        return examples;
    }
}

也因一些不错的 Jackson 映射异常而失败。

元数据不依赖于实际的示例数据（它添加了一些静态公司信息）。我们能否以某种方式实现这样的端点，而不必将所有数据加载到内存中？

Answer 1

来自documentation：

6.20 编写响应数据

响应式写入响应数据

Micronaut 的 HTTP 服务器支持通过以下方式写入响应数据块 返回一个发布者，该发布者发出可以编码到 HTTP 响应。

下表总结了示例返回类型签名和 服务器表现出处理它们的行为：返回类型描述

• Flowable：一个Flowable，将每个内容块作为一个byte[]发出而不阻塞
• Flux：将每个块作为 Netty ByteBuf 发出的 Reactor Flux
• Publisher：将每个内容块作为字符串发出的 Publisher
• Flowable 发出 POJO 时，每个发出的对象默认编码为 JSON，不阻塞

当返回响应式类型时，服务器使用 Transfer-Encoding 的 分块并继续写入数据，直到 Publisher onComplete 方法 被调用。

我理解这一点，因此如果您希望 Micronaut 机制流式传输您的内容，则需要具有 Flowable<item> 或 Flux<item> 或 Publisher<item> 之类的签名，其中 item 是您的响应的一部分，而不是完整的 item .然后，Micronaut 将响应来自 Flowable 或等效的块。

在这种情况下，我想到的一件事是您可以自己拆分成合适的块。这样流式传输大型响应而不将它们缓冲到内存中应该可以工作。

所以是这样的：

@Get("all")
public Flowable<String> getAllExamples() {
    ObjectMapper objectMapper = new ObjectMapper();
    Publisher<Example> dev = dataStore.find();
    return Flowable.fromPublisher(dev)
            .map(mapper::map)
            .concatMap(item -> Flowable.just(objectMapper.writeValueAsString(item), ","))
            .startWith("{\"list\": [")
            .concatWith(Flowable.just("],\"meta\":\"whatever\"}"));
}

这很老套，但似乎适用于这种情况。

---

一些无效的方法：

我确实测试了在自定义 Jackson 映射器中直接写入 JsonGenerator，按照jackson streaming api 中的说明刷新对象，但 micronaut RoutingInboundHandler 似乎没有将响应刷新回最终用户而是缓冲它，导致记不清。方法适用于 Spring Boot，因此它可能是 Micronaut 中缺少的功能。

在使用 Micronaut Writeable（阻塞）响应并尝试在写入数据时刷新数据时，我也发生了相同的缓冲。 I opened an issue about that to micronaut core.