kafka监听topic消费java写法
时间: 2023-06-30 22:17:32 浏览: 95
在Java中使用Kafka监听Topic消费需要使用Kafka的Java客户端API。以下是一个基本的示例代码:
```java
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
public class KafkaConsumerExample {
public static void main(String[] args) {
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singleton("test-topic"));
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
records.forEach(record -> {
System.out.println("Received message: " + record.value());
});
}
}
}
```
在这个示例代码中,我们首先创建了一个KafkaConsumer对象,并设置了一些属性,例如Kafka服务器的地址、消费者组ID、反序列化类等。然后,我们通过subscribe()方法订阅了一个Topic。最后,我们使用poll()方法来获取消息,对每个消息进行处理。
需要注意的是,这个示例代码只是一个基本的示例,实际上,在生产环境中,我们需要考虑更多的因素,例如消息的消费策略、消息的重复消费等。
阅读全文