问题标签 [changestream]

我已经开始处理 MongoDB 更改流，并设置了一些流来监视大约 30 个数据库的更改。我只需要观看两个集合，我想知道是否可以编写一些管道或过滤器来观看具有提供名称或部分名称的集合。例如，我从 ChangeStreamDocument 的 namespaceDocument 中获取集合名称，并且我想在所有数据库上观看等于“玩家”的集合。我正在监视整个服务器的更改，但现在我收到通知每个被改变的集合，我不想要那个，只有特定的集合。

我在想是否可以在此处放置一些过滤器以使流仅监视具有相应名称的集合。

文档：https ://docs.mongodb.com/manual/changeStreams/

我想在 mongo db 上设置一个更改流，但是有一个“无法解决重载方法订阅”，我无法找到问题的根源：

这个简单的示例导致异常

该异常似乎是无害的，并且是异步记录的。

我能做些什么呢？

开始使用 AWS DocumentDb 更改流，希望将写入数据库更改流的事件保留 24 小时，而不是默认的 3 小时。我怎么能做到这一点。标准 AWS 文档仅提供以下内容：

我正在尝试解决 mongodb 更改流错误。我正在关注与...的集合的变化

使用处理程序

和路由器功能

对于我的前端，我使用 Angular 使用行为主题和 Eventsource 进行连接。我遇到的问题是在我关闭页面后，即使前端已关闭事件源，changeStream 似乎仍在监视更改。结果，当更新文档时，spring 会引发以下错误：

2020-01-06 15:22:34.312 错误 27772 --- [andler-executor] org.mongodb.driver.operation：回调 onResult 调用产生错误

com.mongodb.MongoException：状态应该是：在 com.mongodb.MongoException.fromThrowableNonNull(MongoException.java:79) ~[mongodb-driver-core-3.11.1.jar:na] 在 com.mongodb.async.client 打开.AbstractSubscription.onError(AbstractSubscription.java:135) ~[mongodb-driver-async-3.11.1.jar:na] at com.mongodb.async.client.MongoIterableSubscription$2.onResult(MongoIterableSubscription.java:93) ~[mongodb -driver-async-3.11.1.jar:na] at com.mongodb.async.client.MongoIterableSubscription$2.onResult(MongoIterableSubscription.java:85) ~[mongodb-driver-async-3.11.1.jar:na] at com.mongodb.operation.AsyncChangeStreamBatchCursor$3.onResult(AsyncChangeStreamBatchCursor.java:133) ~[mongodb-driver-core-3.11.1.jar:na] at com.mongodb.operation.AsyncChangeStreamBatchCursor$3.onResult(AsyncChangeStreamBatchCursor.java:129) ~[mongodb-driver-core-3.11.1.jar:na] 在 com.mongodb.internal.async.ErrorHandlingResultCallback.onResult(ErrorHandlingResultCallback.java:49) [mongodb-driver-core-3.11.1.jar: na] 在 com.mongodb.operation.AsyncChangeStreamBatchCursor$4.onResult(AsyncChangeStreamBatchCursor.java:168) [mongodb-driver-core-3.11.1.jar:na] 在 com.mongodb.operation.AsyncChangeStreamBatchCursor$4.onResult(AsyncChangeStreamBatchCursor.java :159) [mongodb-driver-core-3.11.1.jar:na] 在 com.mongodb.internal.async.ErrorHandlingResultCallback.onResult(ErrorHandlingResultCallback.java:49) [mongodb-driver-core-3.11.1.jar: na] 在 com.mongodb.operation.AsyncQueryBatchCursor$CommandResultSingleResultCallback.onResult(AsyncQueryBatchCursor.java:331) [mongodb-driver-core-3.11.1.jar:na] 在 com.mongodb.operation。AsyncQueryBatchCursor$CommandResultSingleResultCallback.onResult(AsyncQueryBatchCursor.java:310) [mongodb-driver-core-3.11.1.jar:na] at com.mongodb.internal.async.ErrorHandlingResultCallback.onResult(ErrorHandlingResultCallback.java:49) [mongodb-driver -core-3.11.1.jar:na] 在 com.mongodb.internal.connection.DefaultServer$DefaultServerProtocolExecutor$2.onResult(DefaultServer.java:242) [mongodb-driver-core-3.11.1.jar:na] 在 com .mongodb.internal.async.ErrorHandlingResultCallback.onResult(ErrorHandlingResultCallback.java:49) [mongodb-driver-core-3.11.1.jar:na] 在 com.mongodb.internal.connection.CommandProtocolImpl$1.onResult(CommandProtocolImpl.java: 83) [mongodb-driver-core-3.11.1.jar:na] 在 com.mongodb.internal.connection.DefaultConnectionPool$PooledConnection$1.onResult(DefaultConnectionPool.java:467) [mongodb-driver-core-3.11.1.jar:na] 在 com.mongodb.internal.connection.UsageTrackingInternalConnection$2.onResult(UsageTrackingInternalConnection.java:111) [mongodb-driver-core-3.11.1.jar: na] 在 com.mongodb.internal.async.ErrorHandlingResultCallback.onResult(ErrorHandlingResultCallback.java:49) [mongodb-driver-core-3.11.1.jar:na] 在 com.mongodb.internal.connection.InternalStreamConnection$2$1.onResult (InternalStreamConnection.java:401) [mongodb-driver-core-3.11.1.jar:na] at com.mongodb.internal.connection.InternalStreamConnection$2$1.onResult(InternalStreamConnection.java:376) [mongodb-driver-core- 3.11.1.jar:na] 在 com.mongodb.internal.connection.InternalStreamConnection$MessageHeaderCallback$MessageCallback.onResult(InternalStreamConnection.java:677) [mongodb-driver-core-3.11.1.jar:na] 在 com.mongodb .internal.connection.InternalStreamConnection$MessageHeaderCallback$MessageCallback.onResult(InternalStreamConnection.java:644) [mongodb-driver-core-3.11.1.jar:na] 在 com.mongodb.internal.connection.InternalStreamConnection$5.completed(InternalStreamConnection.java :514) [mongodb-driver-core-3.11.1.jar:na] 在 com.mongodb.internal.connection.InternalStreamConnection$5.completed(InternalStreamConnection.java:511) [mongodb-driver-core-3.11.1.jar :na] 在 com.mongodb.internal.connection.AsynchronousChannelStream$BasicCompletionHandler.completed(AsynchronousChannelStream.java:220) [mongodb-driver-core-3.11.1.jar:na] 在 com.mongodb.internal.connection.AsynchronousChannelStream$ BasicCompletionHandler.completed(AsynchronousChannelStream.java:203) [mongodb-driver-core-3.11.1.jar:na] 在 com.mongodb.internal.connection.tlschannel。async.AsynchronousTlsChannel$3$1.run(AsynchronousTlsChannel.java:151) [mongodb-driver-core-3.11.1.jar:na] at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [na :1.8.0_172] 在 java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:266) [na:1.8.0_172] 在 java.util.concurrent.FutureTask.run(FutureTask.java) [na :1.8.0_172] 在 java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [na:1.8.0_172] 在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [na :1.8.0_172] 在 java.lang.Thread.run(Thread.java:748) [na:1.8.0_172] 引起：java.lang.IllegalStateException：状态应该是：在 com.mongodb.assertions.Assertions 处打开。 isTrue(Assertions.java:70) ~[mongodb-driver-core-3.11.1.jar:na] 在 com.mongodb.internal.session。BaseClientSessionImpl.advanceOperationTime(BaseClientSessionImpl.java:107) ~[mongodb-driver-core-3.11.1.jar:na] at com.mongodb.internal.session.ClientSessionContext.advanceOperationTime(ClientSessionContext.java:70) ~[mongodb-driver -core-3.11.1.jar:na] 在 com.mongodb.internal.connection.ClusterClockAdvancingSessionContext.advanceOperationTime(ClusterClockAdvancingSessionContext.java:76) ~[mongodb-driver-core-3.11.1.jar:na] 在 com.mongodb .internal.connection.InternalStreamConnection.updateSessionContext(InternalStreamConnection.java:537) ~[mongodb-driver-core-3.11.1.jar:na] at >>com.mongodb.internal.connection.InternalStreamConnection.access$800(InternalStreamConnection.java :76) ~[mongodb-driver-core-3.11.1.jar:na] 在 >>com.mongodb.internal.connection.InternalStreamConnection$2$1。onResult(InternalStreamConnection.java:385) [mongodb-driver-core-3.11.1.jar:na] ...省略了14个常用帧

我搜索了高低，但找不到解决方案。任何帮助，将不胜感激！谢谢！

我使用 MongoDB 4.0 作为 3 个节点的副本集。2 个数据节点和 1 个仲裁器。我的应用程序（ASP.NET Core 2.2）使用更改流。考虑这种情况，我的主节点关闭，我的辅助节点成为主节点，我在我的副本集中用一个新节点替换关闭的节点并开始同步。在这种情况下，我的变更流会发生什么？

我的第二个问题是，在我的应用程序readPreference中，secondary当我的辅助节点出现故障时，我希望 MongoDB 自动读取，primary但它没有发生，我的应用程序在读取时出现问题。正常吗？

考虑我有一个包含 3 个节点（2 个数据节点和一个仲裁器 (PSA)）的副本集。当由于某种原因我的一个数据节点出现故障并将其带回来时，在与主节点同步期间，它处于 state STARTUP2。在他那个时候，我会失去我的change stream，因为我的副本集有 2 个数据节点，但我没有大部分节点可以读取。

我该如何处理这个问题？

我还阅读了这个 MongoDB 文档。是否可以将主节点priority值设置为高于辅助节点（即与主节点同步）？即使我的辅助节点处于STARTUP2状态，我也可以通过这样做获得多数吗？

问题：
推理生成过程每秒将大约 300 个推理数据写入 MongoDB 集合。另一个进程利用 MongoDB 的更改流特性来回读这些推论并进行后处理。目前，调用变更流函数API（mongoc_change_stream_next()）时只返回一个推理数据。因此，总共需要 300 次这样的调用才能在 1 秒内获取存储的所有推理数据。但是，每次读取后，需要大约 50ms 的时间来对单个/多个推理数据进行后处理。由于单一数据返回模型，引入了 15 倍的有效延迟。为了解决这个问题，我们正在尝试实施批量读取机制符合 MongoDB 的更改流功能。我们尝试了各种选项来实现相同的功能，但在每次更改流 API 调用后仍然只获得一个数据。有没有办法解决这个问题？

平台：
操作系统：Ubuntu 16.04
Mongo-c-driver：1.15.1
Mongo 服务器：4.0.12

尝试的选项：
将光标的批量大小设置为大于 1。

我最近一直在使用 pymongo 中的 Change Stream 框架来动态更新集合。

我的管道非常简单，如下所示：

它用于以下代码：

此更新经常发生。

我正在监视我的数据库的 COLLSCANS，并且我意识到 watch 方法引起的游标的 getMore 每次调用时都会执行一次 collscan。有时它是一个很小的 ​​collscan，包含一百个 docsExamined，但有时它检查的次数更多。

我找不到建立索引以消除此 collscan 的方法。我在想我错过了什么。我应该将参数传递给光标吗？我应该在某处建立索引吗？

预先感谢您的帮助 ！

我正在尝试使用 MongoDB 中的更改流来监视参与者数组包含用户 ID 的文档中的更改。

我了解 MongoDB 更改流允许用户查看单个字段匹配的文档上的更改；但是，我如何在“参与者”数组包含用户 ID 的地方观察变化？

此代码将参与者视为一个字段，并且只观察参与者等于的地方socket.uid；是否可以观看participants 包含 socket.uid在哪里？