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

Spring Boot + Kafka 实战：从入门到避坑，小白也能轻松上手！

news 2026/3/26 20:30:24

视频看了几百小时还迷糊？关注我，几分钟让你秒懂！

一、为什么我们需要 Kafka？

在现代微服务架构中，系统之间的通信不能总是“你等我、我等你”——这会导致性能瓶颈甚至雪崩。Kafka 就是一个高性能、高吞吐、可扩展的消息中间件，它能帮我们实现：

解耦：生产者和消费者互不干扰。
削峰填谷：流量高峰时缓存消息，避免系统崩溃。
异步处理：比如用户注册后发邮件、记录日志等非核心操作可以异步完成。

🎯 典型应用场景

用户行为日志收集（如点击、浏览）
订单系统异步通知库存、物流
微服务间事件驱动通信

二、Spring Boot 集成 Kafka 快速上手

✅ 步骤 1：添加依赖（`pom.xml`）

<dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency>

注意：Spring Boot 2.7+ 默认支持 Kafka 3.x，无需额外指定版本。

✅ 步骤 2：配置 Kafka（`application.yml`）

spring: kafka: bootstrap-servers: localhost:9092 producer: key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.springframework.kafka.support.serializer.JsonSerializer consumer: group-id: my-group key-deserializer: org.apache.kafka.commonserialization.StringDeserializer value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer properties: spring.json.trusted.packages: "*"

⚠️spring.json.trusted.packages是安全设置，防止反序列化攻击。生产环境应指定具体包名，如com.example.dto。

✅ 步骤 3：定义消息实体类

public class UserEvent { private String userId; private String action; // 如 "REGISTER", "LOGIN" // 构造函数、getter/setter 省略（建议用 Lombok） }

✅ 步骤 4：生产者（发送消息）

@Service public class KafkaProducerService { @Autowired private KafkaTemplate<String, UserEvent> kafkaTemplate; public void sendUserEvent(String topic, UserEvent event) { kafkaTemplate.send(topic, event.getUserId(), event); System.out.println(">>> 消息已发送: " + event); } }

✅ 步骤 5：消费者（接收消息）

@Component public class KafkaConsumerService { @KafkaListener(topics = "user-events", groupId = "my-group") public void listen(UserEvent event) { System.out.println("<<< 收到消息: " + event); // 模拟业务处理：发邮件、写日志等 } }

✅ 步骤 6：Controller 测试接口

@RestController @RequestMapping("/kafka") public class KafkaTestController { @Autowired private KafkaProducerService producerService; @PostMapping("/send") public String send(@RequestParam String userId, @RequestParam String action) { UserEvent event = new UserEvent(); event.setUserId(userId); event.setAction(action); producerService.sendUserEvent("user-events", event); return "消息已发送！"; } }

启动 Kafka（本地可用 Docker）：

docker run -p 9092:9092 --name kafka \ -e KAFKA_BROKER_ID=0 \ -e KAFKA_ZOOKEEPER_CONNECT=localhost:2181 \ -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 \ -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \ bitnami/kafka:latest

如果没有 Zookeeper，可使用 Kafka 3.3+ 的 KRaft 模式（无 Zookeeper），但初学者建议先用传统方式。

三、反例 & 常见错误（避坑指南）

❌ 反例 1：忽略消费者组（group.id）

// 错误：没指定 groupId，多个实例会重复消费！ @KafkaListener(topics = "user-events") public void badListener(UserEvent event) { ... }

✅ 正确做法：明确指定groupId，同一组内消息只会被一个消费者处理。

❌ 反例 2：序列化配置错误

# 错误：value-serializer 写成 StringSerializer，但传的是对象 spring: kafka: producer: value-serializer: org.apache.kafka.common.serialization.StringSerializer

结果：ClassCastException或SerializationException。

✅ 正确：对象用JsonSerializer，字符串用StringSerializer。

❌ 反例 3：消费者处理异常未捕获

@KafkaListener(topics = "user-events") public void riskyListener(UserEvent event) { // 如果这里抛异常，消息会不断重试（默认无限重试！） someRiskyOperation(); }

✅ 解决方案：加 try-catch，或配置死信队列（DLQ）：

@Bean public ConsumerFactory<String, UserEvent> consumerFactory() { Map<String, Object> props = new HashMap<>(); // ... 其他配置 props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class); props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class); return new DefaultKafkaConsumerFactory<>(props); }

更推荐：使用@RetryableTopic（Spring Kafka 2.7+）自动重试并转发到 DLQ。