如何使用 Node 和 Fetch API 通过 HTTP 传输数据

抽象的

本文介绍了使用HTTP 流在 Web 应用程序中实现高效数据可视化的实用指南。

我受到从事利用OpenAI API提供的流支持的 AI 项目的经验启发而撰写了这篇文章，我想分享我的发现，希望这会有所帮助。

什么是 HTTP 流？

HTTP 流式传输是一种通过 HTTP 发送数据的方式，使用一种称为分块传输编码的机制。这意味着服务器可以以多个块的形式发送响应，每个块都有各自的大小和内容。客户端可以在收到第一个块后立即开始处理响应，而无需等待整个响应完成。这可以减少服务器和客户端的延迟和内存占用。

兼容性

大多数现代浏览器和 HTTP 客户端都支持 HTTP 流，并且它与纯 HTTP 和 HTTPS 配合良好。

HTTP/1.1 与 HTTP/2 对比

您还可以使用 HTTP/2 进行流式传输，它比 HTTP/1.1 具有一些优势，例如多路复用和标头压缩。但是，HTTP/2 需要 HTTPS 以及服务器端的一些额外配置。

为了简单起见，本文将重点关注 HTTP/1.1。

为什么要使用 HTTP 流？

HTTP 流对于需要可视化大量数据（例如图表、图形、地图、表格或复杂的 AI 响应等耗时的响应）的 Web 应用程序非常有用，这主要是为了向用户提供更具吸引力的交互式体验。

概念验证

我选择使用纯 javascript、Node.js和标准Fetch API来实现示例作为概念证明，避免使用任何第三方框架，这样我们就不会被技术细节所困扰，但我们将专注于流架构。

❗重要❗ 异步生成器

因为要通过 HTTP实现分块传输编码，我们需要将整体计算分成可以返回部分（且一致）结果的较小任务，所以我们将探索异步生成器，这是一种非常适合实现此目标的令人难以置信的内置 JavaScript 工具。

异步生成器函数的剖析

异步生成器函数是一种特殊的函数，它返回一个符合异步可迭代和异步迭代器协议的AsyncGenerator 对象。
异步生成器函数允许在迭代过程中产生中间结果，暂停当前执行，并允许等待它的代码使用此类结果。

最终，异步生成器函数结合了异步函数和生成器函数的特性，您可以在函数体中同时使用await和yield关键字，因此您可以使用await以符合人体工程学的方式处理异步任务，同时利用生成器函数的惰性特性。

服务器端实现

假设有一个异步生成器函数，可以分块生成数据。为了演示概念，这里有一个简单的函数，它发送数据块，每个数据块之间都有延迟：

async function* generateData() {
    for (let i = 0; i < 1000; i++) {
        // Simulate delay
        await new Promise(resolve => setTimeout(resolve, 1000));
        // Yield data chunk
        yield `data chunk ${i}\n`;
    }
}

在实际情况下，我们会有一个更复杂的逻辑来生成来自外部数据源的数据块，但为了本例的目的，我们将保持简单。

现在我们可以编写一个简单的服务器，通过 HTTP 将数据流传输到客户端。需要注意的是，要迭代数据块，我们可以使用for await语句，该语句创建一个循环，对异步可迭代对象进行迭代。

import http from 'http';

const server = http.createServer(async (req, res) => {
    res.writeHead(200, {
        'Content-Type': 'text/plain',
        'Transfer-Encoding': 'chunked'
    });
    // Asynchronous iterate over the data chunks
    for await (const chunk of generateData()) {
        res.write(chunk);
        console.log(`Sent: ${chunk}`);
    }
    res.end();
});

const PORT = 3000;
server.listen(PORT, () => 
    console.log(`Server running at http://localhost:${PORT}/`) );

从上面的代码可以看出，Node.js中通过 HTTP 实现分块传输编码非常简单，我们异步迭代数据块并将它们写入 HTTP 响应即可。

客户端实现

在客户端，我们使用fetch API来处理来自服务器的流式响应。在这种情况下，我们可以使用附加一个 Reader到响应主体getReader()，它会锁定流并等待服务器发送的每个数据块。

解码数据块

由于数据块是编码的，我们需要先对其进行解码才能使用它。

额外福利：使用异步生成器函数包裹 Reader

我们可以使用异步生成器函数来包装 Reader，该函数允许以流式方式获取数据，并在数据可用时立即生成每个数据块。

/**
 * Generator function to stream responses from fetch calls.
 * 
 * @param {Function} fetchcall - The fetch call to make. Should return a response with a readable body stream.
 * @returns {AsyncGenerator<string>} An async generator that yields strings from the response stream.
 */
async function* streamingFetch( fetchcall ) {

    const response = await fetchcall();
    // Attach Reader
    const reader = response.body.getReader();
    while (true) {
        // wait for next encoded chunk
        const { done, value } = await reader.read();
         // check if stream is done
        if (done) break;
        // Decodes data chunk and yields it
        yield (new TextDecoder().decode(value));
    }
}

现在我们可以使用streamingFetch生成器函数来流式传输来自服务器的数据块，如下所示。

(async () => {

    for await ( let chunk of streamingFetch( () => fetch('http://localhost:3000/') ) ) {
        console.log( chunk )
    }

})();

完成了！✅ 现在，只要服务器有数据块可用，您就可以看到它们了。

记下来👀：

我们在服务器中有一个异步生成器来生成数据块，在客户端中有一个异步生成器来使用这些数据块。

优点

敏捷的用户体验：您可以在数据可用时立即开始显示数据。
可扩展的 API：在内存中累积结果不会造成内存使用量激增。
使用纯 HTTP 和标准 JavaScript API。无需管理任何连接，也无需使用几年后可能过时的复杂框架。

缺点🤔

与使用常规 API 调用相比，实施起来稍微复杂一些。
错误处理变得更加困难，因为流媒体开始时就会发送 HTTP 状态代码 200（见封面图片）。
- 如果流程中途出现问题我们该怎么办？
- 应用程序必须能够确定流是否尚未完成并采取相应的行动。
需要将格式化假设作为合同的一部分或使用非常规格式。

福利💯：来自 OpenAI Chat API 的流式响应

正如开头所述，我深入研究了HTTP 流，以便能够利用OpenAI 流式 API提供的流式支持。本质上，该 API 的流式版本不返回完整答案，而是返回一个异步可迭代对象🤩。
因此，我们可以使用与之前相同的方法😉通过 HTTP 流式传输来自 OpenAI 服务器的数据块，如下所示：

import http from 'http';
import OpenAI from "openai";

const openai = new OpenAI();

(async () => {

    const stream = await openai.chat.completions.create({
        model: "gpt-4",
        messages: [{ role: "user", content: "Say this is a test" }],
        stream: true,
    });

    const server = http.createServer(async (req, res) => {
        res.writeHead(200, {
            'Content-Type': 'text/plain',
            'Transfer-Encoding': 'chunked'
        });

    // Asynchronous iterate over the data chunks
    for await (const chunk of stream) {
        res.write(chunk.choices[0]?.delta?.content || "");
        console.log(`Sent: ${chunk}`);
    }

    res.end();
});

const PORT = 3000;
server.listen(PORT, () => 
    console.log(`Server running at http://localhost:${PORT}/`) );
})();

结论

在本文中，我们了解了如何在 Web 应用程序中使用 HTTP 流进行高效的数据可视化的实用指南。我们探讨了在 HTTP 上使用分块传输编码的优缺点。我们还深入探讨了异步生成器函数的强大功能及其在实现 HTTP 流中的应用。最后，我们了解了使用Node.js、Fetch API和OpenAI Streaming API通过 HTTP 进行流式传输数据的实际用例。

希望这些知识能对你有所帮助。同时，享受编程的乐趣吧！👋

参考

最初于 2024 年 2 月 10 日发布于https://bsorrentino.github.io 。

文章来源：https://dev.to/bsorrentino/how-to-stream-data-over-http-using-node-and-fetch-api-4ij2