Express与Axios中Transfer-Encoding chunked流式请求的错误处理咨询

服务器端（Express）优化 #

HTTP响应头一旦发送（首次调用res.write后），就无法修改状态码。因此需要分两种场景处理错误：

1. 响应头发送前的错误（如MongoDB Cursor初始化失败） #

此时还未向客户端发送任何响应内容，可直接返回错误状态码和信息。

2. 流式传输过程中的错误（如Cursor读取失败、响应流写入失败） #

响应头已发送，只能通过在响应体中插入约定格式的错误信息通知客户端，再终止流。同时要补充res.write的错误捕获逻辑，避免Promise仅处理成功场景。

修改后的服务器代码：

app.get('/stream', async (req, res) => {
  let cursor;
  try {
    // 用batchSize替代手动limit&skip，MongoDB自动分批拉取数据更高效
    cursor = await db.collection('your-collection').find({}).batchSize(100);

    // 设置流式传输响应头（可选，明确传输模式）
    res.setHeader('Transfer-Encoding', 'chunked');
    res.setHeader('Content-Type', 'application/json');

    while (await cursor.hasNext()) {
      try {
        const data = await cursor.next();
        await new Promise((resolve, reject) => {
          // MongoDB文档需转为JSON字符串再发送，加换行符方便客户端按行解析
          res.write(JSON.stringify(data) + '\n', 'utf8', (err) => {
            if (err) {
              console.log(`Write error: ${err.message}`);
              reject(err);
            } else {
              console.log("batch sent");
              resolve();
            }
          });
        });
      } catch (streamError) {
        // 处理流式传输中的错误
        console.log(`Stream error: ${streamError.message}`);
        // 发送约定格式的错误信息
        res.write(JSON.stringify({
          error: true,
          status: 500,
          message: streamError.message
        }) + '\n');
        // 终止响应并释放Cursor资源
        res.end();
        await cursor.close();
        return;
      }
    }

    // 正常结束流，释放资源
    res.end();
    await cursor.close();
  } catch (initError) {
    // 处理初始化阶段的错误（未发送响应头）
    console.log(`Init error: ${initError.message}`);
    res.status(500).json({
      error: true,
      message: initError.message
    });
    if (cursor) await cursor.close();
  }
});

服务器端关键说明 #

• 用batchSize优化MongoDB数据拉取，避免手动分页的低效问题。
• 每个数据块转JSON并换行，方便客户端按行解析。
• 流式传输出错时，发送含error标记的JSON对象，客户端可识别处理。
• 无论正常结束还是出错，都需关闭Cursor释放MongoDB资源。

客户端（Axios）优化 #

AxiosresponseType: 'stream'模式下，需区分两类错误：

1. HTTP状态码错误（如4xx/5xx，未开始流式传输）：进入catch块处理。
2. 流式传输过程中的错误（服务器发送的错误标记）：需在data事件中解析每个块。

修改后的客户端代码：

const { createWriteStream } = require('fs');
const axios = require('axios');

async function fetchStream() {
  try {
    const res = await axios({
      url: 'http://localhost:3000/stream',
      responseType: 'stream'
    });

    const { data } = res;
    const outputStream = createWriteStream('./output.jsonl'); // 示例输出流

    data.on('data', (chunk) => {
      const chunkStr = chunk.toString('utf8').trim();
      if (!chunkStr) return;

      try {
        const parsed = JSON.parse(chunkStr);
        if (parsed.error) {
          // 识别服务器发送的错误标记
          console.error(`Stream error from server: [${parsed.status}] ${parsed.message}`);
          // 终止流并关闭输出资源
          data.destroy();
          outputStream.close();
        } else {
          // 处理正常数据
          outputStream.write(chunkStr + '\n');
        }
      } catch (parseErr) {
        console.error(`Parse chunk error: ${parseErr.message}`);
      }
    });

    data.on('end', () => {
      console.log('Stream finished successfully');
      outputStream.close();
    });

    data.on('error', (streamErr) => {
      console.error(`Connection error: ${streamErr.message}`);
      outputStream.close();
    });

  } catch (axiosErr) {
    // 处理HTTP状态码错误（如初始化阶段的500）
    console.error(`HTTP error: [${axiosErr.response.status}] ${axiosErr.response.data.message}`);
  }
}

fetchStream();

客户端关键说明 #

• 按行解析响应块，匹配服务器的换行符约定。
• 检查解析后的对象是否含error字段，判断是否为服务器发送的错误信息。
• 捕获Axios的catch块，处理HTTP状态码类错误。
• 遇到错误时调用data.destroy()终止流，避免资源泄漏。

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