是否需要关闭MongoDB Change Stream？如何长期维持数据库变更监听运行？

问题1解答 #

要长期保持监听状态，直接删除closeChangeStream函数的调用即可。这个函数的作用就是定时关闭变更流，仅适合短时间测试监听的场景，你需要7*24小时运行的监听场景完全不需要用到它。

问题2解答 #

你遇到的Topology is closed报错本质是代码逻辑顺序问题：你的myWatcher是async函数，但内部仅注册了change事件监听就直接返回了，没有任何阻塞逻辑，所以await myWatcher()会立刻执行完成，触发finally块的client.close()把数据库连接主动关闭，变更流自然就报错了。
如果要长期保持监听：

• 可以直接去掉finally块里的await client.close()调用；
• 也可以在myWatcher函数末尾加一个永远不resolve的Promise，让await myWatcher一直处于等待状态，不会走到finally的关闭逻辑，示例代码如下：

async function myWatcher(client, pipeline = []) {
  const watching = client.db("myDatabase").collection("myCollection").watch(pipeline);

  watching.on("change", (next) => {
    console.log(JSON.stringify(next));
    console.log(`Doing my things...`);
  });

  // 新增：监听变更流错误，避免异常直接导致进程崩溃
  watching.on("error", (err) => {
    console.error('Change stream error:', err);
    // 可以在这里加重连逻辑，断开后重新调用myWatcher初始化监听
  });

  // 新增：永久阻塞，让async函数不返回
  await new Promise(() => {});
}

额外生产环境建议 #

1. 初始化MongoClient的时候可以加上自动重连相关配置，避免网络波动导致连接断开后不会自动恢复：

const client = new MongoClient("mongodb://localhost:27018", {
  retryWrites: true,
  retryReads: true,
  maxPoolSize: 10
});

2. 可以监听系统信号，在进程退出前优雅关闭连接，避免资源泄露：

process.on('SIGINT', async () => {
  console.log('Received exit signal, closing connection...');
  await watching.close();
  await client.close();
  process.exit(0);
});

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