阅读量:9
在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中拉取数据。在这个例子中，我们只是简单地将接收到的消息打印到控制台上。
请确保替换bootstrapServers、groupId和topic为你要连接的Kafka集群的地址、你的消费者组的ID和你要消费的主题。
希望对你有帮助！
