如何有效地关闭 MongoDB changeStream？

Posted 2023-04-18

【中文标题】如何有效地关闭 MongoDB changeStream？

【英文标题】：How to effectivelly close a MongoDB changeStream?

【发布时间】：2019-05-12 18:31:28

【问题描述】：

我一直在使用带有 Node.js 和 socket.io 的 MongoDB changeStreams；

我正在使用一种 pub-sub 架构，对于每个订阅，他应该用新订阅替换以前的 changeStream。

逻辑是这样的：

客户端请求新订阅（它与一个集合有关 变化流） 如果客户端有changeStream，服务端会关闭这个changeStream，用更新的changeStream代替。

订阅的伪代码

// If it has a changeStream
if (socket.dbWatcher["users"]) 
  // close changeStream
  socket.dbWatcher["users"].removeAllListeners("change");
  socket.dbWatcher["users"].close();


// Replace with new changeStream and pipeline
socket.dbWatcher["users"] = db.collection("users").watch(pipeline, resumeAfter)
.on("change", (change) => 
  socket.emit("users", change)
)

事情变得越来越慢，所以我检查了连接数量，发现它远远超过了收集数量（16 个收集对 29 个连接）。

一旦套接字断开连接，我将关闭所有 changeStreams 但连接仍然存在（数据库）。我增加了poolSize，但是有29个连接是不正常的。

这是我用来检查连接的 mongodb 命令：db.serverStatus().connections

【问题讨论】：

MongoDB 部署的拓扑是什么？请记住db.serverStatus().connections 的值包括所有传入连接，包括任何mongo shell 连接和/或来自其他服务器的连接，即副本集或mongos 实例。

我使用默认配置的副本集。只需执行命令mongod --maxConns 64000 --replSet rs0

【参考方案1】：

您可以在此处查看关闭变更流的文档：https://docs.mongodb.com/manual/changeStreams/#resume-a-change-stream

正如您在node driver API 中看到的那样，更改流上有一个close() 方法。

但是，我不太确定快速关闭/重新打开与 MongoDB 更改流的连接是否有效。最好只关注整个用户集合。

所设计的 changestream 功能并非真正适用于传统的 pubsub 模式。有关详细信息，请参阅此问题：Severe performance drop with MongoDB Change Streams