java - 使用 Spring WebClient 响应式读取行

Question

TLDR:如何使用 Spring WebClient(响应式(Reactive))逐行处理 GET 响应？

详细信息:

• 远程服务器返回大小最大为 20 Gb 的响应
• 我的服务单独解析行(每行都编码为 UTF8)并流式传输结果(跳过 99% 的行)
• 我不想将整个响应加载到内存中，例如我想逐行解析服务器更新。

不幸的是我没有找到任何转换的解决方案Flux<ByteBuffer>至Flux<String> (通过在行尾分割)。

问题:是否有嵌入式转换器/解码器可以执行此操作？

可能的解决方案:

• 创建临时缓冲区(最初为空)
• 对于每个输入缓冲区:
  • 将临时缓冲区添加到新的、重新创建的临时缓冲区中。
  • 尝试从此缓冲区读取单行(例如读取直到行尾):
  • 如果还有字节 - 返回此字符串并重复行读取
  • 如果缓冲区已完成(例如，那里没有行分隔符):只需将这些字节复制到临时缓冲区即可。
• 在最后一个缓冲区之后:读取临时缓冲区直到结束。

另外:您不能只将输入缓冲区转换为字符串，因为某些 utf8 字符可以从缓冲区 N 开始并在缓冲区 N+1 处继续。

Answer 1

下面的代码可以工作，但是这是完全同步的代码(可能只具有预取能力)。它使用 Apache Http Components .

HttpClientBuilder.create().build().use { client ->
    val responseHandler = ResponseHandler { response ->
        response.entity.content.use { content ->
            content.bufferedReader().use { buffered ->
                // create class, which can process each line. 
                val processor = StreamedLinesProcessor<TResult>()

                do {
                    val nextLine = buffered.readLine()
                    val needContinue = processor.processNextLine(nextLine)
                } while (needContinue)
                processor.getResult()
            }
        }
    }

    client.execute(HttpGet(url.toString()), responseHandler)
}