SpringBoot集成MongoDB ReactiveRepository连接数暴涨问题求助

1. Reactive仓库继承错误：你的SomeCoolNameReactiveRepository继承了非响应式的MongoRepository，而非ReactiveMongoRepository，这会导致Spring创建非响应式仓库实现，额外占用连接池资源。
2. 连接池未共享：你手动定义了reactiveMongoClient，同时系统会自动配置非响应式MongoClient，两者各自维护独立连接池，叠加后导致连接数暴涨。
3. Listener容器线程模型不匹配：DefaultMessageListenerContainer使用固定线程池（15个线程），每个ChangeStream需要长期占用一个连接，线程池过大直接耗尽连接池。
4. 过滤路径错误：ChangeStream事件中，更新后的完整文档存储在fullDocument字段下，原过滤条件my_collection.fieldToAudit路径无效，可能导致无意义的连接占用。
5. 错误处理不完善：仅打印错误日志，未处理ChangeStream中断后的重连延迟，可能引发无限重连创建新连接。

1. 修正Reactive仓库定义 #

改为继承ReactiveMongoRepository，确保使用响应式实现：

@Component
public interface SomeCoolNameReactiveRepository extends ReactiveMongoRepository<MyCollection, String> {

}

2. 统一Mongo客户端，复用连接池 #

删除手动定义的reactiveMongoClient和reactiveMongoTemplate，依赖Spring自动配置的共享响应式客户端，同时通过配置统一控制连接池参数。修改MongoStreamListenerConfig：

@Configuration
@Slf4j
@EnableReactiveMongoRepositories(basePackages = "com.mongodb.repository")
public class MongoStreamListenerConfig {

    @Value("${spring.data.mongodb.uri}")
    String databaseUrl;

    @Bean
    ReactiveMessageListenerContainer changeStreamListenerContainer(ReactiveMongoTemplate reactiveMongoTemplate,
                                                                   MongoMessageListener consentAuditListener) {
        ReactiveMessageListenerContainer container = new ReactiveMessageListenerContainer(reactiveMongoTemplate);
        // 使用响应式调度器，避免固定线程池耗尽连接
        container.setTaskScheduler(Schedulers.boundedElastic());

        ChangeStreamRequest<MyDocument> request =
                ChangeStreamRequest.builder(consentAuditListener)
                        .collection("my_collection")
                        .filter(newAggregation(match(
                                where("operationType").in("replace", "update")
                                        .and("fullDocument.fieldToAudit").exists(true))
                        ))
                        .fullDocumentLookup(FullDocument.UPDATE_LOOKUP)
                        .build();

        container.register(request, MyDocument.class)
                .doOnError(e -> log.error("ChangeStream registration failed", e))
                .subscribe();
        log.info("> Mongo Stream Listener registered, database: {}", new ConnectionString(databaseUrl).getDatabase());
        return container;
    }

    @Bean
    ErrorHandler getLoggingErrorHandler() {
        return throwable -> {
            log.error("Audit ChangeStream error", throwable);
            // 添加重连延迟，避免无限循环创建连接
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        };
    }
}

3. 统一连接池配置 #

在application.properties中添加全局连接池参数，同时控制响应式和非响应式客户端：

# MongoDB连接池核心配置
spring.data.mongodb.connection.max-pool-size=20
spring.data.mongodb.connection.min-pool-size=5
spring.data.mongodb.connection.max-idle-time=300000
spring.data.mongodb.connection.max-life-time=600000
# ChangeStream专属配置
spring.data.mongodb.change-stream.max-await-time=10000

4. 优化Listener中的阻塞操作 #

将SQS发送和非响应式仓库操作包装为响应式调用，避免阻塞响应式线程：

@Component
@Slf4j
public class MongoMessageListener implements MessageListener<ChangeStreamDocument<Document>, MyDocument> {

    @Autowired
    OtherNonReactiveRepositoryINeed otherNonReactiveRepositoryINeed;
    @Autowired
    QueueMessagingTemplate queueMessagingTemplate;

    @Value("${awsService.sqsAuditQueue}")
    String auditQueue;

    @Override
    public void onMessage(Message<ChangeStreamDocument<Document>, MyDocument> message) {
        // 包装阻塞操作为响应式任务
        Mono.fromCallable(() -> {
            OperationType operationType = message.getRaw().getOperationType();
            log.info("Received {} operation on collection: {}", operationType, message.getProperties().getCollectionName());
            
            // 原有的SQS发送和仓库操作逻辑
            queueMessagingTemplate.convertAndSend(auditQueue, buildAuditPayload(message));
            otherNonReactiveRepositoryINeed.save(getAuditData(message));
            return null;
        }).subscribeOn(Schedulers.boundedElastic())
          .doOnError(e -> log.error("Failed to process audit message", e))
          .subscribe();
    }

    // 补充缺失的工具方法
    private Object buildAuditPayload(Message<ChangeStreamDocument<Document>, MyDocument> message) {
        // 构造SQS消息逻辑
        return null;
    }

    private Object getAuditData(Message<ChangeStreamDocument<Document>, MyDocument> message) {
        // 构造仓库保存数据逻辑
        return null;
    }
}

5. 移除自定义Listener容器类 #

删除MongoStreamListenerContainer，改用Spring提供的ReactiveMessageListenerContainer，适配响应式客户端模型。

• 监控连接池状态：通过MongoDB自带的db.serverStatus().connections命令或Spring Actuator监控连接数、空闲数、活跃数，动态调整池大小。
• 确保单一ChangeStream实例：检查配置是否存在重复注册ChangeStream的情况，单个监听任务只需要一个长期连接。
• 限制非Reactive业务连接占用：如果非Reactive业务流量大，单独调整其连接池分配，避免抢占ChangeStream的连接资源。

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