Apache Kafka实战：Spring Boot消息队列完整指南

前言

Apache Kafka 是分布式消息队列的事实标准，本文带你实战 Spring Boot 整合 Kafka，完成生产者和消费者的完整开发。
一、Kafka 核心概念

• Producer：消息生产者
  
• Consumer：消息消费者
  
• Broker：Kafka 服务节点
  
• Topic：消息主题分类
  
• Partition：Topic 的分区，实现并行处理
  
• Consumer Group：消费者组，实现负载均衡
  

二、Docker 安装 Kafka

1. # docker-compose.yml
2. services:
3.   zookeeper:
4.     image: confluentinc/cp-zookeeper:7.5.0
5.     environment:
6.       ZOOKEEPER_CLIENT_PORT: 2181
8.   kafka:
9.     image: confluentinc/cp-kafka:7.5.0
10.     depends_on: [zookeeper]
11.     ports: ["9092:9092"]
12.     environment:
13.       KAFKA_BROKER_ID: 1
14.       KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
15.       KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
16.       KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"

复制代码

三、Spring Boot 整合 Kafka

1. <dependency>
2.     <groupId>org.springframework.kafka</groupId>
3.     spring-kafka</artifactId>
4. </dependency>
6. # application.yml
7. spring:
8.   kafka:
9.     bootstrap-servers: localhost:9092
10.     producer:
11.       key-serializer: org.apache.kafka.common.serialization.StringSerializer
12.       value-serializer: org.apache.kafka.common.serialization.StringSerializer
13.     consumer:
14.       group-id: my-group
15.       auto-offset-reset: earliest
16.       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
17.       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

复制代码

四、生产者

1. @Service
2. public class OrderProducer {
4.     @Autowired
5.     private KafkaTemplate<String, String> kafkaTemplate;
7.     public void sendOrder(String orderId) {
8.         String message = "{"orderId":""+orderId+"","time":""+LocalDateTime.now()+""}";
9.         kafkaTemplate.send("order-topic", orderId, message);
10.         System.out.println("Sent: " + message);
11.     }
12. }

复制代码

五、消费者

1. @Component
2. public class OrderConsumer {
4.     @KafkaListener(topics = "order-topic", groupId = "my-group")
5.     public void consume(ConsumerRecord<String, String> record) {
6.         System.out.println("Received: key=" + record.key());
7.         System.out.println("Value: " + record.value());
8.         System.out.println("Partition: " + record.partition());
9.         System.out.println("Offset: " + record.offset());
10.     }
11. }

复制代码

六、手动提交 Offset

1. @KafkaListener(topics = "order-topic", groupId = "my-group")
2. public void consumeWithManualAck(
3.         ConsumerRecord<String, String> record,        Acknowledgment ack) {
4.     try {
5.         processOrder(record.value());
6.         ack.acknowledge();  // 手动确认
7.     } catch (Exception e) {
8.         // 处理失败，消息会被重新消费
9.         log.error("Failed to process: {}", record.key(), e);
10.     }
11. }
13. # 开启手动提交
14. spring.kafka.listener.ack-mode: manual

复制代码

七、常用命令速查

1. # 创建 Topic
2. kafka-topics.sh --create --topic order-topic \
3.   --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1
5. # 查看 Topic 列表
6. kafka-topics.sh --list --bootstrap-server localhost:9092
8. # 查看消费者组
9. kafka-consumer-groups.sh --list --bootstrap-server localhost:9092
11. # 控制台消费者
12. kafka-console-consumer.sh --topic order-topic \
13.   --bootstrap-server localhost:9092 --from-beginning

复制代码

总结

Kafka 是微服务架构中异步通信的核心组件。核心要点：Producer/Consumer 模式实现解耦、Partition 实现并行处理、Consumer Group 实现负载均衡、手动提交 Offset 保证消息可靠性。

觉得有帮助请点赞收藏！有问题欢迎评论区交流
来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！