java怎么从kafka中读取数据

42次阅读
没有评论

共计 1528 个字符,预计需要花费 4 分钟才能阅读完成。

在 Java 中,你可以使用 Apache Kafka 的 Java 客户端库来从 Kafka 中读取数据。下面是一个简单的示例代码:

首先,你需要在你的项目中添加 Kafka 的 Java 客户端库的依赖。你可以在你的构建工具(如 Maven 或 Gradle)的配置文件中添加以下依赖:

<!-- Kafka client -->
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.8.0</version>
</dependency>

然后,你可以使用以下代码从 Kafka 中读取数据:

import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Collections;
import java.util.Properties;

public class KafkaConsumerExample {public static void main(String[] args) {// Kafka 集群的地址
        String bootstrapServers = "localhost:9092";
        // 消费者组的 ID
        String groupId = "my-group";
        // 要消费的主题
        String topic = "my-topic";

        // 配置消费者的属性
        Properties properties = new Properties();
        properties.put("bootstrap.servers", bootstrapServers);
        properties.put("group.id", groupId);
        properties.put("key.deserializer", StringDeserializer.class);
        properties.put("value.deserializer", StringDeserializer.class);

        // 创建消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);

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

        // 无限循环从 Kafka 中读取消息
        while (true) {ConsumerRecords<String, String> records = consumer.poll(1000);
            for (ConsumerRecord<String, String> record : records) {System.out.println("Received message: " + record.value());
            }
        }
    }
}

以上代码创建了一个消费者并订阅了一个主题。然后,通过调用 consumer.poll(1000) 来从 Kafka 中拉取数据。在这个例子中,我们只是简单地将接收到的消息打印到控制台上。

请确保替换 bootstrapServersgroupIdtopic为你要连接的 Kafka 集群的地址、你的消费者组的 ID 和你要消费的主题。

希望对你有帮助!

丸趣 TV 网 – 提供最优质的资源集合!

正文完
 
丸趣
版权声明:本站原创文章,由 丸趣 2023-12-12发表,共计1528字。
转载说明:除特殊说明外本站除技术相关以外文章皆由网络搜集发布,转载请注明出处。
评论(没有评论)