能否脱离Spring Boot运行Reactive MongoDB驱动？求代码示例

完全可以脱离Spring Boot使用Reactive MongoDB驱动，你遇到的无修改、无错误问题，核心原因是Reactor的异步数据流未被订阅——Reactor的Mono/Flux属于冷流，只有调用subscribe()方法后，操作才会实际执行。

以下是完整的代码示例，基于你提供的依赖配置实现：

依赖确认 #

你的依赖配置是正确的，包含了官方响应式Mongo驱动、Reactor核心、Spring上下文及日志组件：

implementation("org.mongodb:mongodb-driver-reactivestreams:4.11.1")
implementation("org.slf4j:slf4j-api:2.0.13")
implementation("ch.qos.logback:logback-classic:1.5.6")
implementation("org.springframework:spring-context:6.1.6")
implementation("io.projectreactor:reactor-core:3.6.5")

1. Spring配置类 #

用于创建Mongo客户端和指定集合的Bean：

import com.mongodb.reactivestreams.client.MongoClients;
import com.mongodb.reactivestreams.client.MongoClient;
import com.mongodb.reactivestreams.client.MongoCollection;
import org.bson.Document;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class MongoConfig {

    @Bean
    public MongoClient mongoClient() {
        // 替换为你的MongoDB连接地址
        return MongoClients.create("mongodb://localhost:27017");
    }

    @Bean
    public MongoCollection<Document> userCollection(MongoClient mongoClient) {
        // 指定要操作的数据库和集合
        return mongoClient.getDatabase("test_db").getCollection("users");
    }
}

2. 数据操作类 #

封装MongoDB的增删改查操作，将驱动返回的Publisher转换为Reactor的Mono类型：

import com.mongodb.reactivestreams.client.MongoCollection;
import org.bson.Document;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;

@Component
public class UserRepository {

    private final MongoCollection<Document> userCollection;

    public UserRepository(MongoCollection<Document> userCollection) {
        this.userCollection = userCollection;
    }

    // 插入用户
    public Mono<Void> insertUser(String username, int age) {
        Document user = new Document("username", username)
                .append("age", age);
        return Mono.from(userCollection.insertOne(user))
                .then(); // 仅返回完成信号，忽略插入结果
    }

    // 更新用户年龄
    public Mono<Long> updateUserAge(String username, int newAge) {
        Document filter = new Document("username", username);
        Document update = new Document("$set", new Document("age", newAge));
        return Mono.from(userCollection.updateOne(filter, update))
                .map(result -> result.getModifiedCount());
    }

    // 删除用户
    public Mono<Long> deleteUser(String username) {
        Document filter = new Document("username", username);
        return Mono.from(userCollection.deleteOne(filter))
                .map(result -> result.getDeletedCount());
    }
}

3. 应用启动类 #

初始化Spring上下文，执行数据库操作并订阅数据流：

import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import reactor.core.publisher.Mono;

public class MainApp {

    public static void main(String[] args) {
        AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(MongoConfig.class, UserRepository.class);
        UserRepository userRepository = context.getBean(UserRepository.class);

        // 插入操作：必须调用subscribe()才会执行
        userRepository.insertUser("Alice", 25)
                .doOnSuccess(v -> System.out.println("用户插入成功"))
                .doOnError(e -> System.err.println("插入失败：" + e.getMessage()))
                .subscribe();

        // 更新操作
        userRepository.updateUserAge("Alice", 26)
                .doOnSuccess(count -> System.out.println("更新了" + count + "条记录"))
                .doOnError(e -> System.err.println("更新失败：" + e.getMessage()))
                .subscribe();

        // 删除操作
        userRepository.deleteUser("Alice")
                .doOnSuccess(count -> System.out.println("删除了" + count + "条记录"))
                .doOnError(e -> System.err.println("删除失败：" + e.getMessage()))
                .subscribe();

        // 异步操作需要主线程等待，避免程序直接退出（实际生产环境建议用Reactor线程调度替代）
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }

        context.close();
    }
}

关键注意事项 #

• 必须调用subscribe()：所有Reactor数据流只有被订阅后才会触发执行，这是你之前操作无响应的核心原因。
• 线程管理：示例中用Thread.sleep()是为了简化演示，生产环境建议使用Reactor的block()方法（同步等待）或通过调度器管理异步线程生命周期。
• 错误处理：通过doOnError()或onErrorResume()捕获并处理异常，避免错误被静默忽略。

内容的提问来源于stack exchange，提问作者Антон