Spring Boot 集成 Kafka 实战指南
1. 概述
Apache Kafka 是一个分布式流处理平台，以高吞吐量、数据持久化和容错性著称，广泛应用于实时数据管道构建。
Spring Boot 作为主流的微服务开发框架，提供了开箱即用的集成能力，让 Kafka 的接入变得简单。将 Spring Boot 与 Kafka 组合，能带来以下好处：
- 异步处理：利用消息队列实现生产者和消费者解耦，提升系统响应速度。
- 系统解耦：模块间通过消息通信，降低直接依赖，增强可维护性。
- 流式处理：支持实时数据分析、日志聚合和事件驱动架构。
- 高可靠性：Kafka 的副本机制和持久化策略确保消息不丢失，保证系统稳定性。
2. Kafka 核心概念
在集成之前，先回顾 Kafka 的几个关键特性：
- 发布/订阅模型：生产者发送消息到主题（Topic），消费者订阅主题消费消息。
- 消息日志与偏移量：消息以日志形式持久化存储，每条消息在分区（Partition）内有唯一递增的偏移量（Offset）。
- 分布式架构：集群由多个 Broker 组成，分区可复制到多个 Broker，保证高可用和数据冗余。
- 高吞吐与低延迟：通过批量处理、零拷贝等技术实现高性能。
- 水平扩展：通过增加 Broker 和分区即可线性提升集群处理能力。
- 流处理：Kafka Streams API 支持在集群内直接进行实时数据处理。
3. 在 Spring Boot 中集成 Kafka
3.1. 添加依赖
// Maven
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>2.9.13</version>
</dependency>
// Gradle
implementation 'org.springframework.kafka:spring-kafka:2.9.13'3.2. 配置 application.yml
统一管理 Kafka 相关配置，包括主题名称、消费者组 ID、序列化/反序列化方式等：
custom:
topics:
info: "ENTRY_USER_INFO"
group:
id: "hrfan-consumer-group"
spring:
kafka:
consumer:
bootstrap-servers: 192.168.112.128:9092
group-id: ${custom.group.id}
auto-offset-reset: earliest
key-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
properties:
spring.json.trusted.packages: '*'
spring.deserializer.key.delegate.class: org.apache.kafka.common.serialization.StringDeserializer
spring.deserializer.value.delegate.class: org.apache.kafka.common.serialization.StringDeserializer
producer:
bootstrap-servers: 192.168.112.128:9092
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.springframework.kafka.support.serializer.JsonSerializer4. 生产者实现
利用 KafkaTemplate 即可快速发送消息，支持多种发送模式：
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
@Component
public class SimpleProducer {
private static final Logger log = LoggerFactory.getLogger(SimpleProducer.class);
@Value("${custom.topics.info}")
private String topicName;
@Resource
private KafkaTemplate<String, String> strTemplate;
@Resource
private KafkaTemplate<String, Message<String>> msgTemplate;
public void sendMessages() {
// 1. 无 Key 消息
strTemplate.send(topicName, "这是一条普通消息");
// 2. 带 Key 的消息
strTemplate.send(topicName, "my-key-1", "带有Key的消息");
// 3. 指定分区和 Key
strTemplate.send(topicName, 0, "my-key-2", "指定分区消息");
// 4. 发送 Message 对象（可携带自定义头信息）
Map<String, Object> headersMap = new HashMap<>();
headersMap.put("traceId", UUID.randomUUID().toString());
MessageHeaders headers = new MessageHeaders(headersMap);
Message<String> message = MessageBuilder.createMessage("这是一个Message格式的消息", headers);
msgTemplate.setDefaultTopic(topicName);
msgTemplate.sendDefault("my-key-3", message);
log.info("所有消息发送完毕");
}
}5. 消费者实现
通过 @KafkaListener 注解即可声明一个消费者：
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.stereotype.Component;
import java.util.Map;
@Component
public class SimpleConsumer {
private static final Logger log = LoggerFactory.getLogger(SimpleConsumer.class);
@KafkaListener(
topics = "#{'${custom.topics.info}'.split(',')}",
groupId = "${custom.group.id}",
autoStartup = "true"
)
public void handleMessage(ConsumerRecord<String, String> record,
@Headers Map<String, Object> headers) {
log.info("接收到的消息记录：{}", record);
}
}6. ACK 机制详解
ACK（Acknowledgement）是生产者的消息确认机制，决定了数据的可靠性级别：
- acks=0：不等待确认，性能最高，风险最大。
- acks=1：等待 Leader 写入成功，默认值，兼顾性能与可靠性。
- acks=all：等待所有 ISR 副本确认，安全性最高，延迟最大。
6.1. Spring Kafka 中的 AckMode
Spring Kafka 提供了多种确认模式，通过 AckMode 枚举定义：
- RECORD：每条消息处理完后立即提交 Offset。
- BATCH：每批消息处理完后提交一次 Offset。
- TIME：每隔固定时间间隔提交一次 Offset。
- COUNT：每处理 N 条消息后提交一次 Offset。
- COUNT_TIME：结合数量和时间阈值，先到先提交。
- MANUAL：手动调用
Acknowledgment.acknowledge()提交。 - MANUAL_IMMEDIATE：手动提交后立即执行，不等待。
6.2. 手动提交示例
将自动提交关闭并设置 ack-mode 为 MANUAL_IMMEDIATE：
spring:
kafka:
consumer:
enable-auto-commit: false
properties:
ack-mode: MANUAL_IMMEDIATE消费者代码中注入 Acknowledgment 参数，在需要时手动提交：
@KafkaListener(topics = "#{'${custom.topics.info}'.split(',')}",
groupId = "${custom.group.id}",
autoStartup = "true")
public void manualCommit(ConsumerRecord<String, String> record,
@Headers Map<String, Object> headers,
Acknowledgment acknowledgment) {
log.info("处理消息：{}", record);
// 业务处理完成后手动提交 Offset
acknowledgment.acknowledge();
}7. 常用配置参考
7.1. 生产端关键配置
| 属性 | 说明 |
|---|---|
request.required.acks | ACK 确认级别，0/1/-1(即all) |
producer.type | sync（同步）或 async（异步） |
batch.num.messages | 异步模式下每批次发送的消息数 |
message.send.max.retries | 发送失败重试次数 |
7.2. 消费端关键配置
| 属性 | 说明 |
|---|---|
group.id | 消费者组 ID，用于负载均衡 |
enable.auto.commit | 是否允许自动提交 Offset |
auto.offset.reset | 无初始 Offset 时策略（earliest/latest/none） |
fetch.min.bytes | 每次拉取的最小字节数 |
7.3. 服务端关键配置
| 属性 | 说明 |
|---|---|
broker.id | 节点唯一标识 |
log.dirs | 日志存储目录 |
num.partitions | 默认分区数 |
log.retention.hours | 数据保留时间（小时） |
zookeeper.connect | ZooKeeper 连接地址 |
8. 命令行消费数据
在服务器上可以直接使用 Kafka 提供的命令行工具查看消息：
kafka-console-consumer \
--bootstrap-server 192.168.112.129:9092 \
--topic ENTRY_LIST_SEND_INFO \
--from-beginning该命令会从指定主题的最开始位置读取所有消息并在控制台输出。
