java怎么從kafka讀取數據

124次閱讀

共計 1072 個字符，預計需要花費 3 分鐘才能閱讀完成。

要從 Kafka 讀取數據，可以使用 Kafka 的 Java 客戶端庫。以下是使用 Java 從 Kafka 讀取數據的步驟：

添加 Kafka 客戶端庫的依賴：在 Maven 項目的 pom.xml 文件中，添加以下依賴：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.8.1</version>
</dependency>

創建 Kafka 消費者實例：使用 Kafka 的 Consumer 類創建一個消費者實例。需要指定 Kafka 集群的地址和端口號，以及消費者組 ID。

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);
    }
}