AWS Lambda向React客户端流式传输数据：配置与读取问题

问题 #

我需要在某一事件触发时向客户端发送数据包，尝试使用AWS的streamifyResponse包实现，具体操作如下：

后端尝试的两种实现方式 #

1. 直接写入响应流：

responseStream.setContentType('application/json')

// 事件触发时，获取数据（存储在json变量中）并执行
responseStream.write(json)

// 最后用streamifyResponse包装整个函数
export const function = streamifyResponse(helper)

2. 管道流方式：

// 定义管道：
const pipeline = require("util").promisify(require("stream").pipeline)

// 事件触发时：
const requestStream = Readable.from(Buffer.from(json))
await pipeline(requestStream, zlib.createGzip(), responseStream)

但管道流方式会触发控制台警告：

MaxListenersExceededWarning: Possible EventEmitter memory leak detected. 11 unpipe listeners added to [ResponseStream]. Use emitter.setMaxListeners() to increase limit

前端尝试的读取方式 #

• 使用Amplify的API.post方法和常规fetch方法，只有Amplify API能触发后端流，但无法持续获取数据，而是最后一次性接收。
• 用fetch时尝试通过读取器提取流式数据：

const reader = response.body!.getReader()
console.log("reader", reader)
while (true) {
  const { done, value } = await reader.read();
  if (done) {
    // 处理最后一块数据后退出读取器
      console.log("done streaming")
      return
  }
  console.log("streamed value", value)
}

• 最后在API配置中添加：

url: {
    streaming: true
}

我的问题是：后端的流配置是否正确？如果正确，前端在数据发布时的最佳读取方式是什么？

解答 #

后端流配置问题分析 #

1. 直接写入响应流的方式
   基础用法正确，但需注意几个关键点：

   • 确保helper函数正确接收responseStream作为参数，streamifyResponse会自动处理响应的分块编码（Transfer-Encoding: chunked），无需手动设置。
   • 每次responseStream.write(json)时，建议采用**换行分隔JSON（NDJSON）**格式，即每个JSON字符串后加换行符\n，否则前端无法区分多个独立数据包。示例：

     responseStream.write(JSON.stringify(data) + '\n')
     

   • 不要提前调用responseStream.end()，除非确认所有数据发送完毕，否则会直接终止流连接。

2. 管道流方式的警告问题
   出现MaxListenersExceededWarning是因为每次事件触发都创建新的pipeline，每个pipeline会给responseStream添加多个事件监听器，当触发次数超过默认上限（10个）时就会触发警告。解决方法：

   • 优先使用responseStream.write()直接发送分块数据，无需每次创建新的Readable和pipeline，这种方式更轻量，不会累积监听器。
   • 若必须使用管道，可手动提高responseStream的监听器上限：

     responseStream.setMaxListeners(0) // 0表示无上限，或设置一个足够大的数值
     

前端流式数据读取最佳实践 #

1. Amplify API 优化

   • 你添加的streaming: true配置正确，确保了Amplify以流式方式处理响应。
   • 一次性接收数据的原因通常是后端未分块发送，或Amplify默认缓冲了响应。可在API调用时显式指定响应类型，并使用流式回调处理：

     API.post('yourApiName', '/path', {
       body: {},
       responseType: 'stream',
       onProgress: (progress) => {
         const chunk = progress.data.toString()
         // 按换行分割处理NDJSON格式的数据
         chunk.split('\n').filter(line => line).forEach(line => {
           const data = JSON.parse(line)
           console.log('Received chunk:', data)
         })
       }
     })
     

2. Fetch API 优化

   • 读取器代码逻辑正确，但需补充以下处理：
     • 如果后端启用了gzip压缩，需先对响应解压（现代浏览器支持DecompressionStream）：

       const response = await fetch('/your-endpoint')
       const decompressedStream = response.body.pipeThrough(new DecompressionStream('gzip'))
       const reader = decompressedStream.getReader()
       

     • 将读取到的Uint8Array转换为字符串，并处理NDJSON格式，避免不完整的JSON解析错误：

       const reader = response.body!.getReader()
       let buffer = ''
       while (true) {
         const { done, value } = await reader.read();
         if (done) {
           console.log("done streaming")
           return
         }
         buffer += new TextDecoder().decode(value)
         // 拆分完整行，保留不完整的后续拼接
         const lines = buffer.split('\n')
         buffer = lines.pop() || ''
         lines.forEach(line => {
           if (line) {
             const data = JSON.parse(line)
             console.log('Received chunk:', data)
           }
         })
       }
       

   • 确认后端返回的响应头包含Transfer-Encoding: chunked，streamifyResponse会自动设置该字段，无需手动添加。

内容的提问来源于stack exchange，提问作者kmcclenn