Spring 5+Reactor实现Mongo到Kafka每秒100条吞吐限制方案问询

这是个很典型的Reactor背压场景，你提出的buffer()+定时Fluxzip的思路完全靠谱，我来给你拆解具体实现细节和需要注意的边界情况：

核心思路解析 #

我们需要通过两个Flux的协作实现精准速率控制：

1. 数据缓冲流：从Mongo读取的数据流被缓冲成每100条一批，确保每次处理的量符合吞吐量要求；
2. 速率闸门流：每秒生成一个信号的定时Flux，作为“闸门”控制每批数据的发送时机；
3. Zip操作：将两个流zip在一起，只有当缓冲流攒够100条且闸门流发出信号时，才会处理下一批数据，从而严格控制每秒100条的吞吐量。

具体代码实现 #

1. 基础依赖与配置 #

确保你的项目已经引入Reactive Mongo和Reactive Kafka的依赖（比如Spring Boot对应的starter），然后配置Kafka生产者：

@Configuration
public class ReactiveKafkaConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Bean
    public ReactiveKafkaProducerTemplate<String, YourDataModel> reactiveKafkaProducerTemplate() {
        Map<String, Object> producerProps = new HashMap<>();
        producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        // 可选：配置重试、acks等参数保证消息可靠性
        producerProps.put(ProducerConfig.ACKS_CONFIG, "all");
        producerProps.put(ProducerConfig.RETRIES_CONFIG, 3);

        return new ReactiveKafkaProducerTemplate<>(SenderOptions.create(producerProps));
    }
}

2. 核心同步逻辑 #

假设你已经有了Reactive Mongo的Repository（返回Flux<YourDataModel>），接下来实现同步服务：

@Service
@Slf4j
public class MongoToKafkaSyncService {

    private final YourDataRepository dataRepository;
    private final ReactiveKafkaProducerTemplate<String, YourDataModel> kafkaProducer;

    // 构造注入依赖
    public MongoToKafkaSyncService(YourDataRepository dataRepository,
                                   ReactiveKafkaProducerTemplate<String, YourDataModel> kafkaProducer) {
        this.dataRepository = dataRepository;
        this.kafkaProducer = kafkaProducer;
    }

    public Flux<SendResult<String, YourDataModel>> startControlledSync() {
        // 1. 创建每秒触发一次的速率闸门
        Flux<Long> rateLimiter = Flux.interval(Duration.ofSeconds(1));

        // 2. 从Mongo读取数据，缓冲成每100条一批
        Flux<List<YourDataModel>> bufferedData = dataRepository.findAll()
                .buffer(100); // 攒够100条才输出批次

        // 3. Zip两个流，实现"每100条 + 每秒"的双重控制
        return Flux.zip(bufferedData, rateLimiter, (batch, timerSignal) -> batch)
                .flatMap(this::sendBatchToKafka)
                .doOnError(ex -> log.error("Sync failed with error", ex))
                .doOnComplete(() -> log.info("Mongo to Kafka sync completed"));
    }

    // 批量发送到Kafka
    private Flux<SendResult<String, YourDataModel>> sendBatchToKafka(List<YourDataModel> batch) {
        log.debug("Sending batch of {} messages to Kafka", batch.size());
        return Flux.fromIterable(batch)
                .flatMap(data -> kafkaProducer.send("your-target-topic", data.getId(), data))
                .doOnNext(result -> log.trace("Message sent, offset: {}", result.recordMetadata().offset()));
    }
}

3. 启动同步任务 #

可以通过ApplicationRunner在项目启动时自动触发同步：

@Component
public class SyncStartupRunner implements ApplicationRunner {

    private final MongoToKafkaSyncService syncService;

    public SyncStartupRunner(MongoToKafkaSyncService syncService) {
        this.syncService = syncService;
    }

    @Override
    public void run(ApplicationArguments args) {
        // 非阻塞启动同步，如需阻塞等待完成可改用blockLast()
        syncService.startControlledSync()
                .subscribe();
    }
}

边界情况处理 #

场景1：最后一批数据不足100条 #

上面的代码中，如果Mongo剩余数据不足100条，buffer(100)会一直等待，直到攒够数量。如果希望即使不足100条也能在1秒内发送，可以把buffer(100)替换为bufferTimeout(100, Duration.ofSeconds(1))：

Flux<List<YourDataModel>> bufferedData = dataRepository.findAll()
        .bufferTimeout(100, Duration.ofSeconds(1)); // 满足任一条件就输出批次

这种方式更灵活，适合数据量波动的场景，吞吐量会稳定在每秒约100条。

场景2：Kafka发送失败 #

可以在发送逻辑中添加重试机制，比如：

private Flux<SendResult<String, YourDataModel>> sendBatchToKafka(List<YourDataModel> batch) {
    return Flux.fromIterable(batch)
            .flatMap(data -> kafkaProducer.send("your-target-topic", data.getId(), data)
                    .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))) // 最多重试3次，指数退避
                    .onErrorResume(ex -> {
                        log.error("Failed to send data: {}", data.getId(), ex);
                        return Mono.empty(); // 或者记录失败数据后续重试
                    }));
}

关键原理说明 #

• Reactor背压支持：Reactive Mongo的findAll()返回的Flux天生支持背压，当下游的速率控制生效时，Mongo会自动放慢数据读取速度，不会一次性加载全量数据到内存，这对超大规模集合特别重要；
• Zip操作的阻塞特性：Flux.zip()会等待所有输入流都产生元素才会继续处理，所以即使Mongo的缓冲流很快攒够100条，也会被速率闸门流强制等待1秒，从而严格控制发送速率。

内容的提问来源于stack exchange，提问作者riccardo.cardin