java 代码统计kafka topic一段时间的数据量

您可以使用Kafka提供的Java API来实现统计某个topic在一段时间内的数据量。

以下是实现的步骤：

1. 创建KafkaConsumer对象并指定要订阅的topic。

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-consumer-group");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
consumer.subscribe(Arrays.asList("test-topic"));

2. 使用poll()方法从Kafka集群获取消息。

ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(10));

3. 遍历获取到的消息，并统计数据量。

long count = 0;
for (ConsumerRecord<String, String> record : records) {
    count++;
}
System.out.println("Count: " + count);

4. 关闭KafkaConsumer对象。

consumer.close();

完整的示例代码如下：

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

public class KafkaTopicDataCount {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test-consumer-group");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        consumer.subscribe(Arrays.asList("test-topic"));
        long count = 0;
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(10));
        for (ConsumerRecord<String, String> record : records) {
            count++;
        }
        System.out.println("Count: " + count);
        consumer.close();
    }
}

注意：这段代码只能统计一段时间内的数据量，如果要统计总共的数据量，需要将该代码放在一个无限循环中，并且在每一次循环中进行累加。