Kafka消费者消费的消息数量统计

在 Kafka 消费者的代码中，可以使用一个计数变量来记录消息消费的数量，并在每次消费时将其自增。示例代码如下：

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;

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

public class KafkaConsumerExample {
    public static void main(String[] args) {
        // 消费者配置
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "test-group");
        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        // 创建消费者实例
        Consumer<String, String> consumer = new KafkaConsumer<>(props);

        // 需要订阅的主题
        String topic = "test-topic";

        // 订阅主题
        consumer.subscribe(Collections.singleton(topic));

        // 消费消息，并统计消息数量
        int count = 0;
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("Received message: key=" + record.key() + ", value=" + record.value()
                        + ", partition=" + record.partition() + ", offset=" + record.offset());
                count++;
            }
            // 输出消息数量
            System.out.println("Consumed " + count + " messages in total.");
        }
    }
}

上述代码中，在每次消费消息时，我们都会将计数变量 count 自增，并在循环外面输出总的消息数量。这样，就可以方便地统计 Kafka 消费者消费的消息数量了。