如何在Java Reactor Core中控制Flux发布者的活跃订阅者数量？

你遇到的核心问题是Mongo Reactive的findAll()返回冷Flux，每个订阅都会触发一次独立的Mongo查询，过多并发订阅会导致Mongo压力过载。不需要重写Publisher的subscribe()方法，用Reactor的基础操作符结合信号量就能实现订阅数控制。

实现方案：用信号量限制并发订阅 #

1. 首先定义一个信号量，指定允许的最大活跃订阅数（比如10）：

private final Semaphore subscriptionSemaphore = new Semaphore(10);

2. 包装你的findAll Flux，在订阅前获取信号量许可，订阅结束后释放许可：

public Flux<YourEntity> getLimitedEntities() {
    return Mono.defer(() -> {
            try {
                // 获取许可，无许可时阻塞等待
                subscriptionSemaphore.acquire();
                return Mono.just(true);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return Mono.error(new RuntimeException("获取订阅许可时被中断", e));
            }
        })
        // 获取许可后发起Mongo查询
        .flatMapMany(ignored -> reactiveMongoTemplate.findAll(YourEntity.class))
        // 无论正常结束还是异常终止，都释放许可
        .doFinally(signalType -> subscriptionSemaphore.release());
}

关键说明 #

• 冷发布者特性：Spring Mongo Reactive的查询方法返回冷Flux，每个订阅都会触发新的查询，这是你需要控制订阅数的根本原因。
• 信号量的作用：Semaphore会限制同时获取许可的线程数，也就是同时活跃的订阅数，只有当某个订阅完成（或报错）释放许可后，新的订阅才能发起查询。
• 不需要自定义Publisher：Reactor的Mono.defer和doFinally已经能完美实现订阅前后的资源控制逻辑，避免重复造轮子。

额外场景：如果是控制单订阅内的并发处理 #

如果你是想在单个Flux订阅中控制数据处理的并发数（比如处理每个实体时调用其他异步操作），可以直接用flatMap的并发参数：

flux.flatMap(this::processEntity, 10); // 最多同时处理10个实体

但这和你要控制订阅者数量的场景不同，按需选择即可。

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