使用ReactiveStreams获取MongoDB集合统计耗时过长，Vertx线程阻塞

问题根源 #

你的代码存在三个核心问题：

• 错误处理不完整：onError方法仅打印日志但未完成CompletableFuture，异常场景下Future会一直挂起，最终触发超时。
• 未绑定Vertx事件循环：手动实现的Subscriber回调默认不在Vertx事件线程中执行，可能阻塞事件循环线程，拖慢整体响应。
• 流处理不规范：onComplete未做收尾处理，且onNext后未取消订阅，易造成资源泄漏。

修复现有代码的方案 #

修改getCollectionStats方法，补全错误处理、绑定Vertx上下文、规范流处理：

import com.mongodb.reactivestreams.client.MongoDatabase;
import org.bson.Document;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import io.vertx.core.Vertx;

com.mongodb.reactivestreams.client.MongoClient mongoReactveStreamsClient;
private final Vertx vertx; // 注入或初始化Vertx实例

private CompletableFuture<String> getCollectionStats(String collectionName) {
    MongoDatabase database = mongoReactveStreamsClient.getDatabase(dbName);
    CompletableFuture<String> result = new CompletableFuture<>();

    database.runCommand(new Document("collStats", collectionName)).subscribe(new Subscriber<Document>() {
      private Subscription subscription;

      @Override
      public void onSubscribe(Subscription s) {
        subscription = s;
        // 在Vertx事件循环中请求数据，避免阻塞
        vertx.runOnContext(v -> s.request(1));
      }

      @Override
      public void onNext(Document document) {
        vertx.runOnContext(v -> {
          result.complete(document.toJson());
          subscription.cancel(); // 获取结果后取消订阅，释放资源
        });
      }

      @Override
      public void onError(Throwable t) {
        vertx.runOnContext(v -> {
          logger.error("获取集合统计失败: {}", t.getMessage(), t);
          result.completeExceptionally(t); // 完成Future的异常状态
        });
      }

      @Override
      public void onComplete() {
        // 无结果返回时标记Future异常
        if (!result.isDone()) {
          vertx.runOnContext(v -> result.completeExceptionally(new RuntimeException("获取集合统计无结果返回")));
        }
      }
    });
    return result;
}

更优方案：使用Vertx官方MongoDB客户端 #

Vertx提供了适配自身响应式模型的MongoDB客户端，无需手动实现Subscriber，能完美融入Vertx事件循环，彻底避免阻塞问题。

1. 替换依赖

移除原生ReactiveStreams驱动，添加Vertx MongoDB客户端依赖：

<dependency>
  <groupId>io.vertx</groupId>
  <artifactId>vertx-mongo-client</artifactId>
  <!-- 版本与你的Vertx版本保持一致，例如4.5.1 -->
  <version>${vertx.version}</version>
</dependency>

2. 重写获取集合统计的方法

利用Vertx客户端的响应式API，代码更简洁高效：

import io.vertx.core.Future;
import io.vertx.mongo.client.MongoClient;
import org.bson.Document;

private final MongoClient vertxMongoClient; // 初始化Vertx的MongoClient实例

private Future<String> getCollectionStats(String collectionName) {
    Document command = new Document("collStats", collectionName);
    return vertxMongoClient.runCommand(command)
      .map(document -> document.toJson())
      .onFailure(t -> logger.error("获取集合统计失败: {}", t.getMessage(), t));
}

3. 批量获取多集合统计

用CompositeFuture并行执行多个请求，大幅提升效率：

import io.vertx.core.CompositeFuture;
import java.util.List;
import java.util.ArrayList;
import java.util.stream.Collectors;

public Future<List<String>> getMultipleCollectionStats(List<String> collectionNames) {
    List<Future<String>> futures = collectionNames.stream()
      .map(this::getCollectionStats)
      .collect(Collectors.toList());
    return CompositeFuture.all(futures)
      .map(compositeFuture -> {
          List<String> results = new ArrayList<>();
          for (int i = 0; i < compositeFuture.size(); i++) {
              results.add(compositeFuture.resultAt(i));
          }
          return results;
      });
}

额外优化建议 #

• 升级驱动版本：MongoDB ReactiveStreams 3.9.1版本较老，建议升级到与Vertx客户端兼容的新版，修复已知性能与阻塞问题。
• 排查MongoDB服务器：若服务器响应慢，需检查MongoDB的索引、CPU/内存占用情况，确保collStats命令能快速执行。

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