[Java实战]Spring Boot整合Kafka：高吞吐量消息系统实战（二十七）

一、引言

Apache Kafka作为一款高吞吐量、低延迟的分布式消息队列系统，广泛应用于实时数据处理、日志收集和事件驱动架构。结合Spring Boot的自动化配置能力，可以快速搭建高性能消息系统。本文将从环境搭建、代码实现、原理分析到测试优化，全面解析Spring Boot与Kafka的整合实践。

二、环境准备

1. Kafka安装与启动

下载Kafka：从Apache Kafka官网下载最新版本（推荐3.x+）。

启动Zookeeper（Kafka依赖）：

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

启动Kafka服务：

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

2. 创建Topic

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

说明：手动创建Topic可指定分区数（如3），提升并发处理能力。

三、环境准备(docker)

1. 使用Docker快速启动Kafka

通过Docker可以快速部署Kafka服务，无需手动安装依赖，步骤如下：

创建docker-compose.yml文件：
在项目根目录下新建文件，内容如下：

version: '3'
services:zookeeper:image: docker.1ms.run/confluentinc/cp-zookeeper:7.4.0ports:- "2181:2181"environment:ZOOKEEPER_CLIENT_PORT: 2181ZOOKEEPER_TICK_TIME: 2000kafka:image: docker.1ms.run/confluentinc/cp-kafka:7.4.0ports:- "9092:9092"environment:KAFKA_BROKER_ID: 1KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://192.168.231.132:9092KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false"  # 禁止自动创建Topicdepends_on:- zookeeper

关键配置说明：

KAFKA_ADVERTISED_LISTENERS: 确保客户端能通过localhost:9092访问Kafka。

KAFKA_AUTO_CREATE_TOPICS_ENABLE: 设为false避免自动创建Topic，推荐手动控制。

启动Kafka服务：
执行以下命令启动服务：
```
docker-compose up -d#停掉
#docker-compose down
```

2. 创建Topic

通过Docker执行命令创建Topic：

docker exec -it kafka-kafka-1 kafka-topics --create --topic my-topic --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1

注意：

kafka-kafka-1为容器名称（根据实际名称调整）。
--partitions 3指定分区数，提升并发处理能力。

3.安装成功截图

在这里插入图片描述

四、Spring Boot项目搭建

1. 添加依赖

在pom.xml中引入Spring Kafka：

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

2. 配置文件

application.yml配置Kafka连接及序列化方式：

spring:kafka:bootstrap-servers: localhost:9092producer:key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializerconsumer:group-id: my-groupauto-offset-reset: earliestkey-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializer

关键参数：auto-offset-reset: earliest确保消费者从最早消息开始消费。

五、代码实现

1. 生产者配置

@Service
public class KafkaProducer {@Autowiredprivate KafkaTemplate<String, String> kafkaTemplate;// 发送消息（支持回调）public void sendMessage(String topic, String message) {ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, message);future.addCallback(result -> {System.out.println("发送成功: " + result.getRecordMetadata().offset());}, ex -> {System.out.println("发送失败: " + ex.getMessage());});}
}

高级特性：回调机制可监控消息发送状态。

2. 消费者配置

@Service
public class KafkaConsumer {@KafkaListener(topics = "my-topic", groupId = "my-group")public void consume(String message) {System.out.println("接收到消息: " + message);// 业务处理逻辑}
}

批量消费：通过设置spring.kafka.consumer.max-poll-records可支持批量处理。

3.测试结果

KafkaController编写：

@RestController
public class KafkaController {@Autowiredprivate KafkaProducerService kafkaProducer;@PostMapping("/send")public ResponseEntity<String> sendMs(@RequestBody String request) {kafkaProducer.sendMessage("my-topic","你好");return ResponseEntity.ok("ok");}
}

测试结果：

在这里插入图片描述

六、原理分析

1. Spring Kafka核心组件

KafkaTemplate：封装生产者操作，支持异步发送和事务管理。
@KafkaListener：基于监听器模式，自动创建消费者并订阅Topic。
ConsumerFactory/ProducerFactory：工厂类管理Kafka客户端配置。

2. 高吞吐量优化

生产者端：调整batch.size（批次大小）和linger.ms（等待时间）提升批量发送效率。
消费者端：增加分区数、配置多线程消费（ConcurrentKafkaListenerContainerFactory）。

七、高级特性

1. 自定义分区策略

实现Partitioner接口，指定消息路由规则：

public class CustomPartitioner implements Partitioner {@Overridepublic int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {// 自定义分区逻辑（如按Key哈希）return key.hashCode() % cluster.partitionCountForTopic(topic);}
}

配置文件中指定分区器：

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

2. 事务支持

通过KafkaTransactionManager实现事务消息：

@Autowired
private KafkaTemplate<String, String> kafkaTemplate;public void sendInTransaction() {kafkaTemplate.executeInTransaction(operations -> {operations.send("topic1", "Message1");operations.send("topic2", "Message2");return null;});
}

八、测试步骤

1. 单元测试（使用嵌入式Kafka）

添加测试依赖：

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

编写测试类：

@SpringBootTest
@EmbeddedKafka(topics = "test-topic")
public class KafkaTest {@Autowiredprivate KafkaTemplate<String, String> kafkaTemplate;@Testpublic void testSendAndReceive() {kafkaTemplate.send("test-topic", "Hello Kafka");// 通过监听器验证消息接收}
}