消费kafka topic数据

消费Kafka Topic数据可以使用Kafka Consumer API。下面是一个简单的Java代码示例：

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

public class KafkaConsumerExample {
   public static void main(String[] args) throws Exception {
      String topicName = "test";
      Properties props = new Properties();
      props.put("bootstrap.servers", "localhost:9092");
      props.put("group.id", "test-group");
      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<>(props);
      consumer.subscribe(Collections.singletonList(topicName));
      while (true) {
         ConsumerRecords<String, String> records = consumer.poll(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的消费者，然后订阅该Topic。接下来，使用while循环从Topic中轮询消息，然后遍历消息并打印出它们的偏移量、键和值。

注意：上述示例仅用于演示目的。在实际应用中，应该处理消费者异常，设置适当的配置参数，以及根据需要进行参数调整。