使用Reactor Kafka消费时事件丢失问题排查及代码优化建议

云原生阿岚

2026-7-21

问题描述

使用Reactor Kafka消费事件时，向队列推送7条事件，但消费者仅消费到5条。该问题仅在服务器部署环境出现，本地环境无法复现且服务器端可多次复现。本人是反应式编程新手，附上相关代码，希望明确事件丢失原因并获取更优的代码实践建议。

代码片段

@PostConstruct
List<KafkaReceiver<String, String>> kafkaReceiverList = new ArrayList<>();
for (int i = 0; i < 5; i++) {
    kafkaReceiverList.add(KafkaReceiver.create());
}

@EventListener(ApplicationStartedEvent.class)
for (KafkaReceiver<String, String> receiver : kafkaReceiverList) {
        kafkaReceivers.add(receiver
                .receive()
                .log()
                .bufferTimeout(500,10)
                .flatMap(this::processRecord) // input - List<ReceiverRecord<String, String>>
                .flatMap(this::commitRecord) // input - List<ReceiverRecord<String, String>>
                .subscribe());
    }

 public Flux<Void> commitRecord(List<ReceiverRecord<String, String>> records) {
    log.info(InfoMessageConstants.COMMIT_RECORD, records);
    records.forEach(record -> record.receiverOffset().commit().subscribe());
    return Flux.empty();
 }

@PreDestroy
kafkaReceiverList.forEach(consumer -> {
        try {
            kafkaReceivers.stream()
                    .forEach(Disposable::dispose);
        } catch (Exception ex) {
            log.error("Error closing consumer: ", ex);
        }
    });

补充说明

创建Receiver列表的原因：基于分区创建消费者，实现对消费者数量与分区的独立管控
本地环境是否可复现：否

服务器端复现步骤

停止消费者/服务
向Topic推送事件
启动消费者

事件丢失原因分析

Offset提交异步无等待：commitRecord方法中，直接对每个offset的提交调用subscribe()，属于无等待的异步操作。当方法返回Flux.empty()时，上游流会认为批次处理完成，但实际提交可能未完成。若服务器环境中提交出现延迟或异常，未提交的offset可能导致消息丢失或重复消费，也可能因提交失败未被处理，导致后续消费逻辑异常。
消费者数量与分区不匹配：创建了5个KafkaReceiver，若目标Topic的分区数少于5，多余的消费者会处于空闲状态；若分区数足够，但分区分配过程中出现异常，可能导致部分分区未被分配，对应消息无法被消费。
BufferTimeout触发时机问题：服务器环境网络延迟可能导致消息到达分散，bufferTimeout(500,10)可能在收集到5条消息后触发处理，剩余2条消息因超过500ms被单独批次处理，但如果此时流因错误终止，这部分消息会丢失。
消费者关闭不优雅：@PreDestroy中直接调用Disposable.dispose()关闭消费者，若存在未处理的消息批次（如buffer中正在收集的消息），会被直接丢弃。

优化代码实践建议

可靠处理Offset提交：将offset提交的异步操作合并，等待全部提交完成后再返回，确保提交可靠性，同时处理提交异常：

public Flux<Void> commitRecord(List<ReceiverRecord<String, String>> records) {
    log.info(InfoMessageConstants.COMMIT_RECORD, records);
    return Flux.fromIterable(records)
            .flatMap(record -> record.receiverOffset().commit()
                    .doOnError(ex -> log.error("Failed to commit offset for record: key={}, offset={}", record.key(), record.offset(), ex)))
            .then();
}

匹配消费者数量与分区数：避免创建超过Topic分区数的消费者，可通过Kafka AdminClient动态获取Topic分区数，再创建对应数量的消费者，避免资源浪费和分区分配异常。
优化流处理与错误容错：添加错误处理逻辑，避免单个批次的错误导致整个消费流终止，同时增加消息接收日志定位问题：

@EventListener(ApplicationStartedEvent.class)
public void startConsumers() {
    for (KafkaReceiver<String, String> receiver : kafkaReceiverList) {
        Disposable disposable = receiver
                .receive()
                .doOnNext(record -> log.info("Received record: key={}, value={}, offset={}", record.key(), record.value(), record.offset()))
                .log()
                .bufferTimeout(500, 10)
                .flatMap(this::processRecord)
                .flatMap(this::commitRecord)
                .onErrorContinue((ex, batch) -> log.error("Error processing batch: {}", batch, ex))
                .subscribe();
        kafkaReceivers.add(disposable);
    }
}

优雅关闭消费者：先停止订阅不再接收新消息，再关闭KafkaReceiver，确保已接收的消息处理完成：

@PreDestroy
public void stopConsumers() {
    try {
        // 停止所有订阅，不再接收新消息
        kafkaReceivers.forEach(Disposable::dispose);
        // 关闭每个KafkaReceiver
        kafkaReceiverList.forEach(receiver -> {
            try {
                receiver.close();
            } catch (Exception ex) {
                log.error("Error closing Kafka receiver: ", ex);
            }
        });
    } catch (Exception ex) {
        log.error("Error stopping consumers: ", ex);
    }
}

关键配置检查：

确认auto.offset.reset配置为earliest，确保启动后能消费未处理的历史消息；
检查所有KafkaReceiver的group.id是否一致，保证同组消费者正确分配分区；
开启Kafka客户端日志，监控offset提交状态，排查提交失败情况。

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