杰克逊反序列化阵列通量

Posted 2023-02-15

【中文标题】杰克逊反序列化阵列通量

【英文标题】：Jackson deserialize Flux of arrays

【发布时间】：2021-12-28 19:01:36

【问题描述】：

我正在尝试反序列化 List 的通量。在这种情况下，CustomObjects的分组是必要的：每个List都是一个序列，我后面的处理需要考虑到单独的序列。

我看过this question，它处理了一组对象。事实上，我已经能够使用他们的 Mono 解决方案方法来检索 Mono>。这可行，但似乎效率很低。我想开始处理每个列表，而不是要求 Mono 完成对整个批量集的检索。我们的完整集合可能在 20K 或更多列表的范围内，其中一些我们将丢弃，因此需要更早的处理。

当我尝试使用 Flux 而不是 Mono 时会发生什么：我已经尝试过同时使用 WebClient 和 Feign - 两者的效果相同。单步执行 Jackson 代码，它在处理 Flux 世界时从不拉回 TokenBuffers。这似乎与 Codecs and Jackson JSON 的 Spring 文档中的内容一致 - “当解码到多值发布者（例如 Flux）时，一旦为完全形成的对象接收到足够的字节，每个 TokenBuffer 就会传递给 ObjectMapper”

想法？更好的工作示例？

【参考方案1】：

在application/x-ndjson等流式mimetype的情况下，WebFlux将序列化为JSON并在网络中刷新Flux的每个项目。

@GetMapping(path = "/customObjects", produces = MediaType.APPLICATION_NDJSON_VALUE)
public Flux<Response> streamCustomObjects() 
return Flux.interval(Duration.ofSeconds(1))
                .map(e -> new Response(List.of(new CustomObject("George", 20), new CustomObject("Lefteris", 28))));

以下 WebClient 调用会生成 Flux 或 Responses：

Flux<Response> responses = webClient.get()
      .uri("/customObjects")
      .retrieve()
      .bodyToFlux(Response.class);

responses.subscribe(System.out::println);

样本输出：

(after 1 sec)
    Response(customObjects=[CustomObject(name=George, age=20), CustomObject(name=Lefteris, age=28)])
(after 2 sec)
    Response(customObjects=[CustomObject(name=George, age=20), CustomObject(name=Lefteris, age=28)])
(after 3 sec)
    Response(customObjects=[CustomObject(name=George, age=20), CustomObject(name=Lefteris, age=28)])

【参考方案2】：

确实尝试了已发布的 x-ndjson 建议解决方案，但最终在下游遇到了另一个问题：Flux + Jackson 将不会保留列表的边界。例如，我真的需要知道 [e1, e2, e3] 属于一起。相反，我得到了包含单个元素的 onNext 列表。

我的最终解决方案很简单：我们创建了一个包含单个字段的对象类型，然后包含自定义对象的列表。例如， 序列：[ e1, e2, e3] 。然后让我通过 Flux 接收我的 ListWrapperObject，同时保持每个序列的边界。由于我们不拥有 API，我们需要与我们调用的团队进行谈判，但幸运的是他们能够进行调整。我将这个解决方案提供给下一位寻求这条道路的读者。我还承认，如果您无法修改 API，可能还没有办法通过 Jackson 解决这个问题。