如何终止处理百万级MySQL记录的Node.js命令行脚本

好问题！处理百万级数据的流任务时，不管是主动触发终止（比如满足特定业务条件）还是被动响应用户操作（比如Ctrl+C），都需要优雅终止——既要停止数据处理，还要清理数据库连接、释放资源，避免内存泄漏或数据不一致。下面结合你的代码场景，给你几种实用方案：

1. 主动触发终止（满足业务条件时停止） #

如果需要在处理到某条特定记录、或者完成指定任务后终止脚本，可以在transform函数里主动销毁流，并处理后续的清理逻辑：

const streamTransform = stream.Transform({
  objectMode: true,
  transform: function(data, encoding, callback) {
    // 示例：处理到第500000条记录时终止
    if (data.id === 500000) {
      console.log('达到终止条件，开始停止流...');
      // 销毁当前流，会触发流的'close'或'error'事件
      this.destroy(new Error('主动终止流'));
      return callback();
    }

    doWhatINeedToDo(data);
    callback();
  }
});

const queryStream = conn.query(myQueryToGet1MRecords)
  .stream()
  .pipe(streamTransform);

// 监听流的销毁/结束事件，清理数据库连接
queryStream.on('close', () => {
  console.log('流已关闭，正在断开数据库连接...');
  conn.end(err => {
    if (err) console.error('数据库断开失败:', err);
    console.log('脚本终止完成');
    process.exit(0); // 退出进程
  });
});

// 处理流的错误（包括主动销毁的错误）
queryStream.on('error', (err) => {
  console.error('流错误:', err.message);
  conn.end(() => process.exit(1));
});

2. 响应用户手动终止（Ctrl+C） #

用户在CLI里按Ctrl+C时，Node.js会收到SIGINT信号，我们可以监听这个信号，做优雅终止：

// 监听Ctrl+C信号
process.on('SIGINT', () => {
  console.log('\n收到终止信号，正在优雅停止...');
  
  // 暂停MySQL查询流
  queryStream.destroy();
  
  // 断开数据库连接后退出进程
  conn.end(err => {
    if (err) console.error('数据库断开失败:', err);
    console.log('脚本已终止');
    process.exit(0);
  });
});

// 你的原代码部分
const queryStream = conn.query(myQueryToGet1MRecords)
  .stream()
  .pipe(stream.Transform({
    objectMode: true,
    transform: function(data, encoding, callback){
      doWhatINeedToDo(data);
      callback();
    }
  }));

3. 处理异常导致的终止 #

如果在数据处理中发生错误（比如转换TIFF失败、数据库操作出错），也需要终止流并清理资源：

const streamTransform = stream.Transform({
  objectMode: true,
  transform: async function(data, encoding, callback) {
    try {
      await doWhatINeedToDo(data); // 如果doWhatINeedToDo是异步函数
      callback();
    } catch (err) {
      console.error('处理数据失败:', err);
      // 终止流
      this.destroy(err);
      callback(err);
    }
  }
});

const queryStream = conn.query(myQueryToGet1MRecords)
  .stream()
  .pipe(streamTransform);

queryStream.on('error', (err) => {
  console.error('流终止:', err.message);
  conn.end(() => process.exit(1));
});

关键注意点 #

• 不要直接用process.exit()终止，这样会跳过资源清理（比如数据库连接可能不会正常断开）。
• 销毁流后，一定要监听close或error事件，确保数据库连接被正确关闭。
• 如果你的doWhatINeedToDo是异步操作，要确保在异步完成后再调用callback，避免流的背压问题。

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