要從Kafka讀取數據,可以使用Kafka的Java客戶端庫。以下是使用Java從Kafka讀取數據的步驟:
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.8.1</version>
</dependency>
Properties properties = new Properties();
properties.put("bootstrap.servers", "localhost:9092"); // Kafka集群的地址和端口號
properties.put("group.id", "my-group"); // 消費者組ID
Consumer<String, String> consumer = new KafkaConsumer<>(properties);
subscribe()
方法訂閱一個或多個Kafka主題。consumer.subscribe(Arrays.asList("my-topic")); // 訂閱名為"my-topic"的主題
poll()
方法從Kafka讀取消息。默認情況下,poll()
方法會阻塞,直到有消息可用。while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
String key = record.key();
String value = record.value();
// 處理消息
System.out.println("Received message: key = " + key + ", value = " + value);
}
}
close()
方法來關閉它。consumer.close();
這樣就可以使用Java從Kafka讀取數據了。請注意,以上代碼僅提供了基本的示例,實際應用中可能需要根據具體需求進行適當的修改和調整。