Kafka消费指定时间

Kafka的消费是基于分区（partition）的，每个分区中的消息都是按照顺序进行消费的。消费者可以通过指定偏移量（offset）来订阅分区中的消息。Kafka提供了一些API来获取偏移量，从而可以实现消费指定时间的消息。

偏移量是与时间戳相关的，Kafka中有两种时间戳：消息创建时间（timestamp）和消息提交时间（commit timestamp）。其中，消息创建时间是由生产者在发送消息时设置的时间戳，而消息提交时间是由消费者在将消息标记为已消费时设置的时间戳。

如果要消费指定时间之后的消息，可以使用`consumer.seek()`方法将消费者的偏移量设置为指定时间之后的偏移量。具体步骤如下：

1. 通过`consumer.offsetsForTimes()`方法获取指定时间之后的偏移量，该方法需要传入一个Map对象，其中包含各个分区的时间戳。
2. 通过`consumer.seek()`方法将消费者的偏移量设置为指定时间之后的偏移量。

以下是示例代码：

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.*;

public class KafkaConsumerExample {

    public static void main(String[] args) {

        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test-group");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        String topic = "test-topic";

        // 获取分区列表
        List<PartitionInfo> partitions = consumer.partitionsFor(topic);

        // 构建时间戳Map，key为分区，value为指定时间的毫秒数
        Map<TopicPartition, Long> timestampsToSearch = new HashMap<>();
        for (PartitionInfo partition : partitions) {
            timestampsToSearch.put(new TopicPartition(topic, partition.partition()), System.currentTimeMillis() - 3600000L); // 指定1小时前的时间戳
        }

        // 获取指定时间之后的偏移量
        Map<TopicPartition, OffsetAndTimestamp> offsetAndTimestampMap = consumer.offsetsForTimes(timestampsToSearch);
        for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : offsetAndTimestampMap.entrySet()) {
            TopicPartition partition = entry.getKey();
            OffsetAndTimestamp offsetAndTimestamp = entry.getValue();
            if (offsetAndTimestamp != null) {
                long offset = offsetAndTimestamp.offset();
                consumer.seek(partition, offset);
                System.out.println("Seek to offset " + offset + " for partition " + partition);
            }
        }

        // 订阅分区并消费消息
        consumer.subscribe(Collections.singletonList(topic));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
        }
    }
}

上述代码会订阅`test-topic`主题并消费指定时间之后的消息。在示例中，我们设置了时间戳为1小时前的时间。你可以根据自己的需求来设置时间戳。