如何使用Reactor Kafka消费特定时间范围内的消息

你当前用takeWhile的写法会在任意一条消息超过结束时间时立刻终止整个消费流，这必然会导致其他分区中未处理的符合时间条件的消息被遗漏。要解决这个问题，需要针对每个分区单独跟踪处理状态，只有当所有分区都已处理完符合时间范围的消息后，才停止消费。

具体实现思路 #

1. 维护线程安全的集合，记录已分配的所有分区，以及每个分区是否已完成处理（即该分区已出现超过结束时间的消息，或已消费到分区末尾）
2. 消费消息时做分层判断：
   • 若当前分区已标记完成，直接过滤掉后续消息
   • 若消息时间戳未超过结束时间，正常处理消息
   • 若消息时间戳超过结束时间，标记对应分区为已完成，不再处理该分区后续消息
3. 每次处理后，检查所有已分配分区是否都已完成，仅当全部完成时终止消费流

示例代码 #

import org.apache.kafka.common.TopicPartition;
import reactor.core.publisher.Flux;
import reactor.kafka.receiver.KafkaReceiver;
import reactor.kafka.receiver.ReceiverOptions;
import reactor.kafka.receiver.ReceiverRecord;

import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.time.Duration;

public class TimeRangeKafkaConsumer {

    public void consumeInTimeRange(ReceiverOptions<ByteBuffer, ByteBuffer> receiverOptions,
                                  long startTimeInMillis,
                                  long endTimeInMillis) {

        // 记录当前已分配的所有分区
        Set<TopicPartition> assignedPartitions = Collections.synchronizedSet(new HashSet<>());
        // 记录每个分区的处理完成状态
        ConcurrentMap<TopicPartition, Boolean> partitionDoneMap = new ConcurrentHashMap<>();

        ReceiverOptions<ByteBuffer, ByteBuffer> adjustedOptions = receiverOptions
                .subscription(topicConfig.getTopics())
                .pollTimeout(Duration.ofMillis(topicConfig.getPollWaitTimeoutMs()))
                .addAssignListener(partitions -> {
                    partitions.forEach(partition -> {
                        partition.seekToTimestamp(startTimeInMillis);
                        assignedPartitions.add(partition);
                        partitionDoneMap.put(partition, false);
                    });
                })
                .addRevokeListener(partitions -> {
                    partitions.forEach(partition -> {
                        assignedPartitions.remove(partition);
                        partitionDoneMap.remove(partition);
                    });
                });

        KafkaReceiver.create(adjustedOptions)
                .receive()
                .filter(record -> {
                    TopicPartition tp = new TopicPartition(record.topic(), record.partition());
                    // 分区已完成则过滤后续消息
                    if (partitionDoneMap.getOrDefault(tp, false)) {
                        return false;
                    }
                    // 消息超过结束时间，标记分区完成并过滤
                    if (record.timestamp() >= endTimeInMillis) {
                        partitionDoneMap.put(tp, true);
                        return false;
                    }
                    // 符合时间条件，保留消息
                    return true;
                })
                .doOnNext(this::handleConsumerRecord)
                .takeWhile(__ -> {
                    // 只要有一个分区未完成，就继续消费
                    return assignedPartitions.stream()
                            .anyMatch(tp -> !partitionDoneMap.getOrDefault(tp, true));
                })
                .subscribe();
    }

    private void handleConsumerRecord(ReceiverRecord<ByteBuffer, ByteBuffer> record) {
        // 你的消息处理逻辑
    }
}

关键细节说明 #

• 线程安全集合：用ConcurrentHashMap和同步HashSet处理多线程消费场景下的分区状态更新
• 重平衡处理：通过addAssignListener和addRevokeListener处理分区重平衡，及时更新分区集合与状态
• 过滤逻辑：优先判断分区状态，避免无效消息的处理开销；对超时消息仅标记状态不处理
• 终止条件：takeWhile中检查所有已分配分区的完成状态，确保所有符合条件的消息都被消费后才停止

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