Spring Boot整合Kafka实战指南：从环境搭建到消息处理全解析

一、环境准备

1. 安装 Kafka

   • 下载 Kafka：从 Apache Kafka 官网下载对应版本的 Kafka。

   • 解压并启动 Kafka：

     # 启动 Zookeeper（Kafka 依赖 Zookeeper）
     bin/zookeeper-server-start.sh config/zookeeper.properties# 启动 Kafka
     bin/kafka-server-start.sh config/server.properties

2. 创建主题

   bin/kafka-topics.sh --create --topic test-topic --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1

3. 添加依赖 在 Spring Boot 项目的 pom.xml 文件中添加 Kafka 依赖：

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

二、基础配置（YML）

在 application.yml 文件中配置 Kafka 的基本参数：

spring:kafka:bootstrap-servers: localhost:9092consumer:group-id: test-groupauto-offset-reset: earliestkey-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializerproducer:key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializer

• bootstrap-servers：Kafka 服务的地址。

• group-id：消费者组 ID，同一组的消费者消费不同的分区。

• auto-offset-reset：当 Kafka 中没有初始偏移量或偏移量超出范围时的处理方式，可选值为 earliest（从最早记录开始读取）、latest（从最新记录开始读取）或 none（抛出异常）。

三、生产者与消费者代码实现

1. 生产者

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;@Component
public class KafkaProducer {@Autowiredprivate KafkaTemplate<String, String> kafkaTemplate;public void sendMessage(String topic, String message) {kafkaTemplate.send(topic, message);}
}

• KafkaTemplate：Spring 提供的 Kafka 操作类，用于发送消息。

2. 消费者

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;@Component
public class KafkaConsumer {@KafkaListener(topics = "test-topic")public void receiveMessage(String message) {System.out.println("Received message: " + message);}
}

• @KafkaListener：指定监听的 Kafka 主题。

四、高级用法

1. 消息分区策略

自定义分区器

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;import java.util.Map;public class CustomPartitioner implements Partitioner {@Overridepublic void configure(Map<String, ?> configs) {// 配置初始化}@Overridepublic int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {// 自定义分区逻辑，例如根据 key 的哈希值取模return key.hashCode() % cluster.partitionCountForTopic(topic);}@Overridepublic void close() {// 关闭资源}
}

在 application.yml 中配置自定义分区器：

spring:kafka:producer:partitioner-class: com.example.CustomPartitioner

2. 消息序列化与反序列化

自定义序列化器

import org.apache.kafka.common.serialization.Serializer;public class CustomSerializer implements Serializer<MyObject> {@Overridepublic byte[] serialize(String topic, MyObject data) {// 自定义序列化逻辑，例如使用 JSON 序列化return data.toJson().getBytes();}
}

在 application.yml 中配置自定义序列化器：

spring:kafka:producer:value-serializer: com.example.CustomSerializer

3. 消费者并发处理

在 application.yml 中配置消费者并发数：

spring:kafka:listener:concurrency: 3

• concurrency：每个消费者线程的并发数，可以根据需求调整。

4. 消费者手动提交偏移量

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;@Component
public class KafkaConsumer {@KafkaListener(topics = "test-topic")public void receiveMessage(String message, Acknowledgment acknowledgment) {System.out.println("Received message: " + message);// 处理完消息后手动提交偏移量acknowledgment.acknowledge();}
}

• Acknowledgment：手动提交偏移量的接口。

5. 消费者拦截器

自定义拦截器

import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;import java.util.HashMap;
import java.util.Map;public class CustomConsumerInterceptor implements ConsumerInterceptor<String, String> {@Overridepublic ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> records) {// 消费前的逻辑，例如过滤消息return records;}@Overridepublic void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {// 提交偏移量前的逻辑}@Overridepublic void close() {// 关闭资源}@Overridepublic void configure(Map<String, ?> configs) {// 配置初始化}
}

在 application.yml 中配置自定义拦截器：

spring:kafka:consumer:interceptor-classes: com.example.CustomConsumerInterceptor

6. 生产者拦截器

自定义拦截器

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;import java.util.Map;public class CustomProducerInterceptor implements ProducerInterceptor<String, String> {@Overridepublic ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {// 发送前的逻辑，例如修改消息内容return record;}@Overridepublic void onAcknowledgement(RecordMetadata metadata, Exception exception) {// 消息发送后的逻辑}@Overridepublic void close() {// 关闭资源}@Overridepublic void configure(Map<String, ?> configs) {// 配置初始化}
}

在 application.yml 中配置自定义拦截器：

spring:kafka:producer:interceptor-classes: com.example.CustomProducerInterceptor

7. 消费者重试机制

在 application.yml 中配置消费者重试机制：

spring:kafka:listener:retry:initial-interval: 1000msmax-attempts: 3back-off-multiplier: 2.0max-interval: 10000ms

• initial-interval：初始重试间隔。

• max-attempts：最大重试次数。

• back-off-multiplier：重试间隔的倍数。

• max-interval：最大重试间隔。

8. 消费者死信队列

在 Kafka 中创建死信队列主题：

bin/kafka-topics.sh --create --topic dlq-topic --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1

在 application.yml 中配置死信队列：

spring:kafka:listener:dead-letter:topic: dlq-topic

9. 消息过滤

在 application.yml 中配置消息过滤：

spring:kafka:listener:filter:expression: headers['type'] == 'important'

• expression：SpEL 表达式，用于过滤消息。

10. 消费者负载均衡

Spring Boot 默认支持 Kafka 的消费者负载均衡，无需额外配置。只需在同一个消费者组中启动多个