java实现消费kafka消息

要使用Java实现消费Kafka消息，你可以按照以下步骤进行操作：

1. 添加Kafka依赖：在你的项目中添加Kafka客户端的依赖。你可以在`pom.xml`文件中添加以下依赖：

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

2. 创建Kafka消费者配置：创建一个`Properties`对象来配置Kafka消费者的属性，例如Kafka集群的地址、消费者组ID等。

import java.util.Properties;

public class KafkaConsumerConfig {

    public static Properties getConsumerProperties() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "<kafka服务器地址>");
        props.put("group.id", "<消费者组ID>");
        // 其他配置属性...
        return props;
    }
}

3. 创建消息消费者：使用上述配置创建一个Kafka消费者实例，并指定要订阅的主题。

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Collections;

public class KafkaMessageConsumer {

    public static void main(String[] args) {
        // 获取消费者配置
        Properties props = KafkaConsumerConfig.getConsumerProperties();

        // 创建Kafka消费者实例
        Consumer<String, String> consumer = new KafkaConsumer<>(props);

        // 订阅主题
        consumer.subscribe(Collections.singletonList("<topic名称>"));

        // 消费消息
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            // 处理接收到的消息
            records.forEach(record -> {
                System.out.println("Received message: " + record.value());
                // 在这里编写具体的业务逻辑
            });
        }
    }
}

在上述代码中，我们创建了一个Kafka消费者实例，并订阅了指定的主题。然后，我们使用一个无限循环来持续地消费消息。每次调用`poll`方法从Kafka服务器拉取一批消息，然后遍历这些消息并进行处理。

注意：这只是一个简单的示例来展示如何使用Java消费Kafka消息。在实际的应用中，你可能需要考虑到更多的配置和异常处理。