java操作kafka的核心在于配置生产者和消费者并调用对应方法。1. 添加maven依赖引入kafka-clients库；2. 编写生产者设置bootstrap_servers_config、序列化类并调用send发送producerrecord；3. 编写消费者配置组id、反序列化类并循环调用poll处理consumerrecords；4. 注意启动顺序为先运行kafka服务再执行生产者消费者同时检查topic一致性与网络环境。通过以上步骤即可实现基础的消息生产和消费流程。
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服务端版本一致或接近比较稳妥。
立即学习“Java免费学习笔记（深入）”；
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());
}
}
}
}关键点说明：
-
GROUP_ID_CONFIG是消费者的组ID，同一个组里的消费者会分摊消费任务。 -
AUTO_OFFSET_RESET_CONFIG设置了当没有初始偏移量或偏移量无效时的行为，earliest表示从头开始消费。 -
poll()方法会拉取一批消息，然后进行遍历处理。 - 消费者是持续运行的，所以一般放在一个循环里。
4. 启动顺序和注意事项
- 先启动Kafka服务（ZooKeeper + Kafka Server）
- 再运行生产者和消费者
- Topic可以提前创建，也可以让Kafka自动创建（不推荐用于生产环境）
常见问题：
- 消费者收不到消息？检查topic名是否一致、是否在同一个网络环境。
- 消费者重复消费？可能是offset提交失败，或者没有开启自动提交。
- 生产者发送失败？查看Kafka服务器是否启动、网络是否通。
基本上就这些。Java操作Kafka的核心就是配置好Producer和Consumer的参数，然后调用send和poll方法。虽然看起来简单，但实际使用中需要注意很多细节，比如序列化、分区策略、消费者组管理等。不过对于入门来说，上面的例子已经够用了。
