1.概述

在大數據與人工智能深度融合的新時代，Apache Kafka 已成為企業實時數據流處理的核心引擎。然而，隨著業務規模的不斷擴大，Kafka 集群的監控與運維復雜度也隨之攀升——指標繁多、延遲難察、告警頻繁，傳統監控手段難以快速定位問題與優化性能。正因如此，EFAK-AI（Eagle For Apache Kafka - AI Enhanced，原Kafka Eagle） 應運而生。它在經典 EFAK 的基礎上全面升級，融合 AI 智能分析與分布式監控能力，為 Kafka 運維注入了全新的智能化體驗。本文將帶您深入了解 EFAK-AI 5.0.0 的全新功能與技術亮點。

GitHub開源地址：https://github.com/smartloli/EFAK

2.內容

EFAK-AI (Eagle For Apache Kafka - AI Enhanced) 是一款開源的 Kafka 智能監控與管理平臺，融合了人工智能技術，為 Kafka 運維提供智能化、可視化、自動化的全方位解決方案。

• ?? AI 驅動: 集成主流大語言模型（OpenAI、Claude、DeepSeek 等），提供智能對話式運維
• ?? 實時監控: 全方位監控 Kafka 集群健康狀態、性能指標、消費延遲等關鍵數據
• ?? 高性能: 基于 Spring Boot 3.x 和 JDK 17，采用響應式編程和異步處理
• ?? 易部署: 支持 Docker 一鍵部署和傳統 tar.gz 安裝包兩種方式

 2.1 核心特性

• 多模型支持: 集成 OpenAI、Claude、DeepSeek 等多種大語言模型
• Function Calling: AI 可自動調用后端函數查詢實時數據
• 圖表自動生成: 根據時序數據自動生成可視化圖表
• Kafka 專家: 專業的 Kafka 集群分析、性能優化和故障診斷建議
• 流式對話: 基于 SSE 的實時流式響應，體驗更流暢
• Markdown 渲染: 支持代碼高亮、表格、Mermaid 圖表等豐富格式
• 對話歷史: 完整的會話管理和歷史記錄功能

• 實時監控: Broker 節點狀態、主題分區、消費者組監控
• 性能指標: 吞吐量、延遲、存儲容量等關鍵指標
• 歷史數據: 長期趨勢分析和性能對比
• 多集群支持: 同時管理多個 Kafka 集群

• 智能分片: 基于 Redis 的分布式任務分片執行
• 故障轉移: 自動檢測節點故障并重新分配任務
• 負載均衡: 動態調整任務分配，優化資源利用
• 單節點優化: 自動檢測單節點環境，跳過分片邏輯

• 多渠道告警: 支持釘釘、微信、飛書等多種告警渠道
• 智能閾值: 基于歷史數據的動態閾值調整
• 告警聚合: 避免告警風暴，提供告警聚合和降噪
• 可視化配置: 直觀的告警規則配置界面

2.2 技術架構

模塊結構

EFAK-AI/
├── efak-ai/          # 告警功能模塊
├── efak-core/        # 核心功能模塊 (Kafka 連接、監控邏輯)
├── efak-dto/         # 數據傳輸對象
├── efak-tool/        # 工具類模塊
└── efak-web/         # Web 應用模塊 (控制器、服務、前端)

• 后端框架: Spring Boot 3.4.5
• 數據庫: MySQL 8.0+ (主數據庫)
• 緩存: Redis 6.0+ (分布式鎖、任務調度)
• 消息隊列: Apache Kafka 4.0.0
• ORM: MyBatis 3.0.4
• 前端: Thymeleaf
• 構建工具: Maven 3.6+
• Java 版本: JDK 17

2.3 快速開始

EFAK-AI 提供兩種部署方式：Docker 容器化部署（推薦）和 tar.gz 安裝包部署。

# 克隆項目
git clone https://github.com/smartloli/EFAK-AI.git
cd EFAK-AI

# 運行快速啟動腳本
./quick-start.sh

# 1. 克隆項目
git clone https://github.com/smartloli/EFAK-AI.git
cd EFAK-AI

# 2. 啟動所有服務（包括 MySQL、Redis）
docker-compose up -d

# 3. 查看日志
docker-compose logs -f efak-ai

# 4. 訪問應用
# http://localhost:8080
# 默認賬號: admin / admin

# 使用 nginx profile 啟動
docker-compose --profile nginx up -d

# 通過 http://localhost (80端口) 訪問

# 查看運行狀態
docker-compose ps

# 停止服務
docker-compose down

# 重啟服務
docker-compose restart efak-ai

# 查看日志
docker-compose logs -f

# 克隆項目
git clone https://github.com/smartloli/EFAK-AI.git
cd EFAK-AI

# 執行構建腳本
./build-package.sh

# 生成安裝包: efak-ai-5.0.0.tar.gz

# 傳輸到服務器（如果需要）
scp efak-ai-5.0.0.tar.gz user@server:/opt/

# 解壓
cd /opt
tar -zxvf efak-ai-5.0.0.tar.gz
cd efak-ai-5.0.0

# 目錄結構
# bin/      - 啟動腳本
# config/   - 配置文件
# libs/     - JAR 包
# logs/     - 日志目錄
# sql/      - SQL 腳本

mysql -u root -p
CREATE DATABASE efak_ai CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
USE efak_ai;
SOURCE /opt/efak-ai-5.0.0/sql/ke.sql;

spring:
  datasource:
    url: jdbc:mysql://localhost:3306/efak_ai?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai
    username: root
    password: your_password
  data:
    redis:
      host: localhost
      port: 6379

# 啟動
./bin/start.sh

# 查看日志
tail -f logs/efak-ai.log

# 查看狀態
./bin/status.sh

# 停止
./bin/stop.sh

# 重啟
./bin/restart.sh

# 查看進程（進程名顯示為 KafkaEagle）
ps aux | grep KafkaEagle

3.2 核心功能

功能特性：

• 實時監控: Kafka 集群整體運行狀態
• 可視化圖表: 消息流量、Topic 數量、Consumer 組狀態
• 性能指標: 吞吐量、延遲、存儲使用情況
• 告警統計: 實時告警數量和類型分布
• 響應式設計: 支持桌面和移動端訪問

關鍵指標：

• Broker 數量和狀態
• Topic 總數和活躍數
• Consumer Group 數量
• 消息總量和增長趨勢
• 集群健康度評分

功能特性：

• 多集群管理: 支持管理多個 Kafka 集群
• Broker 詳情: 查看每個 Broker 的詳細信息
• 性能監控: 實時 Broker CPU、內存、磁盤使用率
• 配置管理: 查看和修改集群配置
• 健康檢查: 自動檢測 Broker 健康狀態

支持的操作：

• 添加/刪除集群配置
• 查看 Broker 列表和狀態
• 監控 Broker 性能指標
• 查看 Controller 信息
• 集群元數據查詢

功能特性：

• Topic 列表: 展示所有 Topic 及其關鍵信息
• 高級搜索: 支持按名稱、分區數、副本數搜索
• 統計信息: 消息數量、存儲大小、消費者組數
• 創建 Topic: 可視化創建 Topic 界面
• 刪除 Topic: 安全刪除 Topic（帶確認）
• 批量操作: 支持批量刪除、修改配置

顯示字段：

• Topic 名稱
• 分區數量
• 副本因子
• 消息總數
• 存儲大小
• 消費者組數量
• 創建時間
• 操作按鈕

功能特性：

• 分區詳情: 每個分區的 Offset、Leader、Replicas
• 消息流量: 實時消息生產和消費速率
• 消息查詢: 按 Offset、時間戳、Key 查詢消息
• 配置管理: 查看和修改 Topic 配置
• 分區重分配: 手動觸發分區重新分配
• 消費延遲: 各消費者組的消費進度和延遲

詳細信息：

• 分區分布和狀態
• ISR (In-Sync Replicas) 列表
• Leader 選舉歷史
• 消息保留策略
• 壓縮類型和配置
• 消費者組訂閱情況

功能特性：

• 消費者組列表: 所有活躍的消費者組
• 消費進度: 每個組的消費 Offset 和 Lag
• 延遲告警: 消費延遲超過閾值自動告警
• 消費者詳情: 查看組內每個消費者實例
• 消費趨勢: 消費速率和延遲趨勢圖
• 重置 Offset: 支持重置消費位置

監控指標：

• Consumer Group ID
• 訂閱的 Topic 列表
• 總 Lag 數量
• 消費速率 (msg/s)
• 消費者實例數
• 最后消費時間
• 分區分配策略

功能特性：

• 智能助手: 基于大語言模型的 Kafka 運維助手
• 自然語言交互: 用自然語言查詢 Kafka 狀態
• 數據可視化: 自動生成圖表展示數據
• 問題診斷: 智能分析和解決 Kafka 問題
• 操作建議: 提供最佳實踐和優化建議
• 歷史記錄: 保存對話歷史便于回溯

支持的查詢類型：

• "查詢集群狀態"
• "分析 Topic XYZ 的消費延遲"
• "為什么消費者組 ABC 停止消費？"
• "如何優化 Topic 性能？"
• "生成消息流量趨勢圖"
• "診斷 Broker 性能問題"

AI 能力：

• 自動理解用戶意圖
• 查詢 Kafka 元數據和指標
• 生成可視化圖表 (Chart.js)
• 提供運維建議和最佳實踐
• 解釋 Kafka 概念和參數

功能特性：

• 模型管理: 配置和切換 AI 大語言模型
• API 配置: 管理 OpenAI、Azure、本地模型 API Key
• 參數調優: 調整 Temperature、Max Tokens 等參數
• 模型測試: 在線測試模型響應效果
• 使用統計: 查看 API 調用次數和費用
• 多模型支持: 支持 OpenAI GPT、Azure OpenAI、本地模型

支持的模型：

• OpenAI GPT-5 模型
• Azure OpenAI Service
• 本地部署的開源模型 (如 Llama, ChatGLM)
• 自定義模型接口

配置項：

• 模型名稱和版本
• API Endpoint
• API Key / Token
• Temperature (0.0 - 1.0)
• Max Tokens
• 超時設置
• 重試策略

功能特性：

• 實時監控: Broker、Topic、Consumer 性能指標
• 歷史數據: 支持查詢歷史性能數據
• 自定義圖表: 自定義監控指標和時間范圍
• 性能分析: 自動分析性能瓶頸
• 趨勢預測: 基于歷史數據預測趨勢
• 性能告警: 指標超過閾值自動告警

監控指標：

• Broker 指標:

  • CPU、內存、磁盤使用率
  • 網絡 I/O 吞吐量
  • 請求處理延遲
  • 活躍連接數

• Topic 指標:

  • 消息生產速率
  • 消息消費速率
  • 字節流入/流出速率
  • 分區 Leader 分布

• Consumer 指標:

  • 消費延遲 (Lag)
  • 消費速率
  • 提交頻率
  • 重平衡次數

功能特性：

• 告警規則: 自定義告警規則和閾值
• 多渠道通知: 支持釘釘、微信、飛書、Webhook
• 告警統計: 告警歷史和趨勢分析
• 告警靜默: 臨時屏蔽特定告警
• 智能告警: AI 分析告警原因和建議
• 告警模板: 預定義常用告警規則

告警類型：

• Broker 告警:

  • Broker 下線
  • CPU/內存/磁盤使用率過高
  • 網絡異常

• Topic 告警:

  • 分區 ISR 不足
  • 消息積壓
  • 副本同步延遲

• Consumer 告警:

  • 消費延遲過高
  • 消費者組下線
  • 消費停滯

通知渠道配置：

• 釘釘: Webhook URL
• 微信: Webhook URL
• 飛書: Webhook URL
• Webhook: 自定義 HTTP 回調

功能特性：

• 定時任務: 創建和管理定時任務
• Cron 表達式: 支持復雜的調度規則
• 任務類型: 數據采集、清理、備份、分析
• 執行歷史: 查看任務執行記錄和結果
• 失敗重試: 自動重試失敗的任務
• 任務日志: 詳細的執行日志

支持的任務類型：

• 數據采集任務:

  • Kafka 指標采集
  • JMX 數據采集
  • 日志收集

• 清理任務:

  • 過期數據清理
  • 日志歸檔
  • 臨時文件清理

• 備份任務:

  • 配置備份
  • 元數據導出
  • 數據快照

• 分析任務:

  • 性能報告生成
  • 趨勢分析
  • 異常檢測

調度配置：

• Cron 表達式編輯器
• 執行時間設置
• 并發控制
• 超時設置
• 失敗重試策略

功能特性：

• 用戶管理: 創建、編輯、刪除用戶
• 權限控制: 基于角色的訪問控制 (RBAC)
• 密碼策略: 強制密碼復雜度和定期更換

角色和權限：

• 超級管理員:

  • 所有權限
  • 用戶管理
  • 系統配置

• 管理員:

  • 集群管理
  • Topic 管理
  • 告警配置

• 運維人員:

  • 監控查看
  • 告警處理
  • 任務管理

• 只讀用戶:

  • 查看監控數據
  • 查看告警信息
  • 無修改權限

用戶字段：

• 用戶名
• 角色
• 狀態 (啟用/禁用)
• 最后登錄時間
• 創建時間

功能特性：

• 多集群統一管理: 在一個平臺管理多個 Kafka 集群
• 環境隔離: 支持測試、預發、生產等不同環境的集群管理
• 集群切換: 快速在不同集群間切換查看和操作
• 集群對比: 對比不同集群的配置和性能指標

多環境集群管理：

• 生產環境 (Production):

  • 高可用配置
  • 嚴格的權限控制
  • 完整的審計日志
  • 實時性能監控
  • 7×24 告警響應

• 預發環境 (Staging):

  • 與生產環境配置一致
  • 用于上線前驗證
  • 支持壓力測試
  • 數據脫敏處理

• 測試環境 (Testing):

  • 開發團隊使用
  • 支持快速創建/刪除 Topic
  • 靈活的配置調整
  • 數據自動清理

• 開發環境 (Development):

  • 本地開發調試
  • 無告警限制
  • 支持實驗性功能

集群配置管理：

• 連接配置:

  • Broker 地址列表
  • Zookeeper 地址（如使用）
  • SASL/SSL 認證配置
  • 超時和重試參數

• 環境標識:

  • 集群名稱
  • 環境標簽 (dev/test/staging/prod)
  • 顏色標識（防止誤操作）
  • 備注說明

• 訪問控制:

  • 基于角色的集群訪問權限
  • 生產集群限制高危操作
  • 操作審批流程（生產環境）

集群狀態監控：

• 集群在線狀態
• Broker 總數和存活數
• Topic 和分區統計
• 消費者組數量
• 消息吞吐量
• 存儲使用情況
• 連接健康度

4.總結

總結來看，EFAK-AI 5.0.0 以分布式架構與 AI 智能分析為核心，全面提升了 Kafka 集群的可觀測性與運維效率。它不僅讓監控更實時、告警更精準，還讓問題分析與決策更智能。EFAK-AI 正在讓 Kafka 運維從被動監控邁向主動洞察，助力構建更高效、更智能的數據流生態。

5.結束語

這篇博客就和大家分享到這里，如果大家在研究學習的過程當中有什么問題，可以加群進行討論或發送郵件給我，我會盡我所能為您解答，與君共勉！

另外，博主出新書了《Hadoop與Spark大數據全景解析》、同時已出版的《深入理解Hive》、《Kafka并不難學》和《Hadoop大數據挖掘從入門到進階實戰》也可以和新書配套使用，喜歡的朋友或同學， 可以在公告欄那里點擊購買鏈接購買博主的書進行學習，在此感謝大家的支持。關注下面公眾號，根據提示，可免費獲取書籍的教學視頻。