javascript - 使用 Fetch Streams API 在不使用递归的情况下异步使用分块数据

Question

我正在使用 JavaScript fetch streams API像 this answer 中那样异步使用分块的 JSON .

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

当传入的数据 block 很大(每个数据 block 超过 1000 个 JSON 对象)时，我的代码运行良好 - 速度快，内存使用最少 - 它可以轻松可靠地接收 1,000,000 个 JSON 对象。

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

在 Developer 工具中进行了大量调试后，问题似乎出在代码的递归性质上。

我试图删除递归，但这似乎是必需的，因为 API 依赖于我的代码返回链式 promise ？!

我该如何删除这个递归，或者我应该使用 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以避免字符在多个 block 之间拆分时出现问题。当 JSON 对象在多个 block 之间拆分时，您也会遇到问题。

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

另请注意该演示中对 WritableStream 的使用。 Firefox 尚不支持 AFAIK，但 WritableStream 提供了更简单的语法来使用 block ，而无需显式循环或递归。您可以找到网络流 polyfill here .