Kafka是一個開源的分散式事件流平臺，常被用於高性能數據管道、流分析、數據集成和關鍵任務應用，基於Zookeeper協調的處理平臺，也是一種消息系統，具有更好的吞吐量、內置分區、複製和容錯。 ...
目錄
標簽：Kafka3.Kafka-eagle3；
一、簡介
Kafka是一個開源的分散式事件流平臺，常被用於高性能數據管道、流分析、數據集成和關鍵任務應用，基於Zookeeper協調的處理平臺，也是一種消息系統，具有更好的吞吐量、內置分區、複製和容錯，這使得它成為大規模消息處理應用程式的一個很好的解決方案；
二、環境搭建
1、Kafka部署
1、下載安裝包：kafka_2.13-3.5.0.tgz
2、配置環境變數
open -e ~/.bash_profile
export KAFKA_HOME=/本地路徑/kafka3.5
export PATH=$PATH:$KAFKA_HOME/bin
source ~/.bash_profile
3、該目錄【kafka3.5/bin】啟動zookeeper
zookeeper-server-start.sh ../config/zookeeper.properties
4、該目錄【kafka3.5/bin】啟動kafka
kafka-server-start.sh ../config/server.properties
2、Kafka測試
1、生產者
kafka-console-producer.sh --broker-list localhost:9092 --topic test-topic
>id-1-message
>id-2-message
2、消費者
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic
id-1-message
id-2-message
3、查看topic列表
kafka-topics.sh --bootstrap-server localhost:9092 --list
test-topic
4、查看消息列表
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic --from-beginning --partition 0
id-1-message
id-2-message
3、可視化工具
配置和部署
1、下載安裝包：kafka-eagle-bin-3.0.2.tar.gz
2、配置環境變數
open -e ~/.bash_profile
export KE_HOME=/本地路徑/efak-web-3.0.2
export PATH=$PATH:$KE_HOME/bin
source ~/.bash_profile
3、修改配置文件：system-config.properties
efak.zk.cluster.alias=cluster1
cluster1.zk.list=localhost:2181
efak.url=jdbc:mysql://127.0.0.1:3306/kafka-eagle
4、本地新建資料庫：kafka-eagle，註意用戶名和密碼是否一致
5、啟動命令
efak-web-3.0.2/bin/ke.sh start
命令語法: ./ke.sh {start|stop|restart|status|stats|find|gc|jdk|version|sdate|cluster}
6、本地訪問【localhost:8048】 username:admin password:123456
KSQL語句測試
select * from `test-topic` where `partition` in (0) order by `date` desc limit 5
select * from `test-topic` where `partition` in (0) and msg like '%5%' order by `date` desc limit 3
三、工程搭建
1、工程結構
2、依賴管理
這裡關於依賴的管理就比較複雜了，首先spring-kafka組件選擇與boot框架中spring相同的依賴，即6.0.10版本，在spring-kafka最近的版本中3.0.8符合；
但是該版本使用的是kafka-clients組件的3.3.2版本，在Spring文檔的kafka模塊中，明確說明spring-boot:3.1要使用kafka-clients:3.4，所以從spring-kafka組件中排除掉，重新依賴kafka-clients組件；
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>${spring-kafka.version}</version>
<exclusions>
<exclusion>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>${kafka-clients.version}</version>
</dependency>
3、配置文件
配置kafka連接地址，監聽器的消息應答機制，消費者的基礎模式；
spring:
# kafka配置
kafka:
bootstrap-servers: localhost:9092
listener:
missing-topics-fatal: false
ack-mode: manual_immediate
consumer:
group-id: boot-kafka-group
enable-auto-commit: false
max-poll-records: 10
properties:
max.poll.interval.ms: 3600000
四、基礎用法
1、消息生產
模板類KafkaTemplate用於執行高級的操作，封裝各種消息發送的方法，在該方法中，通過topic和key以及消息主體，實現消息的生產；
@RestController
public class ProducerWeb {
@Resource
private KafkaTemplate<String, String> kafkaTemplate;
@GetMapping("/send/msg")
public String sendMsg (){
try {
// 構建消息主體
JsonMapper jsonMapper = new JsonMapper();
String msgBody = jsonMapper.writeValueAsString(new MqMsg(7,"boot-kafka-msg"));
// 發送消息
kafkaTemplate.send("boot-kafka-topic","boot-kafka-key",msgBody);
} catch (JsonProcessingException e) {
e.printStackTrace();
}
return "OK" ;
}
}
2、消息消費
編寫消息監聽類，通過KafkaListener註解控制監聽的具體信息，在實現消息生產和消費的方法測試後，使用可視化工具kafka-eagle查看topic和消息列表；
@Component
public class ConsumerListener {
private static final Logger log = LoggerFactory.getLogger(ConsumerListener.class);
@KafkaListener(topics = "boot-kafka-topic")
public void listenUser (ConsumerRecord<?,String> record, Acknowledgment acknowledgment) {
try {
String key = String.valueOf(record.key());
String body = record.value();
log.info("\n=====\ntopic:boot-kafka-topic，key{}，body:{}\n=====\n",key,body);
} catch (Exception e){
e.printStackTrace();
} finally {
acknowledgment.acknowledge();
}
}
}
五、參考源碼
文檔倉庫：
https://gitee.com/cicadasmile/butte-java-note
源碼倉庫：
https://gitee.com/cicadasmile/butte-spring-parent
