java kafka消费示例

Kafka是一个高性能、分布式的消息队列，常常被用来处理大量的实时数据流。为了能够从Kafka中消费数据，需要编写相应的消费者代码。

在Java中，使用Kafka消费数据很简单。首先，需要引入Kafka的Java客户端依赖包，在Maven中可以添加以下依赖：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.8.0</version>
    <scope>compile</scope>
</dependency>

接着，创建一个KafkaConsumer对象，并指定所需的配置信息，如Kafka集群的地址、消费组ID等。然后，调用consumer.subscribe()方法来订阅需要消费的主题。

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

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

最后，通过循环调用consumer.poll()方法来获取消息并处理。该方法会返回一个ConsumerRecords对象，其中包含了所有的消费记录。可以遍历这些记录并按需进行处理，如将其打印到控制台、写入文件等。

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("topic=%s, partition=%d, offset=%d, key=%s, value=%s\n",
            record.topic(), record.partition(), record.offset(), record.key(), record.value());
    }
}

除了基本的消费方式外，Kafka还提供了一些高级的消费API，如批量消费、手动提交偏移量等。通过灵活使用这些API，可以实现更加高效、可靠的数据消费。