使用Java连接Kafka需引入kafka-clients依赖，配置生产者序列化参数并发送消息，消费者设置组ID和反序列化器后订阅Topic拉取消息，通过poll循环处理，确保网络通畅与序列化匹配即可实现通信。
用Java连接Kafka发送和消费消息，核心是使用Kafka客户端库，通过配置生产者和消费者来完成通信。整个过程不复杂，只要把依赖、配置和代码逻辑写对，很快就能跑通。
添加Maven依赖
项目基于Maven构建的话，先在pom.xml中引入Kafka客户端依赖。推荐使用官方提供的kafka-clients，通用性强，不绑定特定框架。
以下是常用依赖配置：
org.apache.kafka kafka-clients 3.4.0
如果你用的是Spring Boot，也可以直接引入spring-kafka，它封装了底层操作，使用更方便。
立即学习“Java免费学习笔记（深入）”；
编写Kafka生产者发送消息
生产者负责把数据发到指定的Topic。创建KafkaProducer对象时，需要设置必要的参数。
基础步骤如下：
- 定义Properties配置，包括Kafka服务器地址（bootstrap.servers）、key和value的序列化器（如StringSerializer）。
- 创建KafkaProducer
实例。 - 构造ProducerRecord，指定Topic名称、可选的Key和消息内容。
- 调用producer.send()发送消息，可以加回调处理发送结果。
- 程序结束前调用producer.close()释放资源。
示例代码片段：
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
try (KafkaProducer producer = new KafkaProducer<>(props)) {
ProducerRecord record = new ProducerRecord<>("test-topic", "key1", "Hello Kafka");
producer.send(record, (metadata, exception) -> {
if (exception == null) {
System.out.printf("发送成功，分区：%d，偏移量：%d%n", metadata.partition(), metadata.offset());
} else {
exception.printStackTrace();
}
});
}
编写Kafka消费者接收消息
消费者从Topic订阅并拉取消息，需要配置组ID（group.id），以便Kafka管理消费进度（offset）。
基本流程是：
- 配置Properties，除了服务器地址和反序列化器，还要设置group.id和auto.offset.reset（如earliest从头开始，latest从最新开始）。
- 创建KafkaConsumer实例。
- 调用consumer.subscribe()订阅一个或多个Topic。
- 在循环中调用consumer.poll()获取消息，然后遍历处理。
- 处理完后提交offset（自动或手动），最后关闭消费者。
简单消费示例：
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("auto.offset.reset", "earliest");
try (KafkaConsumer consumer = new KafkaConsumer<>(props)) {
consumer.subscribe(Arrays.asList("test-topic"));
while (true) {
ConsumerRecords records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord 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());
}
}
}
基本上就这些。本地启动Kafka服务，创建好Topic，再运行生产和消费代码，就能看到消息通了。调试时注意看日志，网络和序列化问题最常见。
