问题标签 [changestream]

我正在使用 MongoDB 更改流以便在服务器之间进行间接通信。一个 API 服务器在 DMZ，另一个在 Intranet，DB 服务器也在 DMZ，两个 API 服务器都可以通过 27017 端口与 DB 通信。DMZ API 服务器正在对 DB执行插入操作，并正在侦听“更新”事件以向用户返回响应，而 Intranet API 服务器正在侦听插入事件并仅对这些文档进行更新。一旦 Intranet API 更新文档，就会从 DMZ API 向用户返回响应。希望到目前为止这是有道理的。

这是一个设置，我对数据库服务器有疑问。它抱怨交换内存总是满的，因为它是副本集，所以里面有 3 台服务器，每台都有 4GB 的 RAM。我是否需要添加更多 RAM，如果有人知道，需要多少？

带有负载平衡的mongo更改流

任何人都可以帮助我们如何使用负载平衡服务器实现芒果更改流..？当我进行负载平衡时，我们正在研究微服务架构面临的生产问题，当我在单个服务器上执行任何操作时，相同的代码部署在 4 个服务器上，从所有 4 个服务器触发更改流触发器。所以我想要它应该从执行操作的同一服务器触发

提前致谢

Spring Data Elasticsearch 是否具有 changeStream 功能，例如 mongo？我正在尝试获取数据更改。我正在处理以下场景：我们创建一个 foo 对象。这个 foo 对象可以被多个用户更新。需要将更改推送到 UI。在 mongo 中，您可以使用 ReactiveMongoTemplate changeStream 方法。

当前的 Mongo 实现 (Kotlin)

使用响应式 mongo 模板，我试图只为插入操作收听 6 个集合的 mongo 更改流。我开始收听更改流的代码如下：

更改流启动：

在每个新条目上，我将简历令牌保存在我的一个集合中：

使用变更流选项进行启动。如果集合中存在恢复令牌，请使用它。否则，立即恢复。

抛出的错误：

到目前为止，我对这个功能的工作有不同的体验。在没有现有恢复令牌的情况下启动的应用程序始终按预期工作。当我使用现有的恢复令牌重新启动应用程序时的其他发现如下：

1. 它有时对所有 6 个都非常有效。
2. 刚重新启动后，它启动了一些，其余的则失败了。
3. 重新启动后，启动没有抛出任何错误。但是在正在监视的集合中插入文档时，很少/全部出错。

我了解更改流取决于文档中引用的 oplogs 历史记录。更令我惊讶的是，出错的简历令牌与我现有的任何简历令牌都不匹配，并且也不存在于 oplogs 中。

我已经验证提交给响应式 mongo 模板的简历令牌始终是正确的。

1. 如果我遗漏了什么，请告诉我。
2. 另外，我很想知道如何处理运行的许多更改流中的一个/几个失败。

我很好奇在可用性和性能方面，使用可尾游标与带有过滤器的更改流对上限集合进行插入操作的优缺点。

我有一个包含更新对象的上限集合。此集合唯一可能的操作是插入新更新（作为新记录），我需要将其中继到我的应用程序后端并通过 SSE/WebSockets 实时分发。我将在订阅查询中使用单个参数，它是一个时间戳，它是对象属性的一部分。该集合在该字段上有一个索引。我还将对新添加的记录进行一些基本过滤，因此 Change Streams 的聚合框架会有所帮助。

我读过：MongoDB 中的 changeStream 和可尾游标之间有什么区别，它总结了可尾游标和更改流之间的区别。

此外，关于Capped Collections的 mongodb 文章只说明了 Tailable 游标方法，而关于 Tailable 游标的文章指出它们不使用索引，您应该使用普通游标来获取结果。

简而言之，我需要一个可靠的集合更新流，完全基于在集合中插入新记录。我应该采用哪种方法？

在我的网络应用程序中，我有关注者和通知。问题是如何获取用户列表并报告某些文档的更改，例如更改的价格，或者用户创建了更多属性等。所以我可以使用更改流来观察集合内的更改。

这段代码对我来说看起来很疯狂，我的意思是即使有一些验证等。这又是矫枉过正。

也许更好的方法是控制器内部的一些逻辑，但是我如何在他成功创建或更新文档后首先向用户返回响应，然后运行一些函数，我可以在关注者列表中找到所有用户并为他们创建通知？

你们怎么看？通知系统的示例和解决方案不多，是否有任何资源可以提供帮助

我正在尝试使用更改流来监听我的 mongodb 集群中的更改；但是在学习了几个教程之后，我使用猫鼬的最终实现不起作用。如何使用当前的猫鼬连接来监听数据库中的变化

猫鼬连接：

更改流：

如何使用 mongodb 驱动程序来监视数据库中的更改，同时使用 mongoose 进行数据库连接。Mongoose 文档提供了很少的关于如何使用变更流的示例。在我的情况下，更改流仅在没有提供管道选项时才起作用。我认为这是管道对象的语法问题，但我仔细遵循了 mongodb 文档中的示例。问题要么在于管道对象，要么在于更改流是如何使用 mongoose 实现的。任何帮助表示赞赏。

这是我目前的方法（不起作用）：

我想编写反应式 mongodb 更改并更新 ConcurrentHashMap 的 api，但是当我使用 MongoDatabase.Collection.watch() 时，我的程序会进入无限循环

是否有可能它会在后台运行并且程序会继续运行？或创建多手表收藏？

我不知道这是否是一个严重的问题，因为该应用程序不是为人们在进行数据更新时不断刷新应用程序而设计的。但它仍然感觉像代码气味，似乎是明显的内存泄漏。

具体发生的情况是，在与 MongoDB 的连接期间，changeStream如果我刷新应用程序，然后在它通过连接触发时更新数据，则数据将被复制，并且它会增加我刷新的次数。因此，如果我刷新浏览器 5 次，那么在数据更新时通过管道传输的数据将是 5 组通过流传输的相同数据。

从表面上看，我认为这不是 MongoDB 的changeStream做法，但我不是 100% 确定。当我自己使用changeStream它时，没有 SSE 连接，这不会发生。无论我刷新浏览器多少次，都会按预期进行一对一的更改。

我认为可以缓解这个问题的是，在关闭浏览器或刷新它会触发连接关闭事件时，我可以结束 SSE 连接并重新启动它或刷新它。但是，当我从客户端关闭 SSE 连接时，它不会在刷新重置后重新建立。我相信关闭连接会清除流并重新开始，但我不确定。

架构如下。有一个路由设置为服务器发送事件 SSE，它将端点连接到 Angular 客户端。

这是changeStream代码：

客户端代码如下：

我不知道这是否很重要，或者我是否应该忽略它，因为它发生的可能性很小，但我觉得设置有问题，我需要关闭 SSE 连接并刷新或重新启动它或者是其他东西。