Node.js Streams：如何等待所有数据处理完成后关闭数据库连接？

你遇到的核心问题是：MongoDB的流close/end事件触发时，异步的插入操作还在后台执行，导致计数没达到总数，无法触发连接关闭逻辑。解决思路是同时跟踪流是否读取完毕、所有插入操作是否完成，只有两个条件都满足时再关闭连接。

下面提供两种实用的解决方案：

方案一：计数器+状态跟踪（兼容旧Node.js版本） #

通过标记流状态和计数，在两个关键时机（插入完成、流结束）检查是否满足关闭条件：

var count = 0;
var streamEnded = false; // 标记流是否已读完所有数据
mongo_db.collection(config.collection, function (err, coll) {
  if (err) {
    console.error('获取集合失败:', err);
    return;
  }
  coll.find(config.mongo_query).count(function (e, coll_docs_count) {
    if (e) {
      console.error('统计文档数量失败:', e);
      return;
    }
    var stream = coll.find(config.mongo_query).stream();
    
    // 监听流的end事件：所有数据已从MongoDB读取完毕
    stream.on('end', function () {
      streamEnded = true;
      checkIfAllDone();
    });

    // 必须处理流错误，避免程序卡住
    stream.on('error', function (err) {
      console.error('流读取错误:', err);
      mongo_client.close();
      cb_bucket.disconnect();
    });
    
    stream.on('data', function (doc) {
      // 暂停流，避免大量数据积压在内存（高数据量场景推荐）
      stream.pause();
      
      // 这里执行你的文档处理逻辑
      // ... Do some operations on it
      
      new_db.insert(doc, function (insertErr) {
        if (insertErr) {
          console.error('插入新库失败:', insertErr);
          // 可根据业务需求选择重试、跳过或终止
        }
        count++;
        checkIfAllDone();
        // 恢复流，继续读取下一个文档
        stream.resume();
      });
    });

    // 检查是否满足关闭条件的核心函数
    function checkIfAllDone() {
      if (streamEnded && count === coll_docs_count) {
        console.log('所有文档处理并插入完成！');
        mongo_client.close();
        cb_bucket.disconnect();
      }
    }
  });
});

方案二：Promise.all管理异步操作（现代Node.js推荐） #

如果你的Node.js版本支持Promise（v8+默认支持），可以用Promise包装所有插入操作，等待流结束后统一等待所有插入完成：

mongo_db.collection(config.collection, function (err, coll) {
  if (err) {
    console.error('获取集合失败:', err);
    return;
  }
  coll.find(config.mongo_query).count(function (e, coll_docs_count) {
    if (e) {
      console.error('统计文档数量失败:', e);
      return;
    }
    var stream = coll.find(config.mongo_query).stream();
    const insertPromises = []; // 保存所有插入操作的Promise
    
    stream.on('end', function () {
      // 等待所有插入Promise完成
      Promise.all(insertPromises)
        .then(() => {
          console.log('所有文档插入成功！');
          mongo_client.close();
          cb_bucket.disconnect();
        })
        .catch((err) => {
          console.error('部分插入操作失败:', err);
          // 即使有失败，也可以选择关闭连接（根据业务调整）
          mongo_client.close();
          cb_bucket.disconnect();
        });
    });

    stream.on('error', function (err) {
      console.error('流读取错误:', err);
      mongo_client.close();
      cb_bucket.disconnect();
    });
    
    stream.on('data', function (doc) {
      stream.pause();
      
      // 执行文档处理逻辑
      // ... Do some operations on it
      
      // 将插入操作包装为Promise
      const insertPromise = new Promise((resolve, reject) => {
        new_db.insert(doc, (insertErr, result) => {
          if (insertErr) reject(insertErr);
          else resolve(result);
          stream.resume();
        });
      });
      insertPromises.push(insertPromise);
    });
  });
});

关键注意事项 #

1. 错误处理：务必监听流的error事件，否则流出错时程序会卡住，无法正常关闭连接。
2. 流的暂停/恢复：高数据量场景下，暂停流可以避免内存中堆积过多未处理的文档，防止内存溢出。
3. 插入失败的处理：如果存在插入失败的情况，需要根据业务需求调整计数逻辑（比如跳过失败文档、重试失败操作），避免永远无法触发连接关闭。

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