Spring Data MongoDB Reactive：如何获取db.runCommand()的执行结果？

核心思路 #

在Reactive编程模型下，不能像同步代码那样直接阻塞取值，需利用Reactive订阅机制或在初始化这类特殊场景下短暂阻塞获取结果（不影响整体Reactive特性）。

具体实现方案 #

方案1：基于ObservableSubscriber同步获取结果 #

你已用到SubscriberHelpers.ObservableSubscriber，可通过它的get()方法阻塞等待命令执行结果：

private boolean isChangeStreamsEnabled() {
    final MongoDatabase db = mongoClient.getDatabase("admin");
    final Document document = Document.parse(isChangeStreamsEnabled);
    final Bson bson = document.toBsonDocument();
    final Publisher<Document> commandResult = db.runCommand(bson);

    SubscriberHelpers.ObservableSubscriber<Document> subscriber = new SubscriberHelpers.ObservableSubscriber<>();
    commandResult.subscribe(subscriber);
    
    // 阻塞等待结果，可自定义超时时间
    List<Document> results = subscriber.get(10, TimeUnit.SECONDS);
    // 根据AWS DocumentDB返回结构判断Change Streams是否启用
    return !results.isEmpty() 
           && results.get(0).containsKey("cursor")
           && !((Document) results.get(0).get("cursor")).getList("firstBatch", Document.class).isEmpty();
}

方案2：结合Reactor API实现（推荐，贴合Spring Reactive生态） #

如果项目依赖Reactor（Spring Boot Reactive默认引入），可将Publisher转为Mono，用block()同步获取结果：

import reactor.core.publisher.Mono;
import java.time.Duration;

private boolean isChangeStreamsEnabled() {
    Document command = Document.parse(isChangeStreamsEnabled);
    return Mono.from(mongoClient.getDatabase("admin").runCommand(command.toBsonDocument()))
            .map(doc -> {
                Document cursor = doc.get("cursor", Document.class);
                return cursor != null && !cursor.getList("firstBatch", Document.class).isEmpty();
            })
            .blockOptional(Duration.ofSeconds(10))
            .orElse(false);
}

完整初始化逻辑调整 #

修改initializeDB方法，先检查状态再决定是否执行启用命令：

public void initializeDB() {
    if (!isChangeStreamsEnabled()) {
        enableChangeStreams();
    }
}

private void enableChangeStreams() {
    final MongoDatabase db = mongoClient.getDatabase("admin");
    final Document document = Document.parse(enableChangeStreams);
    final Bson bson = document.toBsonDocument();
    
    SubscriberHelpers.ObservableSubscriber<Document> subscriber = new SubscriberHelpers.ObservableSubscriber<>();
    db.runCommand(bson).subscribe(subscriber);
    // 等待启用命令执行完成
    subscriber.get(10, TimeUnit.SECONDS);
}

注意事项 #

• 仅在应用启动初始化场景使用阻塞调用，业务逻辑中保持Reactive非阻塞特性。
• 替换命令字符串中空的数据库名、集合名为实际业务值。
• 可根据AWS DocumentDB实际返回结构调整结果判断逻辑，确保准确性。

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