Reactor Kafka使用flatMap后如何保证消息至少一次消费且不阻塞流

问题核心分析 #

你当前的问题根源是flatMap的并发处理逻辑会导致偏移量确认顺序和消息消费顺序不一致。比如同分区内偏移量为1、2、3的三条消息，可能偏移量3的消息先完成数据库存储，此时你确认偏移量3后，系统会默认1、2都已处理完成，如果这时候1、2的存库操作实际失败，也不会被重新消费，直接违反至少一次语义。

解决方案 #

方案1：使用concatMap顺序处理（低并发场景） #

如果消费吞吐量要求不高，直接将flatMap替换为concatMap即可。concatMap会严格按照原流的顺序处理消息，只有前一条消息的存库操作完成后才会处理下一条，确认顺序和消息顺序天然一致，不会出现跳偏移量确认的问题。
代码示例：

Flux<ReceiverRecord<String, byte[]>> kafkaFlux = KafkaReceiver.create(options).receive();
kafkaFlux.concatMap(r -> 
    store(r)
        // 存库成功才确认偏移量
        .doOnSuccess(unused -> r.receiverOffset().acknowledge())
        .doOnError(err -> {
            // 可自定义错误日志、重试逻辑，未确认的消息会在重平衡后重新消费
            log.error("消息处理失败，偏移量:{}", r.receiverOffset().offset(), err);
        })
).subscribe();

该方案完全非阻塞，只是牺牲了单流的并发度，适合消费速度满足业务要求的场景。

方案2：按分区分组+分组内顺序处理（兼顾并发与可靠性，生产推荐） #

Kafka的消费并发本身是按分区划分的，不同分区间的消息没有顺序要求，偏移量提交也是按分区独立统计的。你可以先把流按主题分区分组，每个分区的流单独用concatMap顺序处理，不同分区之间可以并行处理，既不会出现跳偏移量确认的问题，又能充分利用多分区的并发能力。
代码示例：

KafkaReceiver.create(options)
    .receive()
    // 按主题分区分组
    .groupBy(record -> record.receiverOffset().topicPartition())
    .flatMap(partitionFlux -> 
        // 每个分区内的消息按顺序处理，分区间并行处理
        partitionFlux.concatMap(r -> 
            store(r)
                .doOnSuccess(unused -> r.receiverOffset().acknowledge())
                .doOnError(err -> {
                    log.error("分区{}消息处理失败，偏移量:{}", 
                        r.receiverOffset().topicPartition(), 
                        r.receiverOffset().offset(), 
                        err);
                    // 可根据需求添加重试逻辑，或者触发消费暂停避免连续报错
                })
        )
    )
    .subscribe();

该方案是生产环境最常用的实现，完全非阻塞，符合至少一次语义要求。

方案3：自定义偏移量跟踪（同分区高并发场景） #

如果你确实需要同一个分区内的消息并发处理，同时要保证至少一次语义，可以自主维护待确认的偏移量队列，只有前面的偏移量全部处理完成后，才批量确认最大的连续成功偏移量。
实现思路参考：

• 每个分区维护一个有序队列，记录每个偏移量的处理状态（处理中/成功/失败）
• 每条消息存库成功后，更新对应偏移量的状态为成功
• 每次状态更新后，从最小的未提交偏移量开始检查连续成功的偏移量，找到最大的连续成功偏移量后再调用确认方法
• 如果有偏移量处理失败，暂停当前分区的消费，重试失败的消息或者触发告警

该方案复杂度最高，但可以实现同分区内的高并发处理，适合吞吐量要求极高的场景。

额外可靠性优化建议 #

• 可以给store操作添加合理的重试机制，比如用retryWhen配置可重试异常的重试次数，避免偶发数据库抖动导致消息消费失败
• 配置ReceiverOptions.maxDeferredCommits参数限制待提交的偏移量数量，避免内存溢出
• 如果数据库支持事务，可以把消息处理和偏移量记录放在同一个事务中，即可实现恰好一次语义。

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