Spring Boot 整合 Kafka：生产环境标准配置与最佳实践

一、环境准备

✅ 启动 Kafka 服务（本地开发示例）：

# 启动 Zookeeper（Kafka 3.0+ 已内置）
bin/zookeeper-server-start.sh config/zookeeper.properties

# 启动 Kafka 服务
bin/kafka-server-start.sh config/server.properties

二、项目搭建（Maven 依赖）

1. 添加 Spring Kafka 依赖

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

💡 关键点：Spring Boot 3.x 已移除 spring-boot-starter-kafka，直接使用 spring-kafka。

三、YAML 配置（application.yml）

✅ 重点：YAML 配置替代 properties，支持对象序列化器配置

spring:
kafka:
bootstrap-servers: localhost:9092
producer:
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: com.example.config.AttendanceSerializer # 自定义序列化器
batch-size: 16384 # 批量发送优化
linger-ms: 100 # 批量发送延迟
retries: 3
acks: all
consumer:
group-id: attendance-group
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: com.example.config.AttendanceDeserializer # 自定义反序列化器
auto-offset-reset: earliest
enable-auto-commit: true
max-poll-records: 500
isolation.level: read_committed
# 事务配置（可选）
transaction-id-prefix: attendance-trans

🌟 YAML 配置优势：

• 层级清晰，避免 . 重复
• 支持嵌套配置（如 producer.batch-size）
• 与 Spring Boot 3.x 完美兼容

四、自定义序列化器（核心代码）

1. 定义消息实体（AttendanceStatisticsSingleDto）

package jnpf.model.attendance.event;

import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import javax.validation.constraints.NotBlank;
import javax.validation.constraints.NotNull;
import java.io.Serializable;
import java.util.Date;

@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class AttendanceStatisticsSingleDto implements Serializable {
private static final long serialVersionUID = 1L;

@NotBlank(message = "租户Id不能为空")
private String tenantId;

@NotBlank(message = "考勤组Id不能为空")
private String groupId;

@NotBlank(message = "用户Id不能为空")
private String userId;

@NotNull(message = "日期不能为空")
private Date day;
}

2. 创建序列化器（AttendanceSerializer.java）

package com.example.config;

import com.fasterxml.jackson.databind.ObjectMapper;
import jnpf.model.attendance.event.AttendanceStatisticsSingleDto;
import org.apache.kafka.common.serialization.Serializer;

import java.util.Map;

public class AttendanceSerializer implements Serializer<AttendanceStatisticsSingleDto> {

private final ObjectMapper objectMapper = new ObjectMapper();

@Override
public void configure(Map<String, ?> configs, boolean isKey) {
// 配置方法，无需额外操作
}

@Override
public byte[] serialize(String topic, AttendanceStatisticsSingleDto data) {
try {
return objectMapper.writeValueAsBytes(data);
} catch (Exception e) {
throw new RuntimeException("序列化失败: " + data, e);
}
}

@Override
public void close() {
// 无需关闭资源
}
}

3. 创建反序列化器（AttendanceDeserializer.java）

package com.example.config;

import com.fasterxml.jackson.databind.ObjectMapper;
import jnpf.model.attendance.event.AttendanceStatisticsSingleDto;
import org.apache.kafka.common.serialization.Deserializer;

import java.util.Map;

public class AttendanceDeserializer implements Deserializer<AttendanceStatisticsSingleDto> {

private final ObjectMapper objectMapper = new ObjectMapper();

@Override
public void configure(Map<String, ?> configs, boolean isKey) {
// 配置方法，无需额外操作
}

@Override
public AttendanceStatisticsSingleDto deserialize(String topic, byte[] data) {
try {
return objectMapper.readValue(data, AttendanceStatisticsSingleDto.class);
} catch (Exception e) {
throw new RuntimeException("反序列化失败: " + topic, e);
}
}

@Override
public void close() {
// 无需关闭资源
}
}

💡 为什么需要自定义序列化器？ Spring Kafka 默认只支持 String/byte[]，复杂对象必须通过序列化器转换为字节数组。

五、完整集成示例

1. 生产者服务（发送 AttendanceStatisticsSingleDto）

package com.example.service;

import com.example.config.AttendanceSerializer;
import jnpf.model.attendance.event.AttendanceStatisticsSingleDto;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class AttendanceProducer {

private final KafkaTemplate<String, AttendanceStatisticsSingleDto> kafkaTemplate;

public AttendanceProducer(KafkaTemplate<String, AttendanceStatisticsSingleDto> kafkaTemplate) {
this.kafkaTemplate = kafkaTemplate;
}

public void sendAttendanceData(AttendanceStatisticsSingleDto attendance) {
// 发送消息到主题 attendance-topic
kafkaTemplate.send("attendance-topic", attendance.getUserId(), attendance);

System.out.println("✅ 消息发送成功: " + attendance.getUserId() + " | " + attendance.getDay());
}
}

2. 消费者服务（接收 AttendanceStatisticsSingleDto）

package com.example.consumer;

import com.example.config.AttendanceDeserializer;
import jnpf.model.attendance.event.AttendanceStatisticsSingleDto;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

@Service
public class AttendanceConsumer {

@KafkaListener(
topics = "attendance-topic",
groupId = "attendance-group",
containerFactory = "kafkaListenerContainerFactory"
)
public void listen(AttendanceStatisticsSingleDto attendance,
Acknowledgment ack,
Headers headers) {

// 获取 Kafka 元数据
int partition = Integer.parseInt(headers.lastHeader("partition").value().toString());
long offset = headers.lastHeader("offset").value().toString().getBytes()[0];

System.out.println("✅ 消息消费成功 | 分区: " + partition +
" | 偏移量: " + offset +
" | 用户: " + attendance.getUserId());

// 手动提交偏移量（可选）
ack.acknowledge();
}
}

3. 配置 Kafka 监听器容器工厂（关键！）

package com.example.config;

import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.ContainerCustomizer;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.support.serializer.DeserializationException;

@Configuration
public class KafkaConfig {

@Bean
public ConsumerFactory<String, AttendanceStatisticsSingleDto> consumerFactory() {
// 配置消费者工厂
return new DefaultKafkaConsumerFactory<>(
Map.of(
"bootstrap.servers", "localhost:9092",
"group.id", "attendance-group",
"key.deserializer", StringDeserializer.class.getName(),
"value.deserializer", AttendanceDeserializer.class.getName()
)
);
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, AttendanceStatisticsSingleDto> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, AttendanceStatisticsSingleDto> factory =
new ConcurrentKafkaListenerContainerFactory<>();

factory.setConsumerFactory(consumerFactory());
factory.setCommonErrorHandler(new DefaultErrorHandler(
new FixedBackOff(1000L, 3), // 重试3次，间隔1秒
new TopicPartitionOffset("attendance-topic.DLQ", 0) // 死信队列
));

// 启用自动提交
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);

return factory;
}
}

🌟 关键配置说明：

• setCommonErrorHandler：实现自动重试 + 死信队列
• setAckMode(ContainerProperties.AckMode.MANUAL)：手动提交偏移量（推荐）
• AttendanceDeserializer：与生产者序列化器严格匹配

六、测试用例（JUnit 5）

package com.example;

import com.example.service.AttendanceProducer;
import jnpf.model.attendance.event.AttendanceStatisticsSingleDto;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

import java.util.Date;

@SpringBootTest
public class KafkaIntegrationTest {

@Autowired
private AttendanceProducer producer;

@Test
public void testSendAttendanceData() {
AttendanceStatisticsSingleDto attendance = AttendanceStatisticsSingleDto.builder()
.tenantId("tenant_001")
.groupId("group_001")
.userId("user_001")
.day(new Date())
.build();

producer.sendAttendanceData(attendance);

// 实际测试需等待消费者处理（此处简化）
System.out.println("✅ 测试消息已发送");
}
}

✅ 测试执行流程：

七、最佳实践与配置优化

八、常见问题排查（YAML 配置特有）

九、总结：Spring Boot + Kafka 核心流程

✅ 关键结论：

💡 立即实践：