Reactor Kafka 中 Out Of Order commits 机制的实现与使用方案

Reactor-Kafka本身没有内置和Spring-Kafka OOO Commits完全对齐的开箱功能，但可以通过手动控制偏移量提交+分区级偏移量追踪的方式，实现「同一分区内所有前置消息处理完成后，才提交当前消息偏移量」的需求，避免未处理消息的偏移量被提前提交。

核心实现逻辑 #

• 关闭自动偏移量提交：配置enable.auto.commit = false，所有偏移量提交逻辑完全由业务侧管控
• 按分区维护偏移量状态：每个分区单独维护有序的偏移量追踪集合，记录已拉取但未处理完成的消息状态
• 处理状态标记：单条消息业务逻辑处理完成后，在对应分区的状态集合中标记该偏移量已完成
• 可提交偏移量计算：每次标记完成后从分区最小偏移量开始连续检查，找到最长的连续已完成偏移量序列，取该序列的最大偏移量+1作为该分区的待提交偏移量
• 批量提交偏移量：按固定条数或固定时间间隔触发偏移量提交，降低提交操作的性能损耗

核心代码示例 #

基础消费与偏移量控制逻辑 #

// 消费者基础配置
Map<String, Object> consumerProps = new HashMap<>();
consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "your-consumer-group");
consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "your-kafka-brokers");

ReceiverOptions<String, String> receiverOptions = ReceiverOptions.create(consumerProps)
        .subscription(Collections.singleton("your-business-topic"));

// 按分区存储偏移量追踪状态，保证线程安全
ConcurrentHashMap<TopicPartition, OffsetTrackingState> partitionStateMap = new ConcurrentHashMap<>();

// 消费流处理逻辑
KafkaReceiver.create(receiverOptions)
        .receive()
        // 异步处理业务逻辑，concurrency建议不超过消费的分区总数，避免同分区消息乱序
        .flatMap(record -> {
            TopicPartition tp = new TopicPartition(record.topic(), record.partition());
            partitionStateMap.computeIfAbsent(tp, k -> new OffsetTrackingState());
            // 自定义业务处理逻辑，处理完成返回消息元数据
            return handleBusiness(record)
                    .thenReturn(new RecordMeta(tp, record.offset()))
                    .onErrorResume(e -> {
                        // 业务处理失败逻辑：重试、投递死信队列等，处理完成后再标记偏移量，避免阻塞提交
                        return handleFailure(record, e).thenReturn(new RecordMeta(tp, record.offset()));
                    });
        }, Runtime.getRuntime().availableProcessors())
        // 按分区分组处理偏移量提交
        .groupBy(RecordMeta::topicPartition)
        .flatMap(groupFlux -> groupFlux
                .doOnNext(meta -> partitionStateMap.get(meta.topicPartition()).markCompleted(meta.offset()))
                // 每10条或每3秒触发一次偏移量提交，可根据业务调整
                .windowTimeout(10, Duration.ofSeconds(3))
                .flatMap(window -> window.reduce((prev, curr) -> curr)
                        .flatMap(lastMeta -> {
                            OffsetTrackingState state = partitionStateMap.get(lastMeta.topicPartition());
                            long commitOffset = state.getCommitOffset();
                            // 存在可提交的新偏移量时执行提交
                            if (commitOffset > 0) {
                                return groupFlux.key().offset(commitOffset).commit();
                            }
                            return Mono.empty();
                        })
                )
        )
        .subscribe();

偏移量状态追踪实现 #

class OffsetTrackingState {
    // 有序存储偏移量状态：key=偏移量，value=是否处理完成
    private final TreeMap<Long, Boolean> offsetStatus = new TreeMap<>();
    private long lastCommitted = -1;

    // 标记偏移量处理完成
    public synchronized void markCompleted(long offset) {
        offsetStatus.put(offset, true);
    }

    // 计算当前可提交的最大连续偏移量
    public synchronized long getCommitOffset() {
        long currentCheck = lastCommitted + 1;
        // 找到最长的连续已完成偏移量序列
        while (offsetStatus.containsKey(currentCheck) && offsetStatus.get(currentCheck)) {
            currentCheck++;
        }
        // 清理已提交的偏移量记录，避免内存溢出
        offsetStatus.headMap(currentCheck).clear();
        long maxCompleted = currentCheck - 1;
        if (maxCompleted > lastCommitted) {
            lastCommitted = maxCompleted;
            // Kafka提交的偏移量是下一条待拉取消息的位置，所以返回值+1
            return maxCompleted + 1;
        }
        // 没有新的可提交偏移量
        return -1;
    }
}

// 消息元数据封装类
record RecordMeta(TopicPartition topicPartition, long offset) {}

注意事项 #

• 消息处理失败后不要直接标记偏移量完成，建议配置有限次数的重试，重试失败的消息投递到死信队列后再标记偏移量完成，避免阻塞整个分区的偏移量提交
• 同分区的状态操作要保证线程安全，示例中用synchronized实现，也可以根据业务场景替换为原子类、读写锁等更高效的线程安全方案
• 偏移量提交频率可根据业务对重复消费的容忍度调整：提交频率越高，服务重启后重复消费的消息越少，但Kafka集群的提交请求压力也越大

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