Spring Boot集成Apache Kafka的實(shí)戰(zhàn)指南
Apache Kafka 是一個(gè)分布式流處理平臺(tái)，廣泛用于構(gòu)建實(shí)時(shí)數(shù)據(jù)管道、日志聚合系統(tǒng)和事件溯源架構(gòu)。Spring Boot 提供了對(duì) Kafka 的良好集成支持，使得開發(fā)者可以非常便捷地在項(xiàng)目中使用 Kafka。
本文將手把手教你如何在 Spring Boot 項(xiàng)目中集成 Kafka，包括生產(chǎn)者（Producer）和消費(fèi)者（Consumer）的實(shí)現(xiàn)，并提供完整的代碼示例。
開發(fā)環(huán)境準(zhǔn)備
Java 17+
Maven 或 Gradle
Spring Boot 3.x
Apache Kafka 3.0+（本地或遠(yuǎn)程）
IDE（如 IntelliJ IDEA、VS Code）
創(chuàng)建 Spring Boot 項(xiàng)目
你可以通過 Spring Initializr 創(chuàng)建一個(gè)新的 Spring Boot 項(xiàng)目，選擇以下依賴：
- Spring Web
- Spring for Apache Kafka
或者手動(dòng)添加 pom.xml 中的依賴：
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
Spring Boot 會(huì)自動(dòng)管理版本兼容性，無需手動(dòng)指定版本號(hào)。
配置 Kafka 連接信息
在 application.yml 或 application.properties 文件中配置 Kafka 相關(guān)參數(shù)：
application.yml 示例：
spring:
kafka:
bootstrap-servers: localhost:9092
consumer:
group-id: my-group
auto-offset-reset: earliest
producer:
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
編寫 Kafka 生產(chǎn)者（Producer）
創(chuàng)建一個(gè)服務(wù)類用于發(fā)送消息到 Kafka 主題。
KafkaProducer.java
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
@Service
public class KafkaProducer {
private final KafkaTemplate<String, String> kafkaTemplate;
public KafkaProducer(KafkaTemplate<String, String> kafkaTemplate) {
this.kafkaTemplate = kafkaTemplate;
}
public void sendMessage(String topic, String message) {
kafkaTemplate.send(topic, message);
System.out.println("Sent message: " + message);
}
}
編寫 Kafka 消費(fèi)者（Consumer）
使用 @KafkaListener 注解監(jiān)聽特定主題的消息。
KafkaConsumer.java
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;
@Service
public class KafkaConsumer {
@KafkaListener(topics = "test-topic", groupId = "my-group")
public void listen(ConsumerRecord<String, String> record) {
System.out.printf("Received message: topic - %s, partition - %d, offset - %d, key - %s, value - %s%n",
record.topic(), record.partition(), record.offset(), record.key(), record.value());
}
}
添加 REST 接口用于測(cè)試發(fā)送消息
為了方便測(cè)試，我們可以創(chuàng)建一個(gè)簡(jiǎn)單的 REST 控制器來觸發(fā)消息發(fā)送。
KafkaController.java
import org.springframework.web.bind.annotation.*;
import org.springframework.beans.factory.annotation.Autowired;
@RestController
@RequestMapping("/kafka")
public class KafkaController {
@Autowired
private KafkaProducer kafkaProducer;
@PostMapping("/send")
public String sendMessage(@RequestParam String msg) {
kafkaProducer.sendMessage("test-topic", msg);
return "Message sent: " + msg;
}
}
啟動(dòng) Kafka 環(huán)境（可選）
如果你還沒有運(yùn)行 Kafka，可以按照以下步驟快速啟動(dòng)：
啟動(dòng) Zookeeper（Kafka 依賴）
bin/zookeeper-server-start.sh config/zookeeper.properties
啟動(dòng) Kafka 服務(wù)
bin/kafka-server-start.sh config/server.properties
創(chuàng)建測(cè)試 Topic
bin/kafka-topics.sh --create --topic test-topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1
測(cè)試接口
啟動(dòng) Spring Boot 應(yīng)用后，訪問如下接口發(fā)送消息：
POST http://localhost:8080/kafka/send?msg=HelloKafka
觀察控制臺(tái)輸出，確認(rèn)是否收到類似以下內(nèi)容：
Received message: topic - test-topic, partition - 0, offset - 5, key - null, value - HelloKafka
擴(kuò)展功能建議
使用 JSON 格式傳輸對(duì)象（自定義序列化/反序列化）
多消費(fèi)者組配置與負(fù)載均衡
異常處理與重試機(jī)制（@DltHandler, SeekToCurrentErrorHandler）
Kafka Streams 實(shí)現(xiàn)實(shí)時(shí)流處理邏輯
配置 SSL、SASL 安全認(rèn)證
結(jié)合 Spring Cloud Stream 構(gòu)建云原生事件驅(qū)動(dòng)架構(gòu)
到此這篇關(guān)于Spring Boot集成Apache Kafka的實(shí)戰(zhàn)指南的文章就介紹到這了,更多相關(guān)SpringBoot集成Apache Kafka內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
這篇文章主要介紹了如何定位java程序中占用cpu最高的線程堆棧信息方法的相關(guān)資料,需要的朋友可以參考下2022-11-11
從java源碼分析線程池(池化技術(shù))的實(shí)現(xiàn)原理
這篇文章主要介紹了從java源碼分析線程池(池化技術(shù))的實(shí)現(xiàn)原理，池化技術(shù)是一種編程技巧，當(dāng)程序出現(xiàn)高并發(fā)時(shí)，能夠明顯的優(yōu)化程序，降低系統(tǒng)頻繁創(chuàng)建銷毀連接等額外開銷，下文更多的相關(guān)介紹需要的小伙伴可以參考一下2022-04-04
今天小編就為大家分享一篇關(guān)于java字符串中${}或者{}等的占位符替換工具類，小編覺得內(nèi)容挺不錯(cuò)的，現(xiàn)在分享給大家，具有很好的參考價(jià)值，需要的朋友一起跟隨小編來看看吧2018-12-12
文章講述了在Java中進(jìn)行批量操作的最佳實(shí)踐,包括使用MyBatis的ExecutorType.BATCH模式,作者建議根據(jù)實(shí)際需求選擇是否進(jìn)行批量操作,因?yàn)榕坎僮鞑⒉豢偸悄芴嵘阅?且有副作用,如無法獲取自增ID等2024-12-12
Java實(shí)現(xiàn)WAV音頻拼接徹底擺脫FFmpeg的輕量本地方案
在音頻處理領(lǐng)域,FFmpeg 是幾乎無所不能的存在,從音頻解碼、格式轉(zhuǎn)換、拼接到混音,幾乎所有任務(wù)都能用一句命令完成,然而,正因?yàn)樗?也意味著笨重,本文給大家介紹了利用Java實(shí)現(xiàn)WAV音頻拼接徹底擺脫FFmpeg的輕量本地方案,需要的朋友可以參考下2025-11-11
本篇文章是對(duì)在Java中一些基礎(chǔ)概念的使用進(jìn)行了詳細(xì)的分析介紹，需要的朋友參考下2013-05-05
Jackson?是?Java?中最流行的?JSON?處理庫之一，它提供了許多注解來簡(jiǎn)化?JSON?的序列化和反序列化過程。這篇文章將介紹一些?Jackson?常用的注解，以幫助您更輕松地處理?JSON?數(shù)據(jù)2023-05-05
Token過期實(shí)現(xiàn)自動(dòng)續(xù)期的方法步驟
本文詳細(xì)解析了Token在防止表單重復(fù)提交、身份驗(yàn)證及過期處理中的應(yīng)用,涵蓋Token的生成流程、過期機(jī)制及續(xù)期方案,特別介紹了單token與雙token兩種策略,需要的朋友可以參考下2026-05-05
這篇文章主要介紹了導(dǎo)入SpringCloud依賴踩的坑及解決方案，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2023-04-04
redis發(fā)布訂閱Java代碼實(shí)現(xiàn)過程解析
這篇文章主要介紹了redis發(fā)布訂閱Java代碼實(shí)現(xiàn)過程解析,文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2019-09-09