java 代码统计kafka topic 每小时的偏移量

可以使用 Kafka 的 Java 客户端 API 来实现统计 Kafka Topic 每小时的偏移量，具体的实现步骤如下：

1. 使用 Kafka 的 Java 客户端 API 创建一个 KafkaConsumer 实例，配置相关的属性，如 Kafka 集群的地址、消费者组、序列化和反序列化器等。

2. 使用 KafkaConsumer 的 assign 方法手动分配 TopicPartition，获取每个 TopicPartition 的当前偏移量（offset）。

3. 使用 KafkaConsumer 的 seek 方法将每个 TopicPartition 的偏移量定位到上一个小时的开始位置，即当前时间的前一个小时。

4. 启动一个定时任务，每小时执行一次，将每个 TopicPartition 的偏移量统计出来，输出到日志或其他存储介质中。

下面是一个简单的示例代码：

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class KafkaOffsetStatistics {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka1:9092,kafka2:9092,kafka3:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "offset-statistics");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("test-topic"));

        for (TopicPartition partition : consumer.assignment()) {
            long currentOffset = consumer.position(partition);
            consumer.seek(partition, getStartOffset(currentOffset));
        }

        while (true) {
            for (ConsumerRecord<String, String> record : consumer.poll(Duration.ofSeconds(1))) {
                // 处理消息
            }
        }
    }

    private static long getStartOffset(long currentOffset) {
        long now = System.currentTimeMillis();
        long lastHour = now - 3600 * 1000;
        long startOffset = currentOffset;
        while (startOffset >= 0) {
            consumer.seek(partition, startOffset);
            ConsumerRecord<String, String> record = consumer.poll(Duration.ofMillis(100)).iterator().next();
            long timestamp = record.timestamp();
            if (timestamp < lastHour) {
                return startOffset + 1;
            }
            startOffset--;
        }
        return 0;
    }

    private static void statisticsOffset(KafkaConsumer<String, String> consumer) {
        for (TopicPartition partition : consumer.assignment()) {
            long currentOffset = consumer.position(partition);
            long startOffset = getStartOffset(currentOffset);
            long hourOffset = currentOffset - startOffset;
            System.out.println(String.format("Topic %s, Partition %d, Hourly Offset %d", partition.topic(), partition.partition(), hourOffset));
        }
    }
}

这段代码会启动一个 KafkaConsumer 实例，订阅一个主题（test-topic），然后手动分配每个 TopicPartition 的偏移量，并将其定位到上一个小时的开始位置。然后启动一个循环，每秒钟轮询一次 Kafka 集群获取消息，处理完消息后调用 statisticsOffset 方法统计每个 TopicPartition 的偏移量。getStartOffset 方法实现了根据当前偏移量计算上一个小时的开始位置的逻辑。