Stream的错误处理

Stream的错误处理

Node.js中的Stream是处理流式数据的核心抽象，无论是文件读写、网络通信还是数据转换，Stream都扮演着重要角色。由于Stream处理的数据通常是分块的、异步的，错误处理变得尤为重要且复杂。

Stream错误事件的基本机制

所有Stream都是EventEmitter的实例，错误通过error事件传播。未处理的error事件会导致进程崩溃，这是Node.js的默认行为：

const fs = require('fs');
const readStream = fs.createReadStream('./nonexistent-file.txt');

// 未捕获的error事件会抛出异常
readStream.on('data', (chunk) => {
  console.log(chunk);
});

正确的做法是始终监听error事件：

readStream.on('error', (err) => {
  console.error('读取文件出错:', err.message);
});

可读流的错误处理

可读流在以下情况会触发error事件：

• 文件不存在（ENOENT）
• 权限不足（EACCES）
• 读取过程中连接中断

const zlib = require('zlib');
const gzip = zlib.createGzip();
const source = fs.createReadStream('./large-file.log');

source
  .on('error', (err) => {
    console.error('源文件错误:', err);
  })
  .pipe(gzip)
  .on('error', (err) => {
    console.error('压缩错误:', err);
  })
  .pipe(fs.createWriteStream('./output.gz'))
  .on('error', (err) => {
    console.error('写入错误:', err);
  });

可写流的错误处理

可写流的错误处理需要特别注意背压（backpressure）情况：

const writable = fs.createWriteStream('./output.txt');

writable.on('error', (err) => {
  console.error('写入失败:', err);
  // 可能需要清理部分写入的文件
});

writable.write('一些数据', (err) => {
  // 每个write操作的回调也会收到错误
  if (err) console.error('写入回调错误:', err);
});

管道链中的错误传播

使用pipe()时，错误不会自动传播：

// 错误不会自动传递
src.pipe(transform).pipe(dest);

// 需要手动监听每个流
src.on('error', handleError);
transform.on('error', handleError);
dest.on('error', handleError);

Node.js 10+引入了stream.pipeline解决这个问题：

const { pipeline } = require('stream');

pipeline(
  fs.createReadStream('./input.txt'),
  zlib.createGzip(),
  fs.createWriteStream('./output.gz'),
  (err) => {
    if (err) {
      console.error('管道操作失败:', err);
    } else {
      console.log('管道操作成功');
    }
  }
);

Transform流的错误处理

转换流需要同时处理读写错误：

const { Transform } = require('stream');

class MyTransform extends Transform {
  _transform(chunk, encoding, callback) {
    try {
      // 处理逻辑
      const result = processChunk(chunk);
      callback(null, result);
    } catch (err) {
      // 同步错误可以通过callback传递
      callback(err);
    }
  }

  _flush(callback) {
    someAsyncOperation((err, data) => {
      if (err) return callback(err);
      callback(null, data);
    });
  }
}

错误恢复策略

根据场景不同，可能需要实现不同的恢复机制：

1. 重试逻辑：

function createRetryableReadStream(filename, retries = 3) {
  let attempts = 0;
  
  function createStream() {
    const stream = fs.createReadStream(filename);
    stream.on('error', (err) => {
      if (attempts++ < retries) {
        setTimeout(() => createStream(), 100 * attempts);
      } else {
        stream.emit('giveup', err);
      }
    });
    return stream;
  }
  
  return createStream();
}

2. 备用数据源：

const primary = fs.createReadStream('./primary-data.json');
const fallback = fs.createReadStream('./fallback-data.json');

primary.on('error', (err) => {
  console.error('主数据源失败，使用备用:', err);
  primary.unpipe();
  fallback.pipe(consumer);
});

高级错误处理模式

对于复杂场景，可以考虑以下模式：

1. 错误聚合：

class ErrorAggregator extends Writable {
  constructor(options) {
    super(options);
    this.errors = [];
  }

  _write(chunk, encoding, callback) {
    someOperation(chunk, (err) => {
      if (err) {
        this.errors.push(err);
        // 继续处理下一个chunk
        return callback();
      }
      callback();
    });
  }

  _final(callback) {
    if (this.errors.length > 0) {
      return callback(new AggregateError(this.errors));
    }
    callback();
  }
}

2. 带状态的错误处理：

function createStatefulStream() {
  let errorState = null;
  
  return new Transform({
    transform(chunk, encoding, callback) {
      if (errorState) {
        // 跳过处理或执行恢复逻辑
        return callback();
      }
      
      try {
        // 正常处理
        callback(null, transform(chunk));
      } catch (err) {
        errorState = err;
        this.emit('error', err);
        callback();
      }
    }
  });
}

与Promise的互操作

现代Node.js支持async/await与Stream的配合：

async function processStream(stream) {
  try {
    for await (const chunk of stream) {
      await processChunk(chunk);
    }
  } catch (err) {
    console.error('异步迭代出错:', err);
    // 可能需要清理资源
  }
}

性能考量

错误处理会影响Stream性能的几个方面：

• 过多的try/catch块会降低吞吐量
• 错误恢复逻辑可能增加内存使用
• 复杂的错误传播路径增加CPU开销

// 低效的错误检查
transform.on('data', (chunk) => {
  try {
    process(chunk);
  } catch (err) {
    transform.emit('error', err);
  }
});

// 更高效的方式
transform.on('error', (err) => {
  // 集中处理
});

transform.on('data', (chunk) => {
  const result = process(chunk); // 假设process可能抛出异常
  // ...
});

调试技巧

调试Stream错误时可以使用以下工具：

1. NODE_DEBUG环境变量：

NODE_DEBUG=stream node app.js

2. 自定义调试流：

class DebugStream extends Transform {
  _transform(chunk, encoding, callback) {
    console.debug('Chunk:', chunk.length, 'bytes');
    this.push(chunk);
    callback();
  }

  _flush(callback) {
    console.debug('Stream ended');
    callback();
  }
}

3. 错误追踪：

stream.on('error', (err) => {
  console.error('错误发生位置:', new Error().stack);
  console.error('原始错误:', err.stack);
});

常见陷阱

1. 错误处理顺序问题：

stream
  .pipe(transform)
  .on('error', (err) => console.error(err)) // 可能太晚注册
  .pipe(output);

// 更好的方式
stream.on('error', (err) => console.error('输入错误:', err));
transform.on('error', (err) => console.error('转换错误:', err));
output.on('error', (err) => console.error('输出错误:', err));

2. 未清理资源：

stream.on('error', (err) => {
  console.error(err);
  // 忘记关闭文件描述符或其他资源
});

// 正确的做法
let fd;
fs.open('file.txt', 'r', (err, fileDescriptor) => {
  if (err) throw err;
  fd = fileDescriptor;
  const stream = fs.createReadStream(null, { fd });
  
  stream.on('error', (err) => {
    fs.close(fd, () => {});
    console.error(err);
  });
});

测试策略

针对Stream错误处理的测试需要特别考虑：

const { PassThrough } = require('stream');

test('应该处理转换错误', (done) => {
  const input = new PassThrough();
  const transform = new MyTransformStream();
  
  transform.on('error', (err) => {
    expect(err.message).toBe('预期的错误');
    done();
  });

  input.pipe(transform);
  input.write('触发错误的数据');
});

模拟错误场景：

function createErroringStream() {
  const stream = new Readable({
    read() {
      this.emit('error', new Error('模拟错误'));
    }
  });
  return stream;
}

与第三方库集成

处理第三方库产生的Stream时：

1. 库可能使用自定义错误：

const libStream = require('some-lib').createStream();

libStream.on('error', (err) => {
  if (err instanceof SomeLib.CustomError) {
    // 特殊处理
  } else {
    // 通用处理
  }
});

2. 错误代码检查：

mysqlConnection.query().stream()
  .on('error', (err) => {
    if (err.code === 'ER_QUERY_INTERRUPTED') {
      // 处理查询中断
    }
  });

浏览器环境差异

浏览器中的Stream API（如Fetch的body）错误处理有所不同：

fetch('/api/stream')
  .then(response => {
    const reader = response.body.getReader();
    
    function read() {
      return reader.read().then(({ done, value }) => {
        if (done) return;
        // 处理数据
        return read();
      });
    }
    
    return read();
  })
  .catch(err => {
    console.error('流读取失败:', err);
  });