当前位置：首页 > news >正文

springboot使用kafka

news 2025/6/17 11:24:05

启动kafka

确保本地已安装并启动 Kafka 服务（或连接远程 Kafka 集群），比如通过 Kafka 官网下载解压后，启动 Zookeeper（老版本 Kafka 依赖，新版本用 KRaft 可不依赖）和 Kafka 服务：

# 启动 Zookeeper（若用 KRaft 模式可跳过）

bin/zookeeper-server-start.sh config/zookeeper.properties

# 启动 Kafka 服务

bin/kafka-server-start.sh config/server.properties

版本：

Kafka 从2.8.0版本开始引入了 KIP-500，提供了无 Zookeeper 的早期访问功能1。不过，此时的实现并不完全，不建议在生产环境中使用。

3.0版本开始真正全面摒弃 Zookeeper，使用新的元数据管理方式 Kraft，提高了 Kafka 的可扩展性、可用性和性能4。

4.0版本是第一个完全无需 Apache Zookeeper 运行的重大版本，将不再支持以 ZK 模式运行或从 ZK 模式迁移。

项目引依赖

<dependencies><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>3.6.0</version> <!-- 版本按需选，建议用较新稳定版 --></dependency>
</dependencies>

创建 Producer 类（编写生产者代码）

import org.apache.kafka.clients.producer.*;
import java.util.Properties;public class KafkaProducerDemo {public static void main(String[] args) {// 1. 配置 Kafka 连接、序列化等参数Properties props = new Properties();props.put("bootstrap.servers", "localhost:9092"); // Kafka 集群地址props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // 键的序列化器props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // 值的序列化器// 2. 创建 Producer 实例Producer<String, String> producer = new KafkaProducer<>(props);// 3. 构造消息（指定主题、键、值）String topic = "test_topic"; // 要发送到的主题，需提前在 Kafka 创建或允许自动创建String key = "key1";String value = "Hello, Kafka from IDEA!";ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);// 4. 发送消息（异步发送 + 回调处理结果）producer.send(record, new Callback() {@Overridepublic void onCompletion(RecordMetadata metadata, Exception exception) {if (exception != null) {System.err.println("消息发送失败：" + exception.getMessage());} else {System.out.printf("消息发送成功！主题：%s，分区：%d，偏移量：%d%n", metadata.topic(), metadata.partition(), metadata.offset());}}});// 5. 关闭 Producer（实际生产环境可能在程序结束时或合适时机关闭）producer.close();}
}

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;public class KafkaProducerExample {private final static String TOPIC = "mytopic";private final static String BOOTSTRAP_SERVERS = "localhost:9092";public static void main(String[] args) {Properties props = new Properties();props.put("bootstrap.servers", BOOTSTRAP_SERVERS);props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");KafkaProducer<String, String> producer = new KafkaProducer<>(props);try {for (int i = 0; i < 10; i++) {String message = "Message " + i;producer.send(new ProducerRecord<>(TOPIC, message));}} catch (Exception e) {e.printStackTrace();} finally {producer.close();}}
}

创建 Consumer 类（编写消费者代码）

import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;public class KafkaConsumerDemo {public static void main(String[] args) {// 1. 配置 Kafka 连接、反序列化、消费者组等参数Properties props = new Properties();props.put("bootstrap.servers", "localhost:9092"); // Kafka 集群地址props.put("group.id", "test_group"); // 消费者组 ID，同一组内消费者协调消费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"); // 没有已提交偏移量时，从最早消息开始消费// 2. 创建 Consumer 实例KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);// 3. 订阅主题String topic = "test_topic";consumer.subscribe(Collections.singletonList(topic));// 4. 循环拉取消息（长轮询）try {while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));for (ConsumerRecord<String, String> record : records) {System.out.printf("收到消息：主题=%s，分区=%d，偏移量=%d，键=%s，值=%s%n", record.topic(), record.partition(), record.offset(), record.key(), record.value());}// 手动提交偏移量（也可配置自动提交，生产环境建议手动更可靠）consumer.commitSync();}} catch (Exception e) {e.printStackTrace();} finally {// 5. 关闭 Consumerconsumer.close();}}
}

import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Collections;
import java.util.Properties;public class KafkaConsumerExample {private final static String TOPIC = "mytopic";private final static String BOOTSTRAP_SERVERS = "localhost:9092";private final static String GROUP_ID = "mygroup";public static void main(String[] args) {Properties props = new Properties();props.put("bootstrap.servers", BOOTSTRAP_SERVERS);props.put("group.id", GROUP_ID);props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);consumer.subscribe(Collections.singletonList(TOPIC));try {while (true) {ConsumerRecords<String, String> records = consumer.poll(100);// 处理接收到的消息records.forEach(record -> {System.out.println("Received message: " + record.value());});}} catch (Exception e) {e.printStackTrace();} finally {consumer.close();}}
}

必须要素：

必要配置：
- bootstrap.servers：Kafka 集群地址。
- group.id：消费者组 ID（相同组内的消费者会负载均衡消费）。
- key.deserializer 和 value.deserializer：消息键和值的反序列化器。
- auto.offset.reset：消费位置重置策略（如 earliest 从最早消息开始消费）。
订阅主题：通过 consumer.subscribe() 订阅目标主题。
消息消费：通过 consumer.poll() 轮询拉取消息，并处理 ConsumerRecords。
偏移量管理：自动提交（enable.auto.commit=true）或手动提交（consumer.commitSync()）消费偏移量。
资源管理：使用后调用 consumer.close() 关闭连接。

与 Kafka 的对比

Kafka的Producer和Consumer需要手动管理连接和资源的关闭，因此在使用完毕后需要调用close方法来关闭Producer（或Consumer）。

总结来说，可以使用KafkaProducer的send方法来替代RabbitTemplate的convertAndSend方法在Kafka中发送消息。

Spring AMQP 是 Spring 框架提供的一个用于简化 AMQP（Advanced Message Queuing Protocol） 消息中间件开发的模块。它基于 AMQP 协议，提供了一套高层抽象和模板类，帮助开发者更便捷地实现消息发送和接收，支持多种 AMQP 消息中间件（如 RabbitMQ、Apache Qpid 等）。

维度	Spring AMQP（RabbitMQ）	Spring Kafka
协议	AMQP（高级消息队列协议）	Kafka 自研协议
消息模型	支持多种交换器类型（Direct、Topic 等）	基于主题（Topic）和分区（Partition）
顺序性	单队列内保证顺序	分区内保证顺序，多分区需按 Key 路由
吞吐量	中等（万级 TPS）	高（十万级 TPS）
适用场景	企业集成、任务调度、事务性消息	大数据、日志收集、实时流处理