面试题：Node.js事件发射器.on('error', callback) 与异步流处理

Question

在处理Node.js的可读流和可写流时，结合事件发射器的.on('error', callback)，如何优雅地处理流操作过程中可能出现的错误，保证数据的完整性和应用的稳定性？请详细阐述实现思路并给出关键代码示例。

Answer 1

实现思路

1. 可读流错误处理：监听可读流的 'error' 事件，当出现错误时，暂停可读流防止进一步读取，关闭相关资源（如文件描述符等），并向可写流发送错误信息（如果有后续处理流程），同时通知应用程序错误发生。
2. 可写流错误处理：监听可写流的 'error' 事件，出现错误时，暂停可读流（如果关联），防止继续写入导致更多错误，关闭相关资源，向应用程序报告错误。
3. 数据完整性保证：在错误处理过程中，记录已处理的数据位置，以便后续恢复或补偿操作（如果可能）。确保未处理完的数据不会丢失，可以考虑缓存或重试机制。
4. 应用稳定性：通过全面的错误处理，避免因流操作错误导致应用程序崩溃。可以使用全局错误处理机制（如 process.on('uncaughtException', callback)）捕获未处理的错误，进行日志记录、报警等操作，维持应用的运行。

关键代码示例

const fs = require('fs');
const readableStream = fs.createReadStream('input.txt');
const writableStream = fs.createWriteStream('output.txt');

// 可读流错误处理
readableStream.on('error', (err) => {
  console.error('可读流发生错误:', err.message);
  readableStream.pause();
  // 这里可以添加关闭文件描述符等操作
  // 如果关联可写流，通知可写流错误
  if (writableStream.writable) {
    writableStream.end();
  }
});

// 可写流错误处理
writableStream.on('error', (err) => {
  console.error('可写流发生错误:', err.message);
  // 如果关联可读流，暂停可读流
  if (readableStream.readable) {
    readableStream.pause();
  }
  // 这里可以添加关闭文件描述符等操作
});

// 管道操作，自动处理背压
readableStream.pipe(writableStream);

// 全局未捕获异常处理
process.on('uncaughtException', (err) => {
  console.error('全局未捕获异常:', err.message);
  // 可以进行日志记录、报警等操作
});