kafka监听topic消费java写法
时间: 2023-09-18 14:05:33 浏览: 216
### 回答1:
以下是使用Java编写Kafka消费者监听Topic的示例代码:
```java
import java.util.Collections;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
public class KafkaConsumerExample {
public static void main(String[] args) {
Properties props = new Properties();
props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "test-group");
props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("test-topic"));
while (true) {
ConsumerRecords<String, String> records = consumer.poll(1000);
for (ConsumerRecord<String, String> record : records) {
System.out.println("Received message: " + record.value());
}
}
}
}
```
在上面的代码中,我们使用了KafkaConsumer类来创建一个Kafka消费者,并订阅了一个名为"test-topic"的Topic。然后,我们使用一个无限循环来轮询Kafka集群,以便在有新消息时能够及时处理。在每次轮询时,我们使用poll方法从Kafka集群中获取消息,并对每个消息进行处理。在这个示例中,我们只是简单地将消息内容打印到控制台。
### 回答2:
Kafka 是一种分布式流处理平台,它可以用于构建高可靠性、可扩展性和可容错性的实时数据流处理系统。使用 Java 编写 Kafka 监听和消费 Topic 的代码如下:
首先,我们需要导入相关的 Kafka 相关包:
```java
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
```
然后,我们可以创建 Kafka 消费者并设置相关属性:
```java
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); // 设置 Kafka 服务器地址
props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group"); // 设置消费者组 ID
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<String, String>(props);
```
接下来,我们可以订阅要监听的 Topic:
```java
consumer.subscribe(Arrays.asList("my-topic")); // 订阅主题,可以是单个主题或者多个主题
```
最后,我们可以循环监听并消费消息:
```java
while (true) {
ConsumerRecords<String, String> records = consumer.poll(100); // 从 Kafka 中获取消息
for (ConsumerRecord<String, String> record : records) {
System.out.printf("Topic: %s, Partition: %d, Offset: %d, Key: %s, Value: %s\n",
record.topic(), record.partition(), record.offset(), record.key(), record.value());
// 在这里进行自定义的消息处理逻辑
}
}
```
以上就是使用 Java 编写 Kafka 监听和消费 Topic 的基本代码。根据实际需求,你可以进一步处理消费的消息,例如将其保存到数据库、进行计算等等。同时,请确保你已经正确配置了 Kafka 的相关参数,包括 Kafka 服务器地址、消费者组 ID 等。
### 回答3:
Kafka是一个分布式的消息队列系统,允许多个消费者同时监听同一个主题(topic)。以下是使用Java编写的Kafka监听topic消费的常见写法:
首先,需要引入Kafka的相关依赖:
```
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.8.0</version>
</dependency>
```
接下来,创建一个Kafka消费者对象,并设置相关属性:
```
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092"); // Kafka集群的地址
props.put("group.id", "my-consumer-group"); // 消费者组ID
props.put("enable.auto.commit", "true"); // 自动提交消费位移
props.put("auto.commit.interval.ms", "1000"); // 消费位移提交间隔时间
props.put("auto.offset.reset", "earliest"); // 从最早的偏移量开始消费
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); // key的反序列化器
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); // value的反序列化器
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
```
然后,订阅要监听的主题(topic):
```
consumer.subscribe(Arrays.asList("my-topic")); // 订阅单个主题
// 或者
consumer.subscribe(Pattern.compile("my-topic.*")); // 订阅多个主题(使用正则表达式匹配)
```
最后,开始消费消息:
```
try {
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100)); // 每100毫秒拉取一次消息
for (ConsumerRecord<String, String> record : records) {
String key = record.key();
String value = record.value();
long offset = record.offset();
int partition = record.partition();
System.out.println("Received message: key = " + key + ", value = " + value + ", offset = " + offset + ", partition = " + partition);
// 处理消息的逻辑代码
}
}
} finally {
consumer.close(); // 关闭消费者
}
```
以上就是使用Java编写的Kafka监听topic消费的基本写法。其中,我们通过创建一个Kafka消费者对象,设置相关属性,订阅要监听的主题,然后在一个无限循环中通过`poll`方法拉取消息进行消费。最后,记得在消费完成后关闭消费者。
阅读全文