当前位置：首页 > 文章列表 > 文章 > java教程 > Java操作Kafka：生产者消费者实战教程

Java操作Kafka：生产者消费者实战教程

2025-08-08 16:09:29 0浏览收藏

想要掌握Java操作Kafka？本文将通过实例教程，带你快速上手Kafka消息队列的生产者和消费者开发。首先，你需要引入kafka-clients Maven依赖。然后，我们将详细讲解如何编写Kafka生产者，配置BOOTSTRAP_SERVERS_CONFIG、序列化类，并通过send方法发送消息到指定Topic。紧接着，我们将介绍Kafka消费者的编写，包括配置组ID、反序列化类，以及利用poll方法循环处理消息。此外，本文还强调了启动顺序的重要性，建议先启动Kafka服务，再运行生产者和消费者。同时，需要注意Topic的一致性和网络环境的连通性。掌握这些关键步骤，你就能轻松实现Java与Kafka的集成，进行高效的消息生产和消费。

Java操作Kafka的核心在于配置生产者和消费者并调用对应方法。1. 添加Maven依赖引入kafka-clients库；2. 编写生产者设置BOOTSTRAP_SERVERS_CONFIG、序列化类并调用send发送ProducerRecord；3. 编写消费者配置组ID、反序列化类并循环调用poll处理ConsumerRecords；4. 注意启动顺序为先运行Kafka服务再执行生产者消费者同时检查topic一致性与网络环境。通过以上步骤即可实现基础的消息生产和消费流程。

Java如何操作Kafka？生产者消费者实例

Java操作Kafka其实不难，核心就是两个角色：生产者（Producer）和消费者（Consumer）。只要理解了基本流程和API用法，就能快速上手。下面通过一个简单的实例，带你了解如何用Java写一个Kafka的生产者和消费者。

1. 添加Maven依赖

在开始写代码之前，先确保你的项目里有Kafka的客户端依赖。如果你用的是Maven，可以在pom.xml中添加：

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

版本号可以根据你使用的Kafka版本适当调整。一般来说，保持客户端版本和Kafka服务端版本一致或接近比较稳妥。

2. 编写Kafka生产者（Producer）

生产者的任务是把消息发到Kafka的某个Topic里。下面是一个简单的例子：

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;

public class KafkaProducerExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        Producer<String, String> producer = new KafkaProducer<>(props);

        ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "Hello Kafka!");

        producer.send(record);
        producer.close();
    }
}

关键点说明：

BOOTSTRAP_SERVERS_CONFIG 是Kafka服务器的地址，通常是host:port形式。
StringSerializer 是序列化类，Kafka传输的是字节，所以需要把字符串转成字节。
ProducerRecord 是要发送的消息对象，构造时传入topic和value，也可以加key。

3. 编写Kafka消费者（Consumer）

消费者从指定的Topic中拉取消息，并进行处理。下面是一个基本的消费者示例：

import org.apache.kafka.clients.consumer.*;
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());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        Consumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("my-topic"));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("Received: %s%n", record.value());
            }
        }
    }
}

关键点说明：