使用 Fetch Streams API 异步消费分块数据而不使用递归

Question

我正在使用 JavaScript fetch streams API 来异步使用分块的 JSON，就像在 this answer 中一样。

我的应用程序可能在一小时内每秒接收多达 25 个小型 JSON 对象（视频中的每一帧一个）。

当传入的块很大（每个块 1000 多个 JSON 对象）时，我的代码运行良好 - 速度快，内存使用量最少 - 它可以轻松可靠地接收 1,000,000 个 JSON 对象。

当传入的块较小（每个块 5 个 JSON 对象）时，我的代码运行不佳 - 速度慢，内存消耗大。浏览器在大约 50,000 个 JSON 对象时死掉。

在开发者工具中进行大量调试后，问题似乎在于代码的递归性质。

我试图删除递归，但它似乎是必需的，因为 API 依赖于我的代码返回对链的承诺？！

如何移除这个递归，或者我应该使用fetch以外的东西？

---

递归代码（有效）

String.prototype.replaceAll = function(search, replacement) {
    var target = this;
    return target.replace(new RegExp(search, 'g'), replacement);
};

results = []

fetch('http://localhost:9999/').then(response => {
    const reader = response.body.getReader();
    td = new TextDecoder("utf-8");
    buffer = "";

    reader.read().then(function processText({ done, value }) {
        if (done) {
          console.log("Stream done.");
          return;
        }

        try {
            decoded = td.decode(value);
            buffer += decoded;
            if (decoded.length != 65536){
                toParse = "["+buffer.trim().replaceAll("\n",",")+"]";
                result = JSON.parse(toParse);
                results.push(...result);
                console.log("Received " + results.length.toString() + " objects")
                buffer = "";
            }
        }
        catch(e){
            // Doesn't need to be reported, because partial JSON result will be parsed next time around (from buffer).
            //console.log("EXCEPTION:"+e);
        }

        return reader.read().then(processText);
    })
});

没有递归的代码（不起作用）

String.prototype.replaceAll = function(search, replacement) {
    var target = this;
    return target.replace(new RegExp(search, 'g'), replacement);
};

results = []
finished = false

fetch('http://localhost:9999/').then(response => {
    const reader = response.body.getReader();
    td = new TextDecoder("utf-8");
    buffer = "";
    lastResultSize = -1

    while (!finished)
        if (lastResultSize < results.length)
        {
            lastResultSize = results.length;
            reader.read().then(function processText({ done, value }) {

                if (done) {
                  console.log("Stream done.");
                  finished = true;
                  return;
                }
                else
                    try {
                        decoded = td.decode(value);
                        //console.log("Received chunk " + decoded.length.toString() + " in length");
                        buffer += decoded;
                        if (decoded.length != 65536){
                            toParse = "["+buffer.trim().replaceAll("\n",",")+"]";
                            result = JSON.parse(toParse);
                            results.push(...result);
                            console.log("Received " + results.length.toString() + " objects")
                            buffer = "";
                            //console.log("Parsed chunk " + toParse.length.toString() + " in length");
                        }
                    }
                    catch(e) {
                        // Doesn't need to be reported, because partial JSON result will be parsed next time around (from buffer).
                        //console.log("EXCEPTION:"+e);
                    }
            })
        }
});

---

为了完整起见，这里是我在测试服务器上使用的 python 代码。注意包含 sleep 的行，它改变了分块行为：

import io
import urllib
import inspect
from http.server import HTTPServer,BaseHTTPRequestHandler
from time import sleep


class TestServer(BaseHTTPRequestHandler):

    def do_GET(self):
        args = urllib.parse.parse_qs(self.path[2:])
        args = {i:args[i][0] for i in args}
        response = ''

        self.send_response(200)
        self.send_header('Content-type', 'text/html')
        self.send_header('Access-Control-Allow-Origin', '*')
        self.send_header('Transfer-Encoding', 'chunked')
        self.end_headers()

        for i in range (1000000):
            self.wfile.write(bytes(f'{{"x":{i}, "text":"fred!"}}\n','utf-8'))
            sleep(0.001)  # Comment this out for bigger chunks sent to the client!

def main(server_port:"Port to serve on."=9999,server_address:"Local server name."=''):
    httpd = HTTPServer((server_address, server_port), TestServer)
    print(f'Serving on http://{httpd.server_name}:{httpd.server_port} ...')
    httpd.serve_forever()


if __name__ == '__main__':
    main()

Answer 1

您缺少的部分是传递给.then() 的函数始终是异步调用的，即使用空堆栈。所以这里没有实际的递归。这也是为什么您的“无递归”版本不起作用的原因。

解决这个问题的简单方法是使用异步函数和 await 语句。如果你像这样调用 read()：

const {value, done} = await reader.read();

...然后您可以循环调用它，它会按照您的预期工作。

我不知道您的内存泄漏具体在哪里，但是您对全局变量的使用似乎是个问题。我建议您始终将'use strict'; 放在代码的顶部，以便编译器为您捕获这些问题。然后在声明变量时使用let 或const。

我建议您使用TextDecoderStream 来避免字符在多个块之间拆分时出现问题。当 JSON 对象在多个块之间拆分时，您也会遇到问题。

请参阅 Append child writable stream demo 了解如何安全地执行此操作（但请注意，您需要 TextDecoderStream，该演示具有“TextDecoder”）。

还要注意该演示中使用了 WritableStream。 Firefox 还不支持它 AFAIK，但 WritableStream 提供了更简单的语法来使用块，而无需显式循环或递归。你可以找到 web 流 polyfill here。