被动Mongo:已超过500的最大操作数（maxQueueWaitSize）

Question

我正在使用ReactiveMongoTemplate MongoDB ChangeStream来监听对MongoDB集合的更新，执行一些查询，并将文档持久化到另一个集合。虽然它在本地运行良好，但在部署到容量很大的UAT后，它开始出现以下错误：

Too many operations are already waiting for a collection. Max number of operations (maxWaitQueueSize) of 500 has been exceeded.

解决这个问题的方法是什么？

我的档案里有application.yml

spring:
   data: 
      mongodb:
          uri: mongodb://host:port/db?authMechanism=<val1>&authSource=<val2>&authechanismProperties=<val3>

这就是简化更改拖缆的外观：

@Autowired
ReactiveMongoTemplate reactiveMongoTemplate;

reactiveMongoTemplate
  .changeStream(Sample.class)
  .watchCollection("sample_collection")
  .filter(
     new Criteria.orOperator(
        where("operationType").is("update"),
        where("operationType").is("insert")
     )
  )
  .listen()
  .flatMap(r->processMessage(r)). // processMessage does some queries to collections including this collection being listened to and upserts to same mongodb in a different collection
  .subscribeOn(Schedulers.boundedElastic())
  .subscribe();

我知道我可能需要添加一些连池才能处理多个连接？但是我如何使用Reactive MongoDB配置它？我是反应式编程的新手。任何指针都会很有帮助。

Answer 1

这里有几件事你可以做：

>

检查是否有一些长阻塞调用正在进行，导致线程被阻塞，并导致创建许多连接，因为之前的连接仍然被占用执行繁重的任务。尝试检查对阻塞这些调用的代码的一些优化。在反应式编程中，您可以使用BlockHound检查阻塞代码的存在。

通过指定waitQueueMultiple或maxPoolSize来增加连接限制https://docs.mongodb.com/manual/reference/connection-string/#connection-池选项

在此之前，您可以使用查看mongo db统计数据来查看当前和允许的连接

db.serverStatus().connections