Kafka 线上问题排查完整手册
本手册专门针对Apache Kafka生产环境故障排查，深入涵盖消息堆积、性能瓶颈、集群故障、分区问题、消费者异常等核心场景，提供丰富的中文背景知识、详细的排查步骤和实战经验总结。
📚 目录导航
- 🚨 快速诊断命令集合
- 🔍 一、Kafka集群状态全面检查
- 📈 二、消息堆积问题深度排查
- ⚡ 三、性能问题全方位分析
- 🔄 四、分区和副本问题处理
- 👥 五、消费者问题排查指南
- 🌐 六、网络和连接问题诊断
- 📊 七、专业监控脚本工具
- 🆘 八、紧急故障处理预案
- ⚙️ 九、生产环境配置优化
- 🎯 十、最佳实践和预防措施
🚨 快速诊断命令集合
当Kafka集群出现问题时，运维人员需要快速获取系统状态信息。以下命令可以在最短时间内了解Kafka集群的健康状况，为进一步的深度排查提供重要线索。
🔍 集群基础状态检查
# 1. 检查Kafka服务进程状态
# 这是最基础的检查，确认Kafka进程是否正常运行
ps aux | grep kafka
jps | grep Kafka # 使用Java进程查看工具# 2. 检查Kafka集群中的Broker列表
# 显示所有活跃的Broker节点，帮助识别是否有节点离线
kafka-broker-api-versions.sh --bootstrap-server localhost:9092# 3. 查看集群元数据信息
# 获取集群ID、控制器信息等核心元数据
kafka-metadata-shell.sh --snapshot /path/to/kafka-logs/__cluster_metadata-0# 4. 检查所有Topic列表和基本信息
# 快速了解集群中有哪些Topic以及它们的基本配置
kafka-topics.sh --bootstrap-server localhost:9092 --list
kafka-topics.sh --bootstrap-server localhost:9092 --describe# 5. 查看集群配置信息
# 检查重要的集群级别配置参数
kafka-configs.sh --bootstrap-server localhost:9092 --entity-type brokers --describe
📊 性能和负载快速评估
# 6. 检查消费者组状态（最重要的性能指标）
# 显示所有消费者组的消费进度和延迟情况
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --all-groups# 7. 查看Topic的消息生产和消费速率
# 实时监控消息的流入流出情况
kafka-run-class.sh kafka.tools.ConsumerPerformance \--bootstrap-server localhost:9092 \--topic test-topic \--messages 1000# 8. 检查磁盘使用情况
# Kafka对磁盘IO要求很高，磁盘问题是常见的性能瓶颈
df -h /kafka-logs # 检查Kafka日志目录磁盘使用率
du -sh /kafka-logs/* # 查看各Topic占用的磁盘空间# 9. 检查网络连接状况
# 网络是分布式系统的关键，需要确保节点间通信正常
netstat -tlnp | grep 9092 # 检查Kafka端口监听状态
ss -an | grep 9092 | wc -l # 统计当前连接数
🚨 故障快速定位
# 10. 查看Kafka错误日志
# 日志是排查问题的第一手资料，特别关注ERROR和WARN级别
tail -f /kafka-logs/server.log | grep -E "(ERROR|WARN)"
grep -i "exception\|error\|timeout" /kafka-logs/server.log | tail -20# 11. 检查JVM内存使用情况
# Kafka是Java应用，JVM内存问题会直接影响性能
jstat -gc $(pgrep -f kafka.Kafka) 2s # 每2秒显示GC情况
jmap -histo $(pgrep -f kafka.Kafka) | head -20 # 查看内存对象分布# 12. 系统资源使用检查
# CPU、内存、磁盘IO都可能成为瓶颈
top -p $(pgrep -f kafka.Kafka) # 查看Kafka进程资源使用
iostat -x 1 5 # 监控磁盘IO情况
free -h # 检查系统内存使用
🔍 一、Kafka集群状态全面检查
📖 Kafka集群架构背景知识
Apache Kafka是一个分布式流处理平台，其核心架构包含以下几个关键组件：
Broker（代理服务器）：Kafka集群中的每个服务器节点都是一个Broker，负责存储数据、处理生产者和消费者请求。一个健康的Kafka集群通常由3个或更多Broker组成，以确保高可用性和数据冗余。
Controller（控制器）：集群中有且仅有一个Broker会被选举为Controller，负责管理整个集群的元数据，包括Topic和分区的创建、删除，Broker的上线下线处理，以及分区Leader的选举等关键操作。Controller的健康状况直接影响整个集群的稳定性。
Zookeeper协调服务：在Kafka 2.8之前的版本中，Zookeeper负责存储集群元数据、进行Leader选举、维护配置信息等。虽然新版本引入了KRaft模式（无Zookeeper），但大部分生产环境仍在使用基于Zookeeper的架构。
分区和副本机制：每个Topic被分为多个Partition（分区），每个分区可以有多个Replica（副本）。副本分布在不同的Broker上，其中一个副本作为Leader处理读写请求，其他副本作为Follower进行数据同步。
🔧 集群状态深度检查流程
第一步：Broker节点健康状况检查
# 检查所有Broker的在线状态和基本信息
echo "=== Kafka Broker状态检查 ==="# 1. 获取集群中所有Broker的列表
kafka-broker-api-versions.sh --bootstrap-server localhost:9092 | \awk '/^[0-9]/ {print "Broker " $1 " (version: " $2 ")"}'# 2. 检查每个Broker的详细信息
for broker_id in $(kafka-broker-api-versions.sh --bootstrap-server localhost:9092 | awk '/^[0-9]/ {print $1}'); doecho "--- Broker $broker_id 详细信息 ---"# 获取Broker配置信息kafka-configs.sh --bootstrap-server localhost:9092 \--entity-type brokers --entity-name $broker_id --describe# 检查Broker的磁盘使用情况echo "Broker $broker_id 磁盘使用:"kafka-log-dirs.sh --bootstrap-server localhost:9092 \--describe --json | jq ".brokers[] | select(.broker == $broker_id)"
done# 3. 检查Broker之间的网络连通性
echo "=== Broker网络连通性测试 ==="
BROKERS=($(kafka-broker-api-versions.sh --bootstrap-server localhost:9092 | \awk '/^[0-9]/ {print $1}'))for broker in "${BROKERS[@]}"; doecho "测试到Broker $broker 的连接..."timeout 5 bash -c "</dev/tcp/localhost/9092" && echo "✅ 连接正常" || echo "❌ 连接失败"
done
第二步：Controller状态和选举机制检查
# Controller是Kafka集群的大脑，负责协调整个集群的工作
echo "=== Controller状态检查 ==="# 1. 识别当前的Controller节点
# Controller信息存储在Zookeeper中，也可以通过JMX指标获取
CONTROLLER_ID=$(kafka-run-class.sh kafka.admin.ZkSecurityMigrator \--zookeeper localhost:2181 2>/dev/null | grep controller | head -1)echo "当前Controller节点: $CONTROLLER_ID"# 2. 检查Controller的选举历史
# 频繁的Controller选举可能表明集群不稳定
echo "=== Controller选举历史 ==="
grep -r "Elected as the new controller" /kafka-logs/server.log | tail -10# 3. 检查Controller负责的Topic和分区信息
echo "=== Controller管理的资源统计 ==="
TOPIC_COUNT=$(kafka-topics.sh --bootstrap-server localhost:9092 --list | wc -l)
PARTITION_COUNT=$(kafka-topics.sh --bootstrap-server localhost:9092 --describe | \grep "PartitionCount" | awk '{sum += $2} END {print sum}')echo "管理的Topic数量: $TOPIC_COUNT"
echo "管理的分区总数: $PARTITION_COUNT"# 4. 监控Controller的性能指标
echo "=== Controller性能指标 ==="
# 通过JMX获取Controller相关指标（需要启用JMX）
if command -v jmxtrans >/dev/null 2>&1; thenecho "Controller选举速率: $(jmxtrans_query kafka.controller:type=ControllerStats,name=LeaderElectionRateAndTimeMs)"echo "分区状态变更速率: $(jmxtrans_query kafka.controller:type=ControllerStats,name=UncleanLeaderElectionsPerSec)"
fi
第三步：Zookeeper连接和元数据完整性检查
# Zookeeper是Kafka的协调服务，其健康状况直接影响Kafka集群
echo "=== Zookeeper连接状态检查 ==="# 1. 检查Zookeeper服务状态
echo "1. Zookeeper服务状态:"
echo "stat" | nc localhost 2181 | head -10# 2. 验证Kafka在Zookeeper中的注册信息
echo "2. Kafka集群在Zookeeper中的注册信息:"
kafka-run-class.sh kafka.admin.ZkSecurityMigrator \--zookeeper localhost:2181 --list 2>/dev/null | grep -E "(brokers|topics|consumers)"# 3. 检查Zookeeper中的Kafka元数据一致性
echo "3. 检查元数据一致性:"
# 比较Zookeeper中的Broker信息与实际运行的Broker
ZK_BROKERS=$(echo "ls /brokers/ids" | kafka-run-class.sh kafka.admin.ZkCli \--server localhost:2181 2>/dev/null | grep "^\[" | tr -d '[],' | tr ' ' '\n')
RUNNING_BROKERS=$(kafka-broker-api-versions.sh --bootstrap-server localhost:9092 | awk '/^[0-9]/ {print $1}')echo "Zookeeper中注册的Broker: $ZK_BROKERS"
echo "实际运行的Broker: $RUNNING_BROKERS"# 4. 检查Zookeeper连接延迟
echo "4. Zookeeper连接延迟测试:"
time echo "stat" | nc localhost 2181 >/dev/null
第四步：集群整体健康度评估
# 综合评估集群的健康状况
echo "=== Kafka集群健康度评估报告 ==="# 1. 计算集群可用性指标
TOTAL_BROKERS=$(kafka-broker-api-versions.sh --bootstrap-server localhost:9092 | wc -l)
HEALTHY_BROKERS=0for broker in $(kafka-broker-api-versions.sh --bootstrap-server localhost:9092 | awk '{print $1}'); doif timeout 3 bash -c "</dev/tcp/localhost/9092" 2>/dev/null; then((HEALTHY_BROKERS++))fi
doneAVAILABILITY_PERCENTAGE=$((HEALTHY_BROKERS * 100 / TOTAL_BROKERS))
echo "集群可用性: $AVAILABILITY_PERCENTAGE% ($HEALTHY_BROKERS/$TOTAL_BROKERS 个Broker正常)"# 2. 检查Topic分区的Leader分布均衡性
echo "=== Topic分区Leader分布分析 ==="
kafka-topics.sh --bootstrap-server localhost:9092 --describe | \
grep -E "Leader: [0-9]+" | awk '{print $6}' | sort | uniq -c | \
awk '{printf "Broker %s: %d 个Leader分区\n", $2, $1}'# 3. 评估存储使用情况
echo "=== 存储使用情况评估 ==="
TOTAL_DISK_USAGE=$(du -sh /kafka-logs 2>/dev/null | awk '{print $1}')
AVAILABLE_DISK=$(df -h /kafka-logs 2>/dev/null | awk 'NR==2 {print $4}')echo "Kafka日志目录使用: $TOTAL_DISK_USAGE"
echo "磁盘剩余空间: $AVAILABLE_DISK"# 4. 生成健康度总结
echo "=== 集群健康度总结 ==="
if [ $AVAILABILITY_PERCENTAGE -eq 100 ]; thenecho "✅ 集群状态: 健康"
elif [ $AVAILABILITY_PERCENTAGE -ge 67 ]; thenecho "⚠️ 集群状态: 部分节点异常，但仍可提供服务"
elseecho "❌ 集群状态: 严重故障，需要立即处理"
fi# 输出关键建议
echo "=== 运维建议 ==="
if [ $AVAILABILITY_PERCENTAGE -lt 100 ]; thenecho "- 立即检查离线的Broker节点"echo "- 验证网络连接和防火墙配置"echo "- 检查磁盘空间和系统资源"
fiecho "- 定期监控Controller选举频率"
echo "- 确保Zookeeper集群稳定运行"
echo "- 监控磁盘使用率，及时清理旧数据"
📈 二、消息堆积问题深度排查
📚 消息堆积问题的业务背景
消息堆积（Message Lag）是Kafka生产环境中最常见也最关键的问题之一。它反映了消费者处理消息的速度跟不上生产者发送消息的速度，导致未处理消息在Topic中持续累积。
消息堆积的业务影响：
- 数据处理延迟：实时业务（如监控告警、订单处理）可能因延迟处理造成业务损失
- 内存压力增加：消费者需要维护更多的offset信息和连接状态
- 磁盘空间占用：未被消费的消息会持续占用磁盘空间，直到达到retention时间
- 系统雪崩风险：严重的消息堆积可能导致整个消息处理链路瘫痪
消息堆积的常见原因：
- 消费者性能瓶颈：消费者处理业务逻辑耗时过长，单位时间内处理消息数量不足
- 消费者异常下线：消费者进程崩溃、网络中断或部署问题导致消费中断
- 分区数配置不合理：分区数过少限制了并发消费能力
- 生产者流量突增：业务高峰期或异常情况导致消息生产速率骤增
- 消费者配置不当：fetch.max.wait.ms、max.poll.records等参数配置不合理
🔍 消息堆积排查详细流程
第一步：全面识别消息堆积情况
#!/bin/bash
# Kafka消息堆积全面分析脚本echo "=== Kafka消息堆积深度分析报告 $(date) ==="# 1. 获取所有消费者组的概况
echo "1. 消费者组概况分析"
echo "================================"# 列出所有消费者组并统计
CONSUMER_GROUPS=$(kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list)
GROUP_COUNT=$(echo "$CONSUMER_GROUPS" | wc -l)
echo "总消费者组数量: $GROUP_COUNT"# 2. 分析每个消费者组的消费状态
echo -e "\n2. 各消费者组详细状态分析"
echo "================================"for group in $CONSUMER_GROUPS; doecho "--- 消费者组: $group ---"# 获取该组的详细消费信息CONSUMER_INFO=$(kafka-consumer-groups.sh --bootstrap-server localhost:9092 \--describe --group $group 2>/dev/null)if [ -n "$CONSUMER_INFO" ]; then# 计算总的LagTOTAL_LAG=$(echo "$CONSUMER_INFO" | awk 'NR>1 && $5 != "-" {sum += $5} END {print sum+0}')# 获取Topic和分区信息TOPICS=$(echo "$CONSUMER_INFO" | awk 'NR>1 {print $1}' | sort | uniq)PARTITION_COUNT=$(echo "$CONSUMER_INFO" | awk 'NR>1' | wc -l)echo " Topic列表: $(echo $TOPICS | tr '\n' ' ')"echo " 分区数量: $PARTITION_COUNT"echo " 总消息堆积: $TOTAL_LAG"# 分析消费状态if [ "$TOTAL_LAG" -gt 100000 ]; thenecho " ⚠️ 状态: 严重堆积 (>100K)"elif [ "$TOTAL_LAG" -gt 10000 ]; thenecho " ⚠️ 状态: 中等堆积 (>10K)"elif [ "$TOTAL_LAG" -gt 1000 ]; thenecho " ⚠️ 状态: 轻微堆积 (>1K)"elseecho " ✅ 状态: 消费正常"fi# 检查是否有消费者实例在线ACTIVE_CONSUMERS=$(echo "$CONSUMER_INFO" | awk 'NR>1 && $7 != "-"' | wc -l)TOTAL_PARTITIONS=$(echo "$CONSUMER_INFO" | awk 'NR>1' | wc -l)echo " 活跃消费者: $ACTIVE_CONSUMERS/$TOTAL_PARTITIONS 个分区有消费者"if [ "$ACTIVE_CONSUMERS" -eq 0 ]; thenecho " ❌ 警告: 没有活跃的消费者实例!"fielseecho " ❌ 无法获取消费者组信息，可能已删除或无权限"fiecho ""
done# 3. 按Topic分析消息堆积情况
echo "3. 按Topic分析消息堆积分布"
echo "================================"# 创建临时文件存储分析结果
TEMP_FILE="/tmp/kafka_lag_analysis_$(date +%s).txt"# 收集所有消费者组的详细信息
for group in $CONSUMER_GROUPS; dokafka-consumer-groups.sh --bootstrap-server localhost:9092 \--describe --group $group 2>/dev/null >> $TEMP_FILE
done# 按Topic统计总的消息堆积
if [ -f "$TEMP_FILE" ]; thenecho "Topic消息堆积统计:"awk 'NR>1 && $1 != "" && $5 != "-" {topic_lag[$1] += $5topic_partitions[$1]++} END {for (topic in topic_lag) {printf " %-30s 堆积: %10d, 分区数: %3d\n", topic, topic_lag[topic], topic_partitions[topic]}}' $TEMP_FILE | sort -k3 -nrrm -f $TEMP_FILE
fi
第二步：消息堆积根因分析
#!/bin/bash
# 消息堆积根因深度分析脚本echo "=== 消息堆积根因分析 ==="# 1. 分析生产者消息生产速率
echo "1. 生产者性能分析"
echo "================================"# 获取所有Topic的消息生产情况
TOPICS=$(kafka-topics.sh --bootstrap-server localhost:9092 --list)for topic in $TOPICS; doecho "--- Topic: $topic ---"# 获取Topic的分区和副本信息TOPIC_INFO=$(kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic $topic)PARTITION_COUNT=$(echo "$TOPIC_INFO" | grep "PartitionCount" | awk '{print $2}')REPLICATION_FACTOR=$(echo "$TOPIC_INFO" | grep "ReplicationFactor" | awk '{print $4}')echo " 分区数: $PARTITION_COUNT, 副本因子: $REPLICATION_FACTOR"# 检查各分区的消息数量分布echo " 分区消息分布:"kafka-run-class.sh kafka.tools.GetOffsetShell \--broker-list localhost:9092 --topic $topic --time -1 | \awk -F: '{printf " 分区 %s: %s 条消息\n", $1, $2}'echo ""
done# 2. 分析消费者性能瓶颈
echo "2. 消费者性能瓶颈分析"
echo "================================"for group in $(kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list); doecho "--- 消费者组: $group ---"# 获取详细的消费信息CONSUMER_DETAIL=$(kafka-consumer-groups.sh --bootstrap-server localhost:9092 \--describe --group $group 2>/dev/null)if [ -n "$CONSUMER_DETAIL" ]; then# 分析消费者的消费模式echo " 消费详情分析:"echo "$CONSUMER_DETAIL" | awk 'BEGIN {printf " %-20s %-10s %-15s %-15s %-10s %-15s\n", "Topic", "分区", "当前Offset", "Log End Offset", "Lag", "消费者ID"print " " sprintf("%*s", 95, "") | "tr ' ' '-'"}NR>1 && $1 != "" {lag = ($5 == "-") ? "N/A" : $5consumer = ($7 == "-") ? "无消费者" : substr($7, 1, 12) "..."printf " %-20s %-10s %-15s %-15s %-10s %-15s\n", $1, $2, $3, $4, lag, consumer# 统计异常情况if ($7 == "-") no_consumer++if ($5 != "-" && $5 > 1000) high_lag++}END {if (no_consumer > 0) print " ⚠️ 发现 " no_consumer " 个分区没有消费者"if (high_lag > 0) print " ⚠️ 发现 " high_lag " 个分区消息堆积严重 (>1000)"}'# 检查消费者组的配置echo " 消费者组配置检查:"GROUP_CONFIG=$(kafka-consumer-groups.sh --bootstrap-server localhost:9092 \--describe --group $group --verbose 2>/dev/null | grep -A 20 "GROUP INFORMATION")if [ -n "$GROUP_CONFIG" ]; thenecho "$GROUP_CONFIG" | grep -E "(protocol|assignment)"fifiecho ""
done# 3. 系统资源瓶颈分析
echo "3. 系统资源瓶颈分析"
echo "================================"# 检查Kafka进程的系统资源使用
KAFKA_PID=$(pgrep -f kafka.Kafka | head -1)if [ -n "$KAFKA_PID" ]; thenecho "Kafka进程资源使用情况 (PID: $KAFKA_PID):"# CPU使用率CPU_USAGE=$(ps -p $KAFKA_PID -o %cpu --no-headers)echo " CPU使用率: ${CPU_USAGE}%"# 内存使用MEMORY_INFO=$(ps -p $KAFKA_PID -o %mem,rss --no-headers)echo " 内存使用: $MEMORY_INFO"# 文件描述符使用FD_COUNT=$(ls /proc/$KAFKA_PID/fd 2>/dev/null | wc -l)FD_LIMIT=$(cat /proc/$KAFKA_PID/limits 2>/dev/null | grep "Max open files" | awk '{print $4}')echo " 文件描述符: $FD_COUNT/$FD_LIMIT"# JVM垃圾回收情况if command -v jstat >/dev/null 2>&1; thenecho " JVM GC统计:"jstat -gc $KAFKA_PID | awk 'NR==1 {print " " $0}NR==2 {printf " %s\n", $0printf " Eden使用率: %.1f%%, Old使用率: %.1f%%\n", ($3/($1+$2+$3))*100, ($8/($7+$8))*100}'fi
fi# 4. 磁盘IO和网络分析
echo "4. 磁盘IO和网络性能分析"
echo "================================"# 磁盘IO统计
echo "磁盘IO情况:"
if command -v iostat >/dev/null 2>&1; theniostat -x 1 3 | grep -A 1 "Device" | tail -n +3 | \awk '{printf " 设备: %s, 读取: %s KB/s, 写入: %s KB/s, 利用率: %s%%\n", $1, $6, $7, $10}'
elseecho " 请安装sysstat包以获取详细的IO统计"
fi# 网络连接统计
echo "网络连接情况:"
KAFKA_CONNECTIONS=$(netstat -an 2>/dev/null | grep :9092 | wc -l)
echo " Kafka端口(9092)连接数: $KAFKA_CONNECTIONS"# 检查网络延迟
echo "网络延迟测试:"
BROKERS=$(kafka-broker-api-versions.sh --bootstrap-server localhost:9092 2>/dev/null | awk '{print $1}')
for broker in $BROKERS; doLATENCY=$(ping -c 3 localhost 2>/dev/null | tail -1 | awk -F'/' '{print $5}')echo " 到Broker $broker 的延迟: ${LATENCY}ms"
done
第三步：消息堆积影响评估和解决方案
#!/bin/bash
# 消息堆积影响评估和解决方案生成脚本echo "=== 消息堆积影响评估和解决方案 ==="# 1. 业务影响评估
echo "1. 业务影响评估"
echo "================================"# 定义不同堆积程度的阈值
CRITICAL_LAG=100000 # 严重堆积阈值
WARNING_LAG=10000 # 警告堆积阈值
NORMAL_LAG=1000 # 正常堆积阈值TOTAL_CRITICAL=0
TOTAL_WARNING=0
TOTAL_NORMAL=0# 分析各消费者组的影响程度
for group in $(kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list); doLAG_INFO=$(kafka-consumer-groups.sh --bootstrap-server localhost:9092 \--describe --group $group 2>/dev/null)if [ -n "$LAG_INFO" ]; thenTOTAL_LAG=$(echo "$LAG_INFO" | awk 'NR>1 && $5 != "-" {sum += $5} END {print sum+0}')if [ "$TOTAL_LAG" -gt "$CRITICAL_LAG" ]; thenecho "🔴 严重影响 - 消费者组: $group, 堆积: $TOTAL_LAG 条消息"echo " 建议: 立即扩容消费者实例或优化消费逻辑"((TOTAL_CRITICAL++))elif [ "$TOTAL_LAG" -gt "$WARNING_LAG" ]; thenecho "🟡 中等影响 - 消费者组: $group, 堆积: $TOTAL_LAG 条消息"echo " 建议: 监控消费速率趋势，准备扩容方案"((TOTAL_WARNING++))elif [ "$TOTAL_LAG" -gt "$NORMAL_LAG" ]; thenecho "🟢 轻微影响 - 消费者组: $group, 堆积: $TOTAL_LAG 条消息"echo " 建议: 持续观察，暂无需特殊处理"((TOTAL_NORMAL++))fifi
doneecho ""
echo "影响程度统计:"
echo " 严重影响的消费者组: $TOTAL_CRITICAL 个"
echo " 中等影响的消费者组: $TOTAL_WARNING 个"
echo " 轻微影响的消费者组: $TOTAL_NORMAL 个"# 2. 自动化解决方案建议
echo -e "\n2. 针对性解决方案建议"
echo "================================"# 生成具体的解决方案
cat << 'EOF'
根据分析结果，推荐以下解决方案：🔧 立即执行方案 (适用于严重堆积):
1. 紧急扩容消费者实例# 快速启动更多消费者实例for i in {1..3}; donohup kafka-console-consumer.sh \--bootstrap-server localhost:9092 \--group emergency-consumer-$i \--topic your-topic > /dev/null 2>&1 &done2. 临时增加分区数 (谨慎操作，会影响消息顺序)kafka-topics.sh --bootstrap-server localhost:9092 \--alter --topic your-topic --partitions 203. 调整消费者配置优化性能# 增加fetch大小，减少网络开销max.poll.records=1000fetch.min.bytes=50000fetch.max.wait.ms=100🔧 中期优化方案 (适用于中等堆积):
1. 优化消费者业务逻辑- 异步处理非关键业务- 批量处理消息- 使用连接池减少资源消耗2. 调整Topic配置# 增加分区数以支持更高并发kafka-topics.sh --bootstrap-server localhost:9092 \--alter --topic your-topic --partitions 50# 调整segment大小优化磁盘IOkafka-configs.sh --bootstrap-server localhost:9092 \--entity-type topics --entity-name your-topic \--alter --add-config segment.bytes=10737418243. 消费者性能调优# JVM参数优化export KAFKA_HEAP_OPTS="-Xmx4G -Xms4G"export KAFKA_JVM_PERFORMANCE_OPTS="-XX:+UseG1GC -XX:MaxGCPauseMillis=20"🔧 长期预防方案:
1. 建立监控告警体系- 设置消息堆积阈值告警- 监控消费者实例健康状态- 跟踪Topic消息生产速率2. 容量规划和弹性伸缩- 根据业务增长预估消息量- 建立自动化扩缩容机制- 定期进行性能压测3. 业务架构优化- 合理设计Topic分区策略- 实现消费者的优雅重启机制- 建立消息处理失败的重试和降级机制
EOF# 3. 生成监控脚本
echo -e "\n3. 持续监控脚本"
echo "================================"cat > /tmp/kafka_lag_monitor.sh << 'EOF'
#!/bin/bash
# Kafka消息堆积持续监控脚本
# 建议添加到crontab中每分钟执行一次ALERT_THRESHOLD=50000 # 告警阈值
LOG_FILE="/var/log/kafka-lag-monitor.log"timestamp=$(date '+%Y-%m-%d %H:%M:%S')for group in $(kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list 2>/dev/null); dototal_lag=$(kafka-consumer-groups.sh --bootstrap-server localhost:9092 \--describe --group $group 2>/dev/null | \awk 'NR>1 && $5 != "-" {sum += $5} END {print sum+0}')if [ "$total_lag" -gt "$ALERT_THRESHOLD" ]; thenmessage="[$timestamp] ALERT: Consumer group $group has high lag: $total_lag messages"echo "$message" >> $LOG_FILE# 这里可以集成邮件、短信、钉钉等告警方式# curl -X POST "your-alert-webhook" -d "$message"echo "$message"fi
done
EOFchmod +x /tmp/kafka_lag_monitor.sh
echo "监控脚本已生成: /tmp/kafka_lag_monitor.sh"
echo "建议添加到crontab: */1 * * * * /tmp/kafka_lag_monitor.sh"echo -e "\n=== 消息堆积分析完成 ==="
⚡ 三、性能问题全方位分析
📖 Kafka性能问题的技术背景
Kafka作为高吞吐量的分布式流处理平台，其性能表现直接影响整个数据流处理链路的效率。性能问题往往是多个因素相互作用的结果，需要从系统架构、配置参数、硬件资源、网络环境等多个维度进行综合分析。
Kafka性能的关键指标：
- 吞吐量（Throughput）：单位时间内处理的消息数量，通常以消息/秒或MB/秒衡量
- 延迟（Latency）：从消息生产到消费完成的端到端时间
- 可用性（Availability）：集群能够正常提供服务的时间百分比
- 资源利用率：CPU、内存、磁盘IO、网络带宽的使用效率
常见的性能瓶颈类型：
- 磁盘IO瓶颈：Kafka大量使用磁盘进行数据持久化，磁盘性能直接影响整体性能
- 网络带宽限制：生产者和消费者与Broker之间的网络传输能力
- JVM垃圾回收：不合理的JVM参数配置导致频繁的Full GC
- CPU计算瓶颈：消息压缩、解压缩、序列化等操作占用大量CPU资源
- 配置参数不当：batch.size、linger.ms、buffer.memory等参数配置不合理
🔍 性能问题深度排查流程
第一步：生产者性能瓶颈分析
#!/bin/bash
# Kafka生产者性能深度分析脚本echo "=== Kafka生产者性能分析报告 ==="# 1. 生产者吞吐量测试
echo "1. 生产者吞吐量基准测试"
echo "================================"# 创建测试Topic
TEST_TOPIC="performance-test-$(date +%s)"
kafka-topics.sh --bootstrap-server localhost:9092 \--create --topic $TEST_TOPIC \--partitions 12 --replication-factor 3echo "测试Topic: $TEST_TOPIC"# 执行生产者性能测试
echo "开始生产者性能测试 (100万条消息, 1KB大小)..."
PRODUCER_RESULT=$(kafka-producer-perf-test.sh \--topic $TEST_TOPIC \--num-records 1000000 \--record-size 1024 \--throughput 100000 \--producer-props bootstrap.servers=localhost:9092 \compression.type=lz4 \batch.size=32768 \linger.ms=5 \buffer.memory=67108864)echo "生产者测试结果:"
echo "$PRODUCER_RESULT" | awk '
/records sent/ {printf " 发送消息数: %s 条\n", $1printf " 平均吞吐量: %s 条/秒\n", $7printf " 平均带宽: %s MB/秒\n", $9printf " 平均延迟: %s ms\n", $12printf " 最大延迟: %s ms\n", $15
}'# 2. 分析不同配置对性能的影响
echo -e "\n2. 生产者配置参数性能影响分析"
echo "================================"# 测试不同batch.size的影响
echo "测试不同batch.size配置的性能影响:"
for batch_size in 1024 8192 32768 65536; doecho " 测试 batch.size=$batch_size..."BATCH_RESULT=$(kafka-producer-perf-test.sh \--topic $TEST_TOPIC \--num-records 100000 \--record-size 1024 \--throughput -1 \--producer-props bootstrap.servers=localhost:9092 \batch.size=$batch_size \linger.ms=5 2>/dev/null)THROUGHPUT=$(echo "$BATCH_RESULT" | grep "records sent" | awk '{print $7}')printf " batch.size=%6d -> 吞吐量: %8s 条/秒\n" $batch_size "$THROUGHPUT"
done# 测试不同linger.ms的影响
echo -e "\n测试不同linger.ms配置的性能影响:"
for linger_ms in 0 1 5 10 50; doecho " 测试 linger.ms=$linger_ms..."LINGER_RESULT=$(kafka-producer-perf-test.sh \--topic $TEST_TOPIC \--num-records 100000 \--record-size 1024 \--throughput -1 \--producer-props bootstrap.servers=localhost:9092 \batch.size=32768 \linger.ms=$linger_ms 2>/dev/null)THROUGHPUT=$(echo "$LINGER_RESULT" | grep "records sent" | awk '{print $7}')LATENCY=$(echo "$LINGER_RESULT" | grep "records sent" | awk '{print $12}')printf " linger.ms=%3d -> 吞吐量: %8s 条/秒, 延迟: %6s ms\n" $linger_ms "$THROUGHPUT" "$LATENCY"
done# 3. 压缩算法性能对比
echo -e "\n3. 消息压缩算法性能对比"
echo "================================"for compression in none gzip snappy lz4 zstd; doecho " 测试压缩算法: $compression..."COMPRESSION_RESULT=$(kafka-producer-perf-test.sh \--topic $TEST_TOPIC \--num-records 50000 \--record-size 1024 \--throughput -1 \--producer-props bootstrap.servers=localhost:9092 \compression.type=$compression \batch.size=32768 \linger.ms=5 2>/dev/null)if [ -n "$COMPRESSION_RESULT" ]; thenTHROUGHPUT=$(echo "$COMPRESSION_RESULT" | grep "records sent" | awk '{print $7}')BANDWIDTH=$(echo "$COMPRESSION_RESULT" | grep "records sent" | awk '{print $9}')printf " %-8s -> 吞吐量: %8s 条/秒, 带宽: %6s MB/秒\n" $compression "$THROUGHPUT" "$BANDWIDTH"fi
done# 清理测试Topic
kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic $TEST_TOPICecho -e "\n=== 生产者性能分析完成 ==="
第二步：消费者性能瓶颈分析
#!/bin/bash
# Kafka消费者性能深度分析脚本echo "=== Kafka消费者性能分析报告 ==="# 1. 消费者吞吐量测试
echo "1. 消费者吞吐量基准测试"
echo "================================"# 首先创建测试数据
TEST_TOPIC="consumer-perf-test-$(date +%s)"
kafka-topics.sh --bootstrap-server localhost:9092 \--create --topic $TEST_TOPIC \--partitions 12 --replication-factor 3echo "准备测试数据..."
kafka-producer-perf-test.sh \--topic $TEST_TOPIC \--num-records 1000000 \--record-size 1024 \--throughput -1 \--producer-props bootstrap.servers=localhost:9092 > /dev/nullecho "开始消费者性能测试..."# 执行消费者性能测试
CONSUMER_RESULT=$(kafka-consumer-perf-test.sh \--bootstrap-server localhost:9092 \--topic $TEST_TOPIC \--messages 1000000 \--threads 1 \--consumer-props fetch.min.bytes=1 \fetch.max.wait.ms=500 \max.poll.records=500)echo "消费者测试结果:"
echo "$CONSUMER_RESULT" | awk -F, '
NR==1 {print " " $0}
NR==2 {printf " 消费消息数: %s 条\n", $4printf " 平均吞吐量: %s 条/秒\n", $1printf " 平均带宽: %s MB/秒\n", $2printf " 消费耗时: %s 秒\n", $6
}'# 2. 不同fetch参数对性能的影响
echo -e "\n2. 消费者fetch参数性能影响分析"
echo "================================"# 重新生成测试数据
kafka-producer-perf-test.sh \--topic $TEST_TOPIC \--num-records 500000 \--record-size 1024 \--throughput -1 \--producer-props bootstrap.servers=localhost:9092 > /dev/nullecho "测试不同fetch.min.bytes配置:"
for fetch_min_bytes in 1 1024 10240 102400; doecho " 测试 fetch.min.bytes=$fetch_min_bytes..."FETCH_RESULT=$(kafka-consumer-perf-test.sh \--bootstrap-server localhost:9092 \--topic $TEST_TOPIC \--messages 100000 \--threads 1 \--consumer-props fetch.min.bytes=$fetch_min_bytes \fetch.max.wait.ms=500 \max.poll.records=500 2>/dev/null)THROUGHPUT=$(echo "$FETCH_RESULT" | tail -1 | awk -F, '{print $1}')printf " fetch.min.bytes=%6d -> 吞吐量: %8s 条/秒\n" $fetch_min_bytes "$THROUGHPUT"
doneecho -e "\n测试不同max.poll.records配置:"
for max_poll_records in 100 500 1000 2000; doecho " 测试 max.poll.records=$max_poll_records..."POLL_RESULT=$(kafka-consumer-perf-test.sh \--bootstrap-server localhost:9092 \--topic $TEST_TOPIC \--messages 100000 \--threads 1 \--consumer-props fetch.min.bytes=1024 \fetch.max.wait.ms=500 \max.poll.records=$max_poll_records 2>/dev/null)THROUGHPUT=$(echo "$POLL_RESULT" | tail -1 | awk -F, '{print $1}')printf " max.poll.records=%4d -> 吞吐量: %8s 条/秒\n" $max_poll_records "$THROUGHPUT"
done# 3. 多线程消费性能测试
echo -e "\n3. 多线程消费性能对比"
echo "================================"# 重新生成测试数据
kafka-producer-perf-test.sh \--topic $TEST_TOPIC \--num-records 500000 \--record-size 1024 \--throughput -1 \--producer-props bootstrap.servers=localhost:9092 > /dev/nullecho "测试不同线程数的消费性能:"
for threads in 1 2 4 8; doecho " 测试 threads=$threads..."THREAD_RESULT=$(kafka-consumer-perf-test.sh \--bootstrap-server localhost:9092 \--topic $TEST_TOPIC \--messages 200000 \--threads $threads \--consumer-props fetch.min.bytes=1024 \fetch.max.wait.ms=500 \max.poll.records=500 2>/dev/null)THROUGHPUT=$(echo "$THREAD_RESULT" | tail -1 | awk -F, '{print $1}')printf " threads=%2d -> 吞吐量: %8s 条/秒\n" $threads "$THROUGHPUT"
done# 清理测试Topic
kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic $TEST_TOPICecho -e "\n=== 消费者性能分析完成 ==="
第三步：Broker和集群性能分析
#!/bin/bash
# Kafka Broker和集群性能深度分析脚本echo "=== Kafka Broker和集群性能分析报告 ==="# 1. Broker系统资源使用分析
echo "1. Broker系统资源使用分析"
echo "================================"KAFKA_PID=$(pgrep -f kafka.Kafka | head -1)if [ -n "$KAFKA_PID" ]; thenecho "Kafka进程 (PID: $KAFKA_PID) 资源使用详情:"# CPU使用情况echo " CPU使用分析:"CPU_INFO=$(ps -p $KAFKA_PID -o pid,ppid,%cpu,%mem,wchan,cmd --no-headers)echo " $CPU_INFO"# 线程数统计THREAD_COUNT=$(ps -eLf | grep $KAFKA_PID | wc -l)echo " 活跃线程数: $THREAD_COUNT"# 内存使用详情echo " 内存使用分析:"if [ -f "/proc/$KAFKA_PID/status" ]; thengrep -E "(VmSize|VmRSS|VmHWM|Threads)" /proc/$KAFKA_PID/status | \awk '{printf " %-15s %s\n", $1, $2 " " $3}'fi# JVM内存和GC分析echo " JVM内存和垃圾回收分析:"if command -v jstat >/dev/null 2>&1; thenecho " JVM堆内存使用情况:"jstat -gc $KAFKA_PID | awk 'NR==1 {printf " %-8s %-8s %-8s %-8s %-8s %-8s %-8s %-8s %-8s\n", "S0C", "S1C", "S0U", "S1U", "EC", "EU", "OC", "OU", "MC"}NR==2 {printf " %-8.1f %-8.1f %-8.1f %-8.1f %-8.1f %-8.1f %-8.1f %-8.1f %-8.1f\n", $1/1024, $2/1024, $3/1024, $4/1024, $5/1024, $6/1024, $7/1024, $8/1024, $9/1024printf " Eden使用率: %.1f%%, Old使用率: %.1f%%\n", ($6/($5+$6))*100, ($8/($7+$8))*100}'echo " GC统计信息:"jstat -gccapacity $KAFKA_PID | awk 'NR==1 {print " " $0}NR==2 {printf " 新生代容量: %.1f MB, 老年代容量: %.1f MB\n", ($2+$3+$6)/1024, ($7+$8)/1024}'fi# 文件描述符使用echo " 文件描述符使用:"FD_COUNT=$(ls /proc/$KAFKA_PID/fd 2>/dev/null | wc -l)FD_LIMIT=$(cat /proc/$KAFKA_PID/limits 2>/dev/null | grep "Max open files" | awk '{print $4}')FD_USAGE_PERCENT=$((FD_COUNT * 100 / FD_LIMIT))echo " 使用数量: $FD_COUNT/$FD_LIMIT ($FD_USAGE_PERCENT%)"if [ $FD_USAGE_PERCENT -gt 80 ]; thenecho " ⚠️ 警告: 文件描述符使用率过高!"fi
fi# 2. 磁盘IO性能分析
echo -e "\n2. 磁盘IO性能分析"
echo "================================"# 获取Kafka日志目录
KAFKA_LOG_DIR=$(grep -r "log.dirs" /opt/kafka/config/server.properties 2>/dev/null | cut -d'=' -f2 | tr -d ' ')
if [ -z "$KAFKA_LOG_DIR" ]; thenKAFKA_LOG_DIR="/tmp/kafka-logs" # 默认目录
fiecho "Kafka日志目录: $KAFKA_LOG_DIR"# 磁盘使用情况
echo " 磁盘空间使用:"
df -h $KAFKA_LOG_DIR | awk 'NR==2 {printf " 总空间: %s, 已用: %s, 可用: %s, 使用率: %s\n", $2, $3, $4, $5
}'# 检查各Topic的磁盘使用
echo " 各Topic磁盘使用排行 (前10名):"
if [ -d "$KAFKA_LOG_DIR" ]; thendu -sh $KAFKA_LOG_DIR/* 2>/dev/null | \sort -rh | head -10 | \awk '{printf " %-40s %s\n", $2, $1}'
fi# 磁盘IO统计
echo " 磁盘IO性能统计:"
if command -v iostat >/dev/null 2>&1; then# 获取包含Kafka日志目录的磁盘设备DISK_DEVICE=$(df $KAFKA_LOG_DIR | awk 'NR==2 {print $1}' | sed 's/[0-9]*$//')echo " 监控磁盘设备: $DISK_DEVICE"iostat -x 1 3 | grep -A 1 "Device" | grep "$DISK_DEVICE" | tail -1 | \awk '{printf " 读取: %s KB/s, 写入: %s KB/s, 平均等待: %s ms, 利用率: %s%%\n", $6, $7, $9, $10}'# 评估IO性能IO_UTIL=$(iostat -x 1 1 | grep "$DISK_DEVICE" | awk '{print $10}' | cut -d'.' -f1)if [ "$IO_UTIL" -gt 80 ]; thenecho " ⚠️ 警告: 磁盘IO利用率过高 ($IO_UTIL%)!"elif [ "$IO_UTIL" -gt 60 ]; thenecho " ⚠️ 注意: 磁盘IO利用率较高 ($IO_UTIL%)"elseecho " ✅ 磁盘IO利用率正常 ($IO_UTIL%)"fi
fi# 3. 网络性能分析
echo -e "\n3. 网络性能分析"
echo "================================"# Kafka网络连接统计
echo " Kafka网络连接分析:"
KAFKA_PORT=9092
TOTAL_CONNECTIONS=$(netstat -an 2>/dev/null | grep ":$KAFKA_PORT " | wc -l)
ESTABLISHED_CONNECTIONS=$(netstat -an 2>/dev/null | grep ":$KAFKA_PORT " | grep ESTABLISHED | wc -l)
TIME_WAIT_CONNECTIONS=$(netstat -an 2>/dev/null | grep ":$KAFKA_PORT " | grep TIME_WAIT | wc -l)echo " 总连接数: $TOTAL_CONNECTIONS"
echo " 已建立连接: $ESTABLISHED_CONNECTIONS"
echo " TIME_WAIT连接: $TIME_WAIT_CONNECTIONS"# 网络流量统计
echo " 网络流量统计:"
if command -v sar >/dev/null 2>&1; then# 使用sar命令获取网络统计NETWORK_STATS=$(sar -n DEV 1 1 | grep -v "^$" | tail -n +3 | grep -E "(eth0|ens|enp)" | head -1)if [ -n "$NETWORK_STATS" ]; thenecho " $NETWORK_STATS" | awk '{printf " 接收: %.2f KB/s, 发送: %.2f KB/s, 接收包: %.0f pps, 发送包: %.0f pps\n", $5, $6, $3, $4}'fi
elif command -v ifstat >/dev/null 2>&1; thenecho " $(ifstat -i eth0 1 1 2>/dev/null | tail -1)"
elseecho " 请安装sysstat或ifstat包以获取详细的网络统计"
fi# 检查网络延迟
echo " 集群内网络延迟测试:"
BROKERS=$(kafka-broker-api-versions.sh --bootstrap-server localhost:9092 2>/dev/null | awk '{print $1}')
for broker in $BROKERS; do# 假设broker ID对应不同的主机if ping -c 3 localhost >/dev/null 2>&1; thenLATENCY=$(ping -c 3 localhost 2>/dev/null | tail -1 | awk -F'/' '{print $5}')echo " 到Broker $broker: ${LATENCY}ms"fi
done# 4. Topic和分区性能分析
echo -e "\n4. Topic和分区性能分析"
echo "================================"echo " Topic分区分布和性能统计:"
TOPICS=$(kafka-topics.sh --bootstrap-server localhost:9092 --list 2>/dev/null)for topic in $TOPICS; doif [ "$topic" != "" ]; thenecho " --- Topic: $topic ---"# 获取Topic基本信息TOPIC_INFO=$(kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic $topic 2>/dev/null)PARTITION_COUNT=$(echo "$TOPIC_INFO" | grep "PartitionCount" | awk '{print $2}')REPLICATION_FACTOR=$(echo "$TOPIC_INFO" | grep "ReplicationFactor" | awk '{print $4}')echo " 分区数: $PARTITION_COUNT, 副本因子: $REPLICATION_FACTOR"# 分析分区Leader分布echo " 分区Leader分布:"echo "$TOPIC_INFO" | grep -E "Partition: [0-9]+" | awk '{print $6}' | sort | uniq -c | \awk '{printf " Broker %s: %d 个Leader分区\n", $2, $1}'# 检查分区大小if [ -d "$KAFKA_LOG_DIR" ]; thenTOPIC_SIZE=$(du -sh $KAFKA_LOG_DIR/${topic}-* 2>/dev/null | awk '{sum+=$1}END{print sum "M"}')echo " 磁盘使用: $TOPIC_SIZE"fi# 检查消费延迟情况CONSUMER_GROUPS=$(kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list 2>/dev/null | \xargs -I {} kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group {} 2>/dev/null | \grep "$topic" | awk '{print $1}' | sort | uniq)if [ -n "$CONSUMER_GROUPS" ]; thenecho " 相关消费者组延迟:"for group in $CONSUMER_GROUPS; doTOTAL_LAG=$(kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group $group 2>/dev/null | \grep "$topic" | awk '$5 != "-" {sum += $5} END {print sum+0}')if [ "$TOTAL_LAG" -gt 0 ]; thenecho " $group: $TOTAL_LAG 条消息堆积"fidonefiecho ""fi
doneecho "=== Broker和集群性能分析完成 ==="
第四步：性能优化建议生成
#!/bin/bash
# Kafka性能优化建议生成脚本echo "=== Kafka性能优化建议报告 ==="# 1. 系统级优化建议
echo "1. 系统级性能优化建议"
echo "================================"# 检查系统资源使用情况
MEMORY_USAGE=$(free | awk 'NR==2{printf "%.1f", $3*100/$2}')
CPU_USAGE=$(top -bn1 | grep "Cpu(s)" | awk '{print $2}' | cut -d'%' -f1)
DISK_USAGE=$(df -h / | awk 'NR==2{print $5}' | cut -d'%' -f1)echo "当前系统资源使用情况:"
echo " 内存使用率: ${MEMORY_USAGE}%"
echo " CPU使用率: ${CPU_USAGE}%"
echo " 磁盘使用率: ${DISK_USAGE}%"echo -e "\n系统级优化建议:"# 内存优化建议
if (( $(echo "$MEMORY_USAGE > 80" | bc -l) )); thencat << 'EOF'
🔧 内存优化建议:
1. 增加系统内存或优化JVM堆内存配置export KAFKA_HEAP_OPTS="-Xmx6G -Xms6G"2. 启用内存映射文件优化# 在server.properties中添加:log.segment.bytes=1073741824log.index.size.max.bytes=104857603. 调整系统内核参数echo 'vm.swappiness=1' >> /etc/sysctl.confecho 'vm.dirty_ratio=80' >> /etc/sysctl.confecho 'vm.dirty_background_ratio=5' >> /etc/sysctl.conf
EOF
fi# CPU优化建议
if (( $(echo "$CPU_USAGE > 70" | bc -l) )); thencat << 'EOF'🔧 CPU优化建议:
1. 启用G1垃圾收集器减少停顿时间export KAFKA_JVM_PERFORMANCE_OPTS="-XX:+UseG1GC -XX:MaxGCPauseMillis=20"2. 调整消息压缩算法# 使用LZ4压缩算法平衡CPU和网络compression.type=lz43. 优化批处理配置batch.size=32768linger.ms=5
EOF
fi# 磁盘优化建议
if [ "$DISK_USAGE" -gt 80 ]; thencat << 'EOF'🔧 磁盘优化建议:
1. 调整数据保留策略log.retention.hours=168 # 7天保留log.segment.bytes=1073741824 # 1GB段文件2. 启用日志压缩log.cleanup.policy=compact3. 使用SSD磁盘并优化文件系统# 推荐使用XFS文件系统mkfs.xfs -f /dev/sdb1mount -o noatime,nodiratime /dev/sdb1 /kafka-logs
EOF
fi# 2. Kafka配置优化建议
echo -e "\n2. Kafka配置参数优化建议"
echo "================================"cat << 'EOF'
根据性能测试结果，推荐以下配置优化:🚀 Broker配置优化 (server.properties):
# 网络和IO优化
num.network.threads=8
num.io.threads=16
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600# 日志配置优化
log.segment.bytes=1073741824
log.index.size.max.bytes=10485760
log.index.interval.bytes=4096
log.flush.interval.messages=10000
log.flush.interval.ms=1000# 副本配置优化
default.replication.factor=3
min.insync.replicas=2
unclean.leader.election.enable=false
replica.fetch.max.bytes=1048576# JVM参数优化
export KAFKA_HEAP_OPTS="-Xmx6G -Xms6G"
export KAFKA_JVM_PERFORMANCE_OPTS="-XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:+UnlockExperimentalVMOptions"🚀 生产者配置优化:
# 高吞吐量配置
bootstrap.servers=broker1:9092,broker2:9092,broker3:9092
acks=1
compression.type=lz4
batch.size=32768
linger.ms=5
buffer.memory=67108864
max.request.size=1048576# 高可靠性配置 (如果需要)
acks=all
retries=2147483647
max.in.flight.requests.per.connection=5
enable.idempotence=true🚀 消费者配置优化:
# 高吞吐量配置
fetch.min.bytes=50000
fetch.max.wait.ms=500
max.poll.records=1000
session.timeout.ms=30000
heartbeat.interval.ms=3000# 内存优化配置
receive.buffer.bytes=65536
send.buffer.bytes=131072
max.partition.fetch.bytes=1048576
EOF# 3. 监控和告警建议
echo -e "\n3. 监控和告警配置建议"
echo "================================"cat << 'EOF'
建立完善的监控体系:📊 关键指标监控:
1. 吞吐量指标- 消息生产速率 (messages/sec)- 消息消费速率 (messages/sec)- 网络带宽使用率 (MB/sec)2. 延迟指标- 端到端消息延迟- 消费者组Lag- 生产者发送延迟3. 系统资源指标- CPU使用率- 内存使用率- 磁盘IO利用率- 网络连接数4. 集群健康指标- Broker在线状态- 分区Leader分布- 副本同步状态🔔 告警配置:
# 消息堆积告警
Consumer Lag > 100000: 严重告警
Consumer Lag > 10000: 警告告警# 系统资源告警
CPU使用率 > 80%: 警告告警
内存使用率 > 85%: 警告告警
磁盘使用率 > 80%: 警告告警# 集群状态告警
Broker离线: 严重告警
分区没有Leader: 严重告警
副本同步延迟 > 1000ms: 警告告警
EOF# 4. 容量规划建议
echo -e "\n4. 容量规划和扩容建议"
echo "================================"# 分析当前集群容量
BROKER_COUNT=$(kafka-broker-api-versions.sh --bootstrap-server localhost:9092 2>/dev/null | wc -l)
TOTAL_PARTITIONS=$(kafka-topics.sh --bootstrap-server localhost:9092 --describe 2>/dev/null | grep "PartitionCount" | awk '{sum += $2} END {print sum+0}')echo "当前集群容量分析:"
echo " Broker数量: $BROKER_COUNT"
echo " 总分区数: $TOTAL_PARTITIONS"
echo " 平均每Broker分区数: $((TOTAL_PARTITIONS / BROKER_COUNT))"cat << 'EOF'📈 容量规划建议:
1. Broker数量规划- 建议每个Broker管理的分区数不超过2000个- 考虑50%的增长空间进行容量规划- 确保至少3个Broker以支持副本机制2. 分区数量规划- 分区数 = 目标吞吐量 / 单分区吞吐量- 建议分区数为Broker数量的2-3倍- 考虑消费者的并发能力确定分区数3. 磁盘容量规划- 磁盘容量 = 日消息量 × 保留天数 × 副本因子 × 1.2(冗余)- 建议使用SSD磁盘提高IO性能- 预留30%的磁盘空间用于临时文件和索引4. 网络带宽规划- 考虑副本同步流量 = 生产流量 × (副本因子 - 1)- 预留50%的网络带宽用于故障恢复和重平衡🔄 自动化扩容策略:
1. 基于消息堆积的自动扩容
2. 基于系统资源使用率的预警
3. 定期的容量评估和调整
4. 流量高峰期的弹性扩容
EOFecho -e "\n=== 性能优化建议报告完成 ==="
🔄 四、分区和副本问题处理
📚 Kafka分区和副本机制深度解析
分区（Partition）和副本（Replica）是Kafka实现高可用性和水平扩展的核心机制。深入理解这两个概念对于排查和解决Kafka问题至关重要。
分区机制的设计原理：
- 数据分片：每个Topic被分为多个Partition，每个Partition是一个有序的消息序列
- 并行处理：多个Partition允许多个消费者并行处理，提高整体吞吐量
- 负载均衡：Partition分布在不同的Broker上，实现负载的均匀分配
- 顺序保证：单个Partition内消息严格有序，但Topic级别不保证全局顺序
副本机制的核心作用：
- 数据冗余：每个Partition可以有多个Replica，防止数据丢失
- 故障恢复：当Leader Replica故障时，可以从Follower Replica中选举新的Leader
- 读写分离：Leader负责处理读写请求，Follower负责数据同步
- 一致性保证：通过ISR（In-Sync Replicas）机制确保数据一致性
常见的分区和副本问题：
- 分区分布不均：某些Broker承载过多分区，造成热点问题
- Leader分布不均：部分Broker成为性能瓶颈
- 副本同步延迟：Follower Replica跟不上Leader的更新速度
- 脑裂问题：网络分区导致出现多个Controller
- 分区离线：某些分区无法提供服务
🔍 分区问题排查详细流程
第一步：分区分布和负载分析
#!/bin/bash
# Kafka分区分布和负载深度分析脚本echo "=== Kafka分区分布和负载分析报告 ==="# 1. 全局分区分布概况
echo "1. 集群分区分布概况"
echo "================================"# 获取所有Broker列表
BROKERS=$(kafka-broker-api-versions.sh --bootstrap-server localhost:9092 2>/dev/null | awk '{print $1}' | sort -n)
BROKER_COUNT=$(echo "$BROKERS" | wc -l)echo "集群Broker列表: $(echo $BROKERS | tr '\n' ' ')"
echo "Broker总数: $BROKER_COUNT"# 统计总分区数和Topic数
TOTAL_PARTITIONS=0
TOTAL_TOPICS=0# 创建临时文件存储分析数据
TEMP_ANALYSIS="/tmp/kafka_partition_analysis_$(date +%s).txt"echo "Topic,Partitions,ReplicationFactor,Leader_Distribution" > $TEMP_ANALYSIS# 分析每个Topic的分区分布
TOPICS=$(kafka-topics.sh --bootstrap-server localhost:9092 --list 2>/dev/null)for topic in $TOPICS; doif [ -n "$topic" ]; then((TOTAL_TOPICS++))echo "--- 分析Topic: $topic ---"# 获取Topic详细信息TOPIC_INFO=$(kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic $topic 2>/dev/null)PARTITION_COUNT=$(echo "$TOPIC_INFO" | grep "PartitionCount" | awk '{print $2}')REPLICATION_FACTOR=$(echo "$TOPIC_INFO" | grep "ReplicationFactor" | awk '{print $4}')echo " 分区数: $PARTITION_COUNT"echo " 副本因子: $REPLICATION_FACTOR"TOTAL_PARTITIONS=$((TOTAL_PARTITIONS + PARTITION_COUNT))# 分析Leader分布echo " Leader分区分布:"LEADER_DISTRIBUTION=$(echo "$TOPIC_INFO" | grep -E "Partition: [0-9]+" | awk '{print $6}' | sort | uniq -c)echo "$LEADER_DISTRIBUTION" | awk '{printf " Broker %s: %d 个Leader分区\n", $2, $1}'# 检查分区分布是否均匀MAX_LEADERS=$(echo "$LEADER_DISTRIBUTION" | awk '{print $1}' | sort -nr | head -1)MIN_LEADERS=$(echo "$LEADER_DISTRIBUTION" | awk '{print $1}' | sort -n | head -1)if [ $((MAX_LEADERS - MIN_LEADERS)) -gt 2 ]; thenecho " ⚠️ 警告: Leader分区分布不均匀 (最大: $MAX_LEADERS, 最小: $MIN_LEADERS)"elseecho " ✅ Leader分区分布相对均匀"fi# 检查副本分布echo " 副本分布检查:"echo "$TOPIC_INFO" | grep -E "Partition: [0-9]+" | while read line; doPARTITION_ID=$(echo "$line" | awk '{print $2}')REPLICAS=$(echo "$line" | awk '{print $8}')ISR=$(echo "$line" | awk '{print $10}')REPLICA_COUNT=$(echo "$REPLICAS" | tr ',' '\n' | wc -l)ISR_COUNT=$(echo "$ISR" | tr ',' '\n' | wc -l)if [ "$REPLICA_COUNT" -ne "$REPLICATION_FACTOR" ]; thenecho " ⚠️ 分区 $PARTITION_ID: 副本数异常 ($REPLICA_COUNT/$REPLICATION_FACTOR)"fiif [ "$ISR_COUNT" -lt "$REPLICATION_FACTOR" ]; thenecho " ⚠️ 分区 $PARTITION_ID: ISR副本不足 ($ISR_COUNT/$REPLICATION_FACTOR)"fidone# 保存分析数据LEADER_DIST_STR=$(echo "$LEADER_DISTRIBUTION" | awk '{printf "%s:%s ", $2, $1}')echo "$topic,$PARTITION_COUNT,$REPLICATION_FACTOR,$LEADER_DIST_STR" >> $TEMP_ANALYSISecho ""fi
doneecho "=== 集群分区统计总结 ==="
echo "总Topic数: $TOTAL_TOPICS"
echo "总分区数: $TOTAL_PARTITIONS"
echo "平均每Topic分区数: $((TOTAL_PARTITIONS / TOTAL_TOPICS))"
echo "平均每Broker分区数: $((TOTAL_PARTITIONS / BROKER_COUNT))"# 2. Broker级别的分区负载分析
echo -e "\n2. Broker级别的分区负载分析"
echo "================================"# 统计每个Broker的分区负载
echo "每个Broker的详细分区统计:"
for broker in $BROKERS; doecho "--- Broker $broker ---"# 统计该Broker上的Leader分区数LEADER_COUNT=$(kafka-topics.sh --bootstrap-server localhost:9092 --describe 2>/dev/null | \grep -E "Partition: [0-9]+" | grep "Leader: $broker" | wc -l)# 统计该Broker上的总副本数（包括Leader和Follower）REPLICA_COUNT=$(kafka-topics.sh --bootstrap-server localhost:9092 --describe 2>/dev/null | \grep -E "Partition: [0-9]+" | grep -E "Replicas:.*$broker" | wc -l)echo " Leader分区数: $LEADER_COUNT"echo " 总副本数: $REPLICA_COUNT"echo " Follower副本数: $((REPLICA_COUNT - LEADER_COUNT))"# 计算负载比例LEADER_PERCENTAGE=$(echo "scale=2; $LEADER_COUNT * 100 / $TOTAL_PARTITIONS" | bc)REPLICA_PERCENTAGE=$(echo "scale=2; $REPLICA_COUNT * 100 / ($TOTAL_PARTITIONS * 3)" | bc) # 假设平均副本因子为3echo " Leader负载占比: ${LEADER_PERCENTAGE}%"echo " 副本负载占比: ${REPLICA_PERCENTAGE}%"# 检查是否存在负载不均衡EXPECTED_LEADER_PERCENTAGE=$(echo "scale=2; 100 / $BROKER_COUNT" | bc)DEVIATION=$(echo "scale=2; $LEADER_PERCENTAGE - $EXPECTED_LEADER_PERCENTAGE" | bc | tr -d '-')if (( $(echo "$DEVIATION > 10" | bc -l) )); thenecho " ⚠️ 警告: Leader负载偏差较大 (偏差: ${DEVIATION}%)"elseecho " ✅ Leader负载相对均衡"fiecho ""
done# 3. 分区热点分析
echo "3. 分区热点分析"
echo "================================"# 分析哪些Topic可能成为热点
echo "潜在热点Topic分析:"
awk -F, 'NR>1 {split($4, leaders, " ")max_leaders = 0for (i in leaders) {split(leaders[i], pair, ":")if (pair[2] > max_leaders) max_leaders = pair[2]}hotspot_ratio = max_leaders / $2 * 100if (hotspot_ratio > 50) {printf " ⚠️ %s: 单Broker承载%.1f%%的Leader分区\n", $1, hotspot_ratio}
}' $TEMP_ANALYSIS# 清理临时文件
rm -f $TEMP_ANALYSISecho -e "\n=== 分区分布分析完成 ==="
第二步：副本同步状态检查
#!/bin/bash
# Kafka副本同步状态深度检查脚本echo "=== Kafka副本同步状态检查报告 ==="# 1. ISR (In-Sync Replicas) 状态检查
echo "1. ISR状态全面检查"
echo "================================"PROBLEMATIC_PARTITIONS=0
TOTAL_PARTITIONS_CHECKED=0# 检查所有Topic的ISR状态
TOPICS=$(kafka-topics.sh --bootstrap-server localhost:9092 --list 2>/dev/null)for topic in $TOPICS; doif [ -n "$topic" ]; thenecho "--- 检查Topic: $topic ---"TOPIC_INFO=$(kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic $topic 2>/dev/null)REPLICATION_FACTOR=$(echo "$TOPIC_INFO" | grep "ReplicationFactor" | awk '{print $4}')echo " 副本因子: $REPLICATION_FACTOR"echo " 分区ISR状态检查:"# 分析每个分区的ISR状态echo "$TOPIC_INFO" | grep -E "Partition: [0-9]+" | while read line; doPARTITION_ID=$(echo "$line" | awk '{print $2}')LEADER=$(echo "$line" | awk '{print $6}')REPLICAS=$(echo "$line" | awk '{print $8}')ISR=$(echo "$line" | awk '{print $10}')# 统计副本数和ISR数REPLICA_COUNT=$(echo "$REPLICAS" | tr ',' ' ' | wc -w)ISR_COUNT=$(echo "$ISR" | tr ',' ' ' | wc -w)((TOTAL_PARTITIONS_CHECKED++))printf " 分区 %-3s: Leader=%s, 副本=[%s], ISR=[%s]\n" $PARTITION_ID "$LEADER" "$REPLICAS" "$ISR"# 检查ISR状态异常情况if [ "$ISR_COUNT" -lt "$REPLICATION_FACTOR" ]; thenecho " ⚠️ ISR副本不足: 期望 $REPLICATION_FACTOR 个，实际 $ISR_COUNT 个"((PROBLEMATIC_PARTITIONS++))# 分析哪些副本掉出了ISROUT_OF_SYNC_REPLICAS=$(comm -23 <(echo "$REPLICAS" | tr ',' '\n' | sort) <(echo "$ISR" | tr ',' '\n' | sort))if [ -n "$OUT_OF_SYNC_REPLICAS" ]; thenecho " 掉出ISR的副本: $(echo $OUT_OF_SYNC_REPLICAS | tr '\n' ',')"fielif [ "$ISR_COUNT" -eq "$REPLICATION_FACTOR" ]; thenecho " ✅ ISR状态正常"fi# 检查Leader是否在ISR中if ! echo "$ISR" | grep -q "$LEADER"; thenecho " ❌ 严重错误: Leader不在ISR中!"((PROBLEMATIC_PARTITIONS++))fidoneecho ""fi
doneecho "=== ISR状态检查总结 ==="
echo "检查的分区总数: $TOTAL_PARTITIONS_CHECKED"
echo "存在问题的分区数: $PROBLEMATIC_PARTITIONS"
ISR_HEALTH_PERCENTAGE=$(echo "scale=2; ($TOTAL_PARTITIONS_CHECKED - $PROBLEMATIC_PARTITIONS) * 100 / $TOTAL_PARTITIONS_CHECKED" | bc)
echo "ISR健康度: ${ISR_HEALTH_PERCENTAGE}%"# 2. 副本延迟分析
echo -e "\n2. 副本延迟深度分析"
echo "================================"# 检查副本延迟的JMX指标（如果启用了JMX）
echo "副本延迟监控指标获取:"if command -v jmxtrans >/dev/null 2>&1; thenecho " 通过JMX获取副本延迟指标..."# 获取副本延迟相关指标REPLICA_FETCH_RATE=$(jmxtrans_query "kafka.server:type=FetcherStats,name=BytesPerSec" 2>/dev/null || echo "N/A")REPLICA_LAG=$(jmxtrans_query "kafka.server:type=FetcherLagMetrics,name=ConsumerLag" 2>/dev/null || echo "N/A")echo " 副本fetch速率: $REPLICA_FETCH_RATE"echo " 副本延迟: $REPLICA_LAG"
elseecho " JMX未启用或jmxtrans工具不可用"echo " 建议启用JMX监控获取详细的副本同步指标"
fi# 通过日志分析副本同步情况
echo -e "\n通过日志分析副本同步状况:"
KAFKA_LOG_FILE="/opt/kafka/logs/server.log"if [ -f "$KAFKA_LOG_FILE" ]; thenecho " 最近的副本同步相关日志:"grep -i "replica" "$KAFKA_LOG_FILE" | tail -10 | while read line; doecho " $line"done# 检查是否有副本同步异常SYNC_ERRORS=$(grep -c -i "replica.*error\|replica.*fail" "$KAFKA_LOG_FILE")if [ "$SYNC_ERRORS" -gt 0 ]; thenecho " ⚠️ 发现 $SYNC_ERRORS 条副本同步错误日志"elseecho " ✅ 未发现副本同步错误"fi
elseecho " Kafka日志文件不存在或路径不正确: $KAFKA_LOG_FILE"
fi# 3. 副本分布均衡性检查
echo -e "\n3. 副本分布均衡性检查"
echo "================================"# 分析副本在各Broker上的分布
BROKERS=$(kafka-broker-api-versions.sh --bootstrap-server localhost:9092 2>/dev/null | awk '{print $1}' | sort -n)echo "各Broker副本分布统计:"
for broker in $BROKERS; doecho "--- Broker $broker ---"# 统计该Broker作为Leader的分区数LEADER_COUNT=$(kafka-topics.sh --bootstrap-server localhost:9092 --describe 2>/dev/null | \grep -E "Partition: [0-9]+" | grep "Leader: $broker" | wc -l)# 统计该Broker的总副本数TOTAL_REPLICAS=$(kafka-topics.sh --bootstrap-server localhost:9092 --describe 2>/dev/null | \grep -E "Partition: [0-9]+" | grep -E "Replicas:.*$broker" | wc -l)# 统计该Broker在ISR中的副本数ISR_REPLICAS=$(kafka-topics.sh --bootstrap-server localhost:9092 --describe 2>/dev/null | \grep -E "Partition: [0-9]+" | grep -E "Isr:.*$broker" | wc -l)echo " Leader副本数: $LEADER_COUNT"echo " 总副本数: $TOTAL_REPLICAS"echo " ISR副本数: $ISR_REPLICAS"echo " 掉出ISR的副本数: $((TOTAL_REPLICAS - ISR_REPLICAS))"# 计算副本健康度if [ "$TOTAL_REPLICAS" -gt 0 ]; thenREPLICA_HEALTH=$(echo "scale=2; $ISR_REPLICAS * 100 / $TOTAL_REPLICAS" | bc)echo " 副本健康度: ${REPLICA_HEALTH}%"if (( $(echo "$REPLICA_HEALTH < 95" | bc -l) )); thenecho " ⚠️ 警告: 该Broker副本健康度较低"elseecho " ✅ 副本健康度良好"fifiecho ""
done# 4. 生成副本修复建议
echo "4. 副本问题修复建议"
echo "================================"if [ "$PROBLEMATIC_PARTITIONS" -gt 0 ]; thencat << 'EOF'
发现副本同步问题，建议采取以下修复措施:🔧 立即修复措施:
1. 重启有问题的Broker节点# 优雅重启Kafka服务systemctl stop kafkasleep 10systemctl start kafka2. 手动触发分区重新分配# 生成重新分配计划kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \--topics-to-move-json-file topics.json \--broker-list "1,2,3" --generate# 执行重新分配kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \--reassignment-json-file reassignment.json --execute3. 调整副本同步相关配置# 在server.properties中调整以下参数replica.fetch.max.bytes=1048576replica.fetch.wait.max.ms=500replica.high.watermark.checkpoint.interval.ms=5000🔧 预防性措施:
1. 监控副本延迟指标- 设置ISR副本数量告警- 监控副本同步延迟- 定期检查副本分布均衡性2. 优化网络和磁盘性能- 确保Broker间网络稳定- 使用高性能SSD磁盘- 调整操作系统网络参数3. 合理设置副本因子- 根据可用性需求设置合适的副本因子- 避免副本因子过大影响性能- 确保至少有3个Broker支持副本机制
EOF
elseecho "✅ 未发现副本同步问题，集群副本状态健康"
fiecho -e "\n=== 副本同步状态检查完成 ==="
第三步：分区重平衡和修复
#!/bin/bash
# Kafka分区重平衡和修复工具脚本echo "=== Kafka分区重平衡和修复工具 ==="# 1. 分区分布均衡性评估
echo "1. 分区分布均衡性评估"
echo "================================"# 获取集群基本信息
BROKERS=$(kafka-broker-api-versions.sh --bootstrap-server localhost:9092 2>/dev/null | awk '{print $1}' | sort -n)
BROKER_COUNT=$(echo "$BROKERS" | wc -l)
BROKER_LIST=$(echo "$BROKERS" | tr '\n' ',' | sed 's/,$//')echo "集群Broker列表: $BROKER_LIST"
echo "Broker总数: $BROKER_COUNT"# 分析当前分区分布
declare -A BROKER_LEADER_COUNT
declare -A BROKER_REPLICA_COUNT# 初始化计数器
for broker in $BROKERS; doBROKER_LEADER_COUNT[$broker]=0BROKER_REPLICA_COUNT[$broker]=0
done# 统计各Broker的分区分布
echo "正在分析分区分布..."
TOPICS=$(kafka-topics.sh --bootstrap-server localhost:9092 --list 2>/dev/null)TOTAL_LEADERS=0
TOTAL_REPLICAS=0for topic in $TOPICS; doif [ -n "$topic" ]; thenTOPIC_INFO=$(kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic $topic 2>/dev/null)echo "$TOPIC_INFO" | grep -E "Partition: [0-9]+" | while read line; doLEADER=$(echo "$line" | awk '{print $6}')REPLICAS=$(echo "$line" | awk '{print $8}')# 统计Leader分区BROKER_LEADER_COUNT[$LEADER]=$((${BROKER_LEADER_COUNT[$LEADER]} + 1))TOTAL_LEADERS=$((TOTAL_LEADERS + 1))# 统计副本分区IFS=',' read -ra REPLICA_ARRAY <<< "$REPLICAS"for replica in "${REPLICA_ARRAY[@]}"; doBROKER_REPLICA_COUNT[$replica]=$((${BROKER_REPLICA_COUNT[$replica]} + 1))TOTAL_REPLICAS=$((TOTAL_REPLICAS + 1))donedonefi
done# 计算分布均衡度
echo -e "\n当前分区分布统计:"
echo "Broker ID Leader分区数 总副本数 Leader占比 副本占比"
echo "================================================================"MAX_LEADER_DIFF=0
MAX_REPLICA_DIFF=0
EXPECTED_LEADERS_PER_BROKER=$((TOTAL_LEADERS / BROKER_COUNT))
EXPECTED_REPLICAS_PER_BROKER=$((TOTAL_REPLICAS / BROKER_COUNT))for broker in $BROKERS; doLEADER_COUNT=$(kafka-topics.sh --bootstrap-server localhost:9092 --describe 2>/dev/null | \grep -E "Partition: [0-9]+" | grep "Leader: $broker" | wc -l)REPLICA_COUNT=$(kafka-topics.sh --bootstrap-server localhost:9092 --describe 2>/dev/null | \grep -E "Partition: [0-9]+" | grep -E "Replicas:.*$broker" | wc -l)LEADER_PERCENTAGE=$(echo "scale=1; $LEADER_COUNT * 100 / $TOTAL_LEADERS" | bc)REPLICA_PERCENTAGE=$(echo "scale=1; $REPLICA_COUNT * 100 / $TOTAL_REPLICAS" | bc)printf "%-10s %-12s %-12s %-12s %-12s\n" \"$broker" "$LEADER_COUNT" "$REPLICA_COUNT" "${LEADER_PERCENTAGE}%" "${REPLICA_PERCENTAGE}%"# 计算最大偏差LEADER_DIFF=$((LEADER_COUNT > EXPECTED_LEADERS_PER_BROKER ? LEADER_COUNT - EXPECTED_LEADERS_PER_BROKER : EXPECTED_LEADERS_PER_BROKER - LEADER_COUNT))REPLICA_DIFF=$((REPLICA_COUNT > EXPECTED_REPLICAS_PER_BROKER ? REPLICA_COUNT - EXPECTED_REPLICAS_PER_BROKER : EXPECTED_REPLICAS_PER_BROKER - REPLICA_COUNT))if [ $LEADER_DIFF -gt $MAX_LEADER_DIFF ]; thenMAX_LEADER_DIFF=$LEADER_DIFFfiif [ $REPLICA_DIFF -gt $MAX_REPLICA_DIFF ]; thenMAX_REPLICA_DIFF=$REPLICA_DIFFfi
doneecho ""
echo "均衡性评估:"
echo "期望每Broker Leader分区数: $EXPECTED_LEADERS_PER_BROKER"
echo "期望每Broker副本数: $EXPECTED_REPLICAS_PER_BROKER"
echo "最大Leader分区偏差: $MAX_LEADER_DIFF"
echo "最大副本偏差: $MAX_REPLICA_DIFF"# 评估是否需要重平衡
REBALANCE_NEEDED=false
REBALANCE_REASON=""if [ $MAX_LEADER_DIFF -gt $((EXPECTED_LEADERS_PER_BROKER / 4)) ]; thenREBALANCE_NEEDED=trueREBALANCE_REASON="Leader分区分布不均衡"
fiif [ $MAX_REPLICA_DIFF -gt $((EXPECTED_REPLICAS_PER_BROKER / 4)) ]; thenREBALANCE_NEEDED=trueREBALANCE_REASON="$REBALANCE_REASON 副本分布不均衡"
fiif [ "$REBALANCE_NEEDED" = "true" ]; thenecho "⚠️ 建议执行重平衡操作: $REBALANCE_REASON"
elseecho "✅ 当前分区分布相对均衡，无需重平衡"
fi# 2. 自动化重平衡方案生成
echo -e "\n2. 自动化重平衡方案生成"
echo "================================"if [ "$REBALANCE_NEEDED" = "true" ]; then# 创建重平衡配置目录REBALANCE_DIR="/tmp/kafka_rebalance_$(date +%s)"mkdir -p "$REBALANCE_DIR"echo "创建重平衡工作目录: $REBALANCE_DIR"# 生成Topic列表文件TOPICS_FILE="$REBALANCE_DIR/topics_to_move.json"echo '{"topics":[' > "$TOPICS_FILE"TOPIC_LIST=""for topic in $TOPICS; doif [ -n "$topic" ]; thenif [ -n "$TOPIC_LIST" ]; thenTOPIC_LIST="$TOPIC_LIST,"fiTOPIC_LIST="$TOPIC_LIST{\"topic\":\"$topic\"}"fidoneecho "$TOPIC_LIST" >> "$TOPICS_FILE"echo '],"version":1}' >> "$TOPICS_FILE"echo "生成的Topic列表文件:"cat "$TOPICS_FILE"# 生成重平衡计划echo -e "\n正在生成重平衡计划..."REASSIGNMENT_FILE="$REBALANCE_DIR/reassignment_plan.json"kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \--topics-to-move-json-file "$TOPICS_FILE" \--broker-list "$BROKER_LIST" \--generate > "$REBALANCE_DIR/reassignment_output.txt" 2>&1if [ $? -eq 0 ]; then# 提取重分配计划sed -n '/Proposed partition reassignment configuration/,/^$/p' "$REBALANCE_DIR/reassignment_output.txt" | \grep -v "Proposed partition reassignment configuration" | \grep -v "^$" > "$REASSIGNMENT_FILE"echo "✅ 重平衡计划生成成功"echo "计划文件位置: $REASSIGNMENT_FILE"# 显示重平衡计划摘要echo -e "\n重平衡计划摘要:"PARTITIONS_TO_MOVE=$(grep -o '"partition":' "$REASSIGNMENT_FILE" | wc -l)echo "需要迁移的分区数: $PARTITIONS_TO_MOVE"# 生成执行脚本EXECUTE_SCRIPT="$REBALANCE_DIR/execute_rebalance.sh"cat > "$EXECUTE_SCRIPT" << EOF
#!/bin/bash
# Kafka分区重平衡执行脚本
# 生成时间: $(date)echo "开始执行Kafka分区重平衡..."
echo "重平衡计划文件: $REASSIGNMENT_FILE"# 执行重平衡
kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \\--reassignment-json-file "$REASSIGNMENT_FILE" \\--executeecho "重平衡操作已提交，正在执行中..."# 监控重平衡进度
echo "监控重平衡进度 (每30秒检查一次):"
while true; doSTATUS=\$(kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \\--reassignment-json-file "$REASSIGNMENT_FILE" \\--verify 2>/dev/null)if echo "\$STATUS" | grep -q "Reassignment of partition.*is still in progress"; thenecho "\$(date): 重平衡进行中..."PROGRESS=\$(echo "\$STATUS" | grep -c "completed successfully")TOTAL=\$(echo "\$STATUS" | wc -l)echo "进度: \$PROGRESS/\$TOTAL 个分区完成迁移"sleep 30elseecho "\$(date): 重平衡完成!"echo "\$STATUS"breakfi
doneecho "重平衡操作完成，建议重新检查集群分区分布"
EOFchmod +x "$EXECUTE_SCRIPT"echo "执行脚本已生成: $EXECUTE_SCRIPT"# 生成回滚脚本ROLLBACK_SCRIPT="$REBALANCE_DIR/rollback_rebalance.sh"cat > "$ROLLBACK_SCRIPT" << EOF
#!/bin/bash
# Kafka分区重平衡回滚脚本
# 生成时间: $(date)echo "开始回滚Kafka分区重平衡..."# 获取当前分区分配作为回滚计划
CURRENT_ASSIGNMENT="\$REBALANCE_DIR/current_assignment.json"# 提取当前分配信息
sed -n '/Current partition replica assignment/,/Proposed partition reassignment configuration/p' \\"$REBALANCE_DIR/reassignment_output.txt" | \\grep -v "Current partition replica assignment" | \\grep -v "Proposed partition reassignment configuration" | \\grep -v "^$" > "\$CURRENT_ASSIGNMENT"# 执行回滚
kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \\--reassignment-json-file "\$CURRENT_ASSIGNMENT" \\--executeecho "回滚操作已提交"
EOFchmod +x "$ROLLBACK_SCRIPT"echo "回滚脚本已生成: $ROLLBACK_SCRIPT"# 提供执行建议cat << EOF🔧 重平衡执行建议:
1. 在业务低峰期执行重平衡操作
2. 执行前备份重要数据和配置
3. 监控集群性能和磁盘空间
4. 必要时可以中止重平衡操作执行命令:$EXECUTE_SCRIPT如需回滚:$ROLLBACK_SCRIPT手动执行重平衡:kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \\--reassignment-json-file $REASSIGNMENT_FILE --execute检查重平衡进度:kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \\--reassignment-json-file $REASSIGNMENT_FILE --verify
EOFelseecho "❌ 重平衡计划生成失败"cat "$REBALANCE_DIR/reassignment_output.txt"fi
elseecho "当前集群分区分布均衡，无需执行重平衡操作"
fi# 3. 特定问题修复工具
echo -e "\n3. 特定分区问题修复工具"
echo "================================"# 检查离线分区
echo "检查离线分区:"
OFFLINE_PARTITIONS=$(kafka-topics.sh --bootstrap-server localhost:9092 --describe --unavailable-partitions 2>/dev/null)if [ -n "$OFFLINE_PARTITIONS" ]; thenecho "发现离线分区:"echo "$OFFLINE_PARTITIONS"# 生成离线分区修复脚本OFFLINE_REPAIR_SCRIPT="/tmp/repair_offline_partitions.sh"cat > "$OFFLINE_REPAIR_SCRIPT" << 'EOF'
#!/bin/bash
# 离线分区修复脚本echo "开始修复离线分区..."# 获取离线分区列表
OFFLINE_PARTITIONS=$(kafka-topics.sh --bootstrap-server localhost:9092 --describe --unavailable-partitions)if [ -n "$OFFLINE_PARTITIONS" ]; thenecho "发现以下离线分区:"echo "$OFFLINE_PARTITIONS"# 尝试重启相关的Brokerecho "建议检查并重启以下操作:"echo "1. 检查相关Broker的日志文件"echo "2. 确认磁盘空间和系统资源"echo "3. 重启有问题的Broker节点"echo "4. 如果数据损坏，考虑从副本恢复"# 对于严重情况，可以考虑重新创建分区echo ""echo "紧急情况下的数据恢复选项:"echo "⚠️ 警告: 以下操作会导致数据丢失，请谨慎使用"echo ""echo "# 如果确认数据不可恢复，可以删除并重新创建Topic"echo "# kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic problematic_topic"echo "# kafka-topics.sh --bootstrap-server localhost:9092 --create --topic problematic_topic --partitions N --replication-factor 3"
elseecho "✅ 未发现离线分区"
fi
EOFchmod +x "$OFFLINE_REPAIR_SCRIPT"echo "离线分区修复脚本已生成: $OFFLINE_REPAIR_SCRIPT"
elseecho "✅ 未发现离线分区"
fi# 检查不同步的副本
echo -e "\n检查不同步的副本:"
UNDER_REPLICATED=$(kafka-topics.sh --bootstrap-server localhost:9092 --describe --under-replicated-partitions 2>/dev/null)if [ -n "$UNDER_REPLICATED" ]; thenecho "发现副本不足的分区:"echo "$UNDER_REPLICATED"echo -e "\n副本修复建议:"echo "1. 检查掉出ISR的Broker状态"echo "2. 确认网络连接和磁盘IO正常"echo "3. 调整副本同步相关配置参数"echo "4. 考虑增加副本因子提高可用性"
elseecho "✅ 所有分区副本状态正常"
fiecho -e "\n=== 分区重平衡和修复工具完成 ==="
👥 五、消费者问题排查指南
📚 Kafka消费者架构和工作原理
Kafka消费者是整个数据流处理链路的关键环节，其性能和稳定性直接影响业务数据的实时性和准确性。深入理解Kafka消费者的工作机制是排查问题的基础。
消费者组（Consumer Group）机制：
- 负载均衡：同一消费者组内的多个消费者实例会自动分配Topic的不同分区
- 故障恢复：当某个消费者实例故障时，其负责的分区会自动分配给组内其他消费者
- 水平扩展：通过增加消费者实例数量来提高消费能力
- Offset管理：消费者组统一管理消费进度，确保消息不会重复消费或丢失
消费者的关键组件：
- Coordinator：负责管理消费者组的成员关系和分区分配
- Heartbeat机制：消费者定期向Coordinator发送心跳，证明自己仍然活跃
- Rebalance过程：当消费者组成员发生变化时，重新分配分区的过程
- Offset提交：将消费进度持久化到Kafka集群中
常见的消费者问题类型：
- 消费延迟：消费者处理速度跟不上生产者发送速度，导致消息堆积
- 重复消费：由于Offset提交策略不当或异常情况导致消息被重复处理
- 消息丢失：不合理的Offset提交时机可能导致消息丢失
- Rebalance频繁：消费者频繁加入和离开消费者组，影响消费效率
- 消费者假死：消费者进程存在但无法正常处理消息
- 分区分配不均：某些消费者承担过多分区，造成负载不均衡
🔍 消费者问题深度排查流程
第一步：消费者组状态和健康度检查
#!/bin/bash
# Kafka消费者组状态和健康度深度检查脚本echo "=== Kafka消费者组健康度检查报告 ==="# 1. 消费者组基本信息收集
echo "1. 消费者组基本信息收集"
echo "================================"# 获取所有消费者组列表
CONSUMER_GROUPS=$(kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list 2>/dev/null)
GROUP_COUNT=$(echo "$CONSUMER_GROUPS" | wc -l)echo "集群消费者组总数: $GROUP_COUNT"
echo "消费者组列表:"
echo "$CONSUMER_GROUPS" | head -20
if [ $GROUP_COUNT -gt 20 ]; thenecho "... (显示前20个，总共$GROUP_COUNT个)"
fi# 2. 详细分析每个消费者组
echo -e "\n2. 消费者组详细健康度分析"
echo "================================"HEALTHY_GROUPS=0
WARNING_GROUPS=0
CRITICAL_GROUPS=0# 创建详细报告文件
REPORT_FILE="/tmp/consumer_groups_health_report_$(date +%s).txt"
echo "Consumer Group Health Report - $(date)" > "$REPORT_FILE"
echo "=================================================" >> "$REPORT_FILE"for group in $CONSUMER_GROUPS; doif [ -n "$group" ]; thenecho "--- 分析消费者组: $group ---"echo "--- Consumer Group: $group ---" >> "$REPORT_FILE"# 获取消费者组详细信息GROUP_DETAIL=$(kafka-consumer-groups.sh --bootstrap-server localhost:9092 \--describe --group $group 2>/dev/null)if [ -n "$GROUP_DETAIL" ]; then# 解析基本信息TOTAL_PARTITIONS=$(echo "$GROUP_DETAIL" | awk 'NR>1' | wc -l)ACTIVE_CONSUMERS=$(echo "$GROUP_DETAIL" | awk 'NR>1 && $7 != "-"' | wc -l)TOTAL_LAG=$(echo "$GROUP_DETAIL" | awk 'NR>1 && $5 != "-" {sum += $5} END {print sum+0}')# 获取涉及的Topic数量TOPICS=$(echo "$GROUP_DETAIL" | awk 'NR>1 {print $1}' | sort | uniq)TOPIC_COUNT=$(echo "$TOPICS" | wc -l)echo " 基本信息:"echo " 涉及Topic数: $TOPIC_COUNT"echo " 分区总数: $TOTAL_PARTITIONS"echo " 活跃消费者: $ACTIVE_CONSUMERS"echo " 消息堆积: $TOTAL_LAG"# 写入报告文件{echo " Topics: $(echo $TOPICS | tr '\n' ' ')"echo " Total Partitions: $TOTAL_PARTITIONS"echo " Active Consumers: $ACTIVE_CONSUMERS"echo " Total Lag: $TOTAL_LAG"} >> "$REPORT_FILE"# 健康度评估HEALTH_STATUS="健康"HEALTH_LEVEL="HEALTHY"# 检查活跃消费者比例if [ $ACTIVE_CONSUMERS -eq 0 ]; thenHEALTH_STATUS="严重异常 - 无活跃消费者"HEALTH_LEVEL="CRITICAL"((CRITICAL_GROUPS++))elif [ $ACTIVE_CONSUMERS -lt $((TOTAL_PARTITIONS / 2)) ]; thenHEALTH_STATUS="异常 - 活跃消费者不足"HEALTH_LEVEL="WARNING"((WARNING_GROUPS++))elif [ $TOTAL_LAG -gt 100000 ]; thenHEALTH_STATUS="异常 - 消息严重堆积"HEALTH_LEVEL="WARNING"((WARNING_GROUPS++))elif [ $TOTAL_LAG -gt 10000 ]; thenHEALTH_STATUS="注意 - 消息堆积较多"HEALTH_LEVEL="WARNING"((WARNING_GROUPS++))else((HEALTHY_GROUPS++))fiecho " 健康状态: $HEALTH_STATUS"echo " Health Status: $HEALTH_STATUS" >> "$REPORT_FILE"# 分析消费速率和趋势echo " 消费详情分析:"echo "$GROUP_DETAIL" | awk 'BEGIN {printf " %-20s %-8s %-12s %-12s %-10s %-15s\n", "Topic", "分区", "当前Offset", "最新Offset", "Lag", "消费者ID"print " " sprintf("%*s", 83, "") | "tr ' ' '-'"}NR>1 && $1 != "" {consumer_id = ($7 == "-") ? "无消费者" : substr($7, 1, 12) "..."lag = ($5 == "-") ? "N/A" : $5printf " %-20s %-8s %-12s %-12s %-10s %-15s\n", substr($1, 1, 20), $2, $3, $4, lag, consumer_id}'# 检查是否有分区没有消费者NO_CONSUMER_PARTITIONS=$(echo "$GROUP_DETAIL" | awk 'NR>1 && $7 == "-"' | wc -l)if [ $NO_CONSUMER_PARTITIONS -gt 0 ]; thenecho " ⚠️ 警告: $NO_CONSUMER_PARTITIONS 个分区没有消费者"fi# 检查消费者分布是否均匀echo " 消费者负载分布:"echo "$GROUP_DETAIL" | awk 'NR>1 && $7 != "-" {consumers[$7]++} END {for (consumer in consumers) {printf " 消费者 %s: %d 个分区\n", substr(consumer, length(consumer)-11), consumers[consumer]}}'elseecho " ❌ 无法获取消费者组详细信息"HEALTH_STATUS="无法获取状态"((CRITICAL_GROUPS++))fiecho "" >> "$REPORT_FILE"echo ""fi
done# 3. 生成健康度统计摘要
echo "3. 消费者组健康度统计摘要"
echo "================================"TOTAL_ANALYZED=$((HEALTHY_GROUPS + WARNING_GROUPS + CRITICAL_GROUPS))
HEALTHY_PERCENTAGE=$(echo "scale=1; $HEALTHY_GROUPS * 100 / $TOTAL_ANALYZED" | bc)
WARNING_PERCENTAGE=$(echo "scale=1; $WARNING_GROUPS * 100 / $TOTAL_ANALYZED" | bc)
CRITICAL_PERCENTAGE=$(echo "scale=1; $CRITICAL_GROUPS * 100 / $TOTAL_ANALYZED" | bc)echo "健康度统计:"
echo " ✅ 健康状态: $HEALTHY_GROUPS 个 (${HEALTHY_PERCENTAGE}%)"
echo " ⚠️ 警告状态: $WARNING_GROUPS 个 (${WARNING_PERCENTAGE}%)"
echo " ❌ 严重异常: $CRITICAL_GROUPS 个 (${CRITICAL_PERCENTAGE}%)"# 写入统计摘要到报告文件
{echo ""echo "SUMMARY:"echo "Healthy Groups: $HEALTHY_GROUPS ($HEALTHY_PERCENTAGE%)"echo "Warning Groups: $WARNING_GROUPS ($WARNING_PERCENTAGE%)"echo "Critical Groups: $CRITICAL_GROUPS ($CRITICAL_PERCENTAGE%)"
} >> "$REPORT_FILE"echo -e "\n详细报告已保存到: $REPORT_FILE"# 4. 生成问题消费者组的修复建议
if [ $WARNING_GROUPS -gt 0 ] || [ $CRITICAL_GROUPS -gt 0 ]; thenecho -e "\n4. 问题消费者组修复建议"echo "================================"cat << 'EOF'
发现异常的消费者组，建议采取以下修复措施:🔧 立即处理措施 (严重异常):
1. 重启消费者应用# 检查消费者应用进程ps aux | grep your-consumer-app# 重启消费者服务systemctl restart your-consumer-service2. 检查消费者配置# 验证bootstrap.servers配置# 检查group.id是否正确# 确认认证信息无误3. 快速恢复消费# 临时启动紧急消费者kafka-console-consumer.sh \--bootstrap-server localhost:9092 \--group emergency-group \--topic your-topic🔧 性能优化措施 (警告状态):
1. 增加消费者实例# 水平扩展消费者# 确保实例数不超过分区数2. 优化消费者配置max.poll.records=1000fetch.min.bytes=50000fetch.max.wait.ms=500session.timeout.ms=300003. 优化业务处理逻辑# 异步处理非关键业务# 批量处理消息# 优化数据库连接池🔧 长期监控措施:
1. 建立消费者监控- 监控消费者组Lag- 监控消费者实例健康状态- 设置告警阈值2. 定期健康检查- 每日执行消费者组健康检查- 分析消费趋势和模式- 预测消费能力需求
EOF
elseecho -e "\n✅ 所有消费者组状态健康，无需特殊处理"
fiecho -e "\n=== 消费者组健康度检查完成 ==="
第二步：消费者Lag深度分析
#!/bin/bash
# Kafka消费者Lag深度分析和趋势预测脚本echo "=== Kafka消费者Lag深度分析报告 ==="# 1. 实时Lag监控和分析
echo "1. 实时消费者Lag监控分析"
echo "================================"# 创建Lag监控数据文件
LAG_DATA_FILE="/tmp/consumer_lag_data_$(date +%s).csv"
echo "Timestamp,ConsumerGroup,Topic,Partition,CurrentOffset,LogEndOffset,Lag,ConsumerID" > "$LAG_DATA_FILE"CONSUMER_GROUPS=$(kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list 2>/dev/null)echo "开始收集Lag数据..."
CURRENT_TIME=$(date +%s)for group in $CONSUMER_GROUPS; doif [ -n "$group" ]; thenecho "收集消费者组 $group 的Lag数据..."GROUP_DETAIL=$(kafka-consumer-groups.sh --bootstrap-server localhost:9092 \--describe --group $group 2>/dev/null)if [ -n "$GROUP_DETAIL" ]; thenecho "$GROUP_DETAIL" | awk -v timestamp="$CURRENT_TIME" -v group="$group" 'NR>1 && $1 != "" {lag = ($5 == "-") ? 0 : $5consumer = ($7 == "-") ? "none" : $7printf "%s,%s,%s,%s,%s,%s,%s,%s\n", timestamp, group, $1, $2, $3, $4, lag, consumer}' >> "$LAG_DATA_FILE"fifi
doneecho "Lag数据收集完成，数据文件: $LAG_DATA_FILE"# 2. Lag统计分析和排名
echo -e "\n2. Lag统计分析和问题排名"
echo "================================"echo "按消费者组统计Lag排名 (前20名):"
awk -F, 'NR>1 {group_lag[$2] += $7; group_partitions[$2]++}
END {for (group in group_lag) {if (group_lag[group] > 0) {printf "%-30s %10d %5d\n", group, group_lag[group], group_partitions[group]}}
}' "$LAG_DATA_FILE" | sort -k2 -nr | head -20 | awk '
BEGIN {printf "%-30s %-12s %-8s\n", "消费者组", "总Lag", "分区数"print sprintf("%*s", 52, "") | "tr ' ' '-'"
}
{printf "%-30s %10s %6s\n", $1, $2, $3
}'echo -e "\n按Topic统计Lag排名 (前15名):"
awk -F, 'NR>1 {topic_lag[$3] += $7; topic_groups[$3]++}
END {for (topic in topic_lag) {if (topic_lag[topic] > 0) {printf "%-25s %10d %3d\n", topic, topic_lag[topic], topic_groups[topic]}}
}' "$LAG_DATA_FILE" | sort -k2 -nr | head -15 | awk '
BEGIN {printf "%-25s %-12s %-10s\n", "Topic", "总Lag", "消费者组数"print sprintf("%*s", 49, "") | "tr ' ' '-'"
}
{printf "%-25s %10s %8s\n", $1, $2, $3
}'# 3. Lag趋势分析（需要历史数据）
echo -e "\n3. Lag趋势分析和预测"
echo "================================"# 模拟历史数据收集（实际环境中应该有持续的数据收集）
HISTORY_DIR="/var/log/kafka-lag-history"
mkdir -p "$HISTORY_DIR"# 保存当前数据到历史目录
cp "$LAG_DATA_FILE" "$HISTORY_DIR/lag_$(date +%Y%m%d_%H%M%S).csv"echo "Lag趋势分析 (基于可用的历史数据):"# 检查是否有历史数据
HISTORY_FILES=$(ls "$HISTORY_DIR"/lag_*.csv 2>/dev/null | tail -5)if [ -n "$HISTORY_FILES" ]; thenecho "发现历史Lag数据文件 $(echo "$HISTORY_FILES" | wc -l) 个"# 分析最近几次的Lag变化趋势echo -e "\n最近的Lag变化趋势 (最多显示5个时间点):"TEMP_TREND_FILE="/tmp/lag_trend_analysis.txt"> "$TEMP_TREND_FILE"for file in $HISTORY_FILES; doTIMESTAMP=$(basename "$file" .csv | cut -d'_' -f2-)READABLE_TIME=$(date -j -f "%Y%m%d_%H%M%S" "$TIMESTAMP" "+%Y-%m-%d %H:%M:%S" 2>/dev/null || echo "$TIMESTAMP")TOTAL_LAG=$(awk -F, 'NR>1 {sum += $7} END {print sum+0}' "$file")echo "$READABLE_TIME $TOTAL_LAG" >> "$TEMP_TREND_FILE"doneecho "时间 总Lag数量 变化趋势"echo "=================================================="PREVIOUS_LAG=0while read timestamp lag; doif [ $PREVIOUS_LAG -eq 0 ]; thenTREND="基准"elseDIFF=$((lag - PREVIOUS_LAG))if [ $DIFF -gt 0 ]; thenTREND="↑ +$DIFF"elif [ $DIFF -lt 0 ]; thenTREND="↓ $DIFF"elseTREND="→ 无变化"fifiprintf "%-18s %12s %s\n" "$timestamp" "$lag" "$TREND"PREVIOUS_LAG=$lagdone < "$TEMP_TREND_FILE"rm -f "$TEMP_TREND_FILE"elseecho "暂无历史数据，建议建立持续的Lag监控机制"# 生成Lag监控脚本LAG_MONITOR_SCRIPT="/tmp/kafka_lag_monitor.sh"cat > "$LAG_MONITOR_SCRIPT" << 'EOF'
#!/bin/bash
# Kafka Lag持续监控脚本
# 建议添加到crontab中每5分钟执行一次HISTORY_DIR="/var/log/kafka-lag-history"
mkdir -p "$HISTORY_DIR"LAG_FILE="$HISTORY_DIR/lag_$(date +%Y%m%d_%H%M%S).csv"
echo "Timestamp,ConsumerGroup,Topic, }NR==2 {printf " %-8s %-8s %-8s %-8s %-8s %-8s %-8s %-8s %-8s\n", $1, $2, $3, $4, $5, $6, $7, $8, $9# 计算使用率eden_used = ($6 > 0) ? ($6/$5)*100 : 0old_used = ($8 > 0) ? ($8/$7)*100 : 0printf " Eden区使用率: %.1f%%, Old区使用率: %.1f%%\n", eden_used, old_used}'echo " GC统计信息:"jstat -gccapacity $KAFKA_PID | awk 'NR==2 {printf " 新生代容量: %.1f MB, 老年代容量: %.1f MB\n", $1/1024, $4/1024}'# GC性能分析echo " GC性能分析:"jstat -gcutil $KAFKA_PID 1 3 | awk 'NR==1 {printf " %s\n", $0}NR>1 {printf " %s\n", $0if ($7 > 80) print " ⚠️ Old区使用率过高: " $7 "%"if ($8 > 5) print " ⚠️ Full GC时间过长: " $8 "ms"}'fi# 文件描述符使用情况echo " 文件描述符使用分析:"if [ -d "/proc/$KAFKA_PID/fd" ]; thenFD_COUNT=$(ls /proc/$KAFKA_PID/fd 2>/dev/null | wc -l)FD_LIMIT=$(cat /proc/$KAFKA_PID/limits 2>/dev/null | grep "Max open files" | awk '{print $4}')FD_USAGE_PERCENT=$((FD_COUNT * 100 / FD_LIMIT))printf " 当前使用: %d/%d (%.1f%%)\n" $FD_COUNT $FD_LIMIT $FD_USAGE_PERCENTif [ $FD_USAGE_PERCENT -gt 80 ]; thenecho " ⚠️ 文件描述符使用率过高，可能影响性能"fifi
elseecho " ❌ 未找到Kafka进程，请检查服务状态"
fi# 2. 磁盘IO性能深度分析
echo -e "\n2. 磁盘IO性能深度分析"
echo "================================"echo "Kafka数据目录磁盘性能分析:"
KAFKA_LOG_DIRS="/kafka-logs" # 根据实际配置修改if [ -d "$KAFKA_LOG_DIRS" ]; then# 磁盘使用情况echo " 磁盘使用情况:"df -h $KAFKA_LOG_DIRS | awk 'NR>1 {printf " 磁盘: %s, 已用: %s, 可用: %s, 使用率: %s\n", $1, $3, $4, $5}'# 各Topic磁盘占用echo " 各Topic磁盘占用TOP 10:"du -sh $KAFKA_LOG_DIRS/* 2>/dev/null | sort -hr | head -10 | \awk '{printf " %-30s %s\n", $2, $1}'# 磁盘IO性能测试echo " 磁盘IO性能测试:"if command -v iostat >/dev/null 2>&1; thenecho " 当前磁盘IO统计 (5秒采样):"iostat -x 1 5 | grep -A 100 "Device" | tail -n +2 | \awk 'NF > 0 && !/Device/ {printf " 设备: %-10s 读取: %6.1f KB/s, 写入: %6.1f KB/s, 利用率: %5.1f%%\n", $1, $6, $7, $10if ($10 > 80) printf " ⚠️ 磁盘利用率过高: %.1f%%\n", $10}'elseecho " 请安装sysstat包获取详细IO统计"fi# 检查磁盘延迟echo " 磁盘延迟测试:"TEST_FILE="$KAFKA_LOG_DIRS/io_test_tmp"WRITE_LATENCY=$(dd if=/dev/zero of=$TEST_FILE bs=1M count=100 oflag=direct 2>&1 | \grep "copied" | awk '{print $8}' | cut -d',' -f1)rm -f $TEST_FILEecho " 磁盘写入延迟: ${WRITE_LATENCY}s (100MB)"
elseecho " ❌ Kafka日志目录不存在: $KAFKA_LOG_DIRS"
fi# 3. 网络性能分析
echo -e "\n3. 网络性能分析"
echo "================================"echo "Kafka网络连接和性能分析:"# 网络连接统计
echo " 网络连接统计:"
KAFKA_PORT=9092
CONNECTION_STATS=$(netstat -an 2>/dev/null | grep :$KAFKA_PORT)
TOTAL_CONNECTIONS=$(echo "$CONNECTION_STATS" | wc -l)
ESTABLISHED_CONNECTIONS=$(echo "$CONNECTION_STATS" | grep ESTABLISHED | wc -l)
LISTEN_CONNECTIONS=$(echo "$CONNECTION_STATS" | grep LISTEN | wc -l)printf " 总连接数: %d, 已建立: %d, 监听: %d\n" $TOTAL_CONNECTIONS $ESTABLISHED_CONNECTIONS $LISTEN_CONNECTIONS# 连接状态分布
echo " 连接状态分布:"
echo "$CONNECTION_STATS" | awk '{print $6}' | sort | uniq -c | \
awk '{printf " %-15s %d 个连接\n", $2, $1}'# 网络带宽使用情况
echo " 网络接口带宽使用情况:"
if command -v sar >/dev/null 2>&1; thensar -n DEV 1 3 | grep -E "(eth|ens|eno)" | tail -3 | \awk '{printf " 接口: %-8s 接收: %8.2f KB/s, 发送: %8.2f KB/s\n", $2, $5, $6}'
elseecho " 请安装sysstat包获取网络统计"
fi# 网络延迟测试
echo " 集群内网络延迟测试:"
BROKERS=$(kafka-broker-api-versions.sh --bootstrap-server localhost:9092 2>/dev/null | awk '{print $1}' | head -3)
for broker in $BROKERS; do# 这里假设broker ID对应的是localhost，实际环境需要根据配置修改LATENCY=$(ping -c 3 localhost 2>/dev/null | tail -1 | awk -F'/' '{print $5}' 2>/dev/null || echo "N/A")printf " 到Broker %s的延迟: %s ms\n" $broker "$LATENCY"
done# 4. Topic和分区性能分析
echo -e "\n4. Topic和分区性能分析"
echo "================================"echo "Topic性能指标分析:"# 获取所有Topic列表
TOPICS=$(kafka-topics.sh --bootstrap-server localhost:9092 --list 2>/dev/null)
TOPIC_COUNT=$(echo "$TOPICS" | wc -l)
echo " 总Topic数量: $TOPIC_COUNT"# 分析Topic分区分布
echo " Topic分区分布分析:"
TOTAL_PARTITIONS=0
for topic in $(echo "$TOPICS" | head -10); do # 只分析前10个Topic避免输出过多TOPIC_DESC=$(kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic $topic 2>/dev/null)PARTITION_COUNT=$(echo "$TOPIC_DESC" | grep "PartitionCount:" | awk '{print $2}')REPLICATION_FACTOR=$(echo "$TOPIC_DESC" | grep "ReplicationFactor:" | awk '{print $4}')printf " %-25s 分区: %3d, 副本: %d\n" $topic $PARTITION_COUNT $REPLICATION_FACTORTOTAL_PARTITIONS=$((TOTAL_PARTITIONS + PARTITION_COUNT))
doneecho " 总分区数: $TOTAL_PARTITIONS"# 分区Leader分布均衡性分析
echo " 分区Leader分布均衡性:"
kafka-topics.sh --bootstrap-server localhost:9092 --describe 2>/dev/null | \
grep -E "Leader: [0-9]+" | awk '{print $6}' | sort | uniq -c | \
awk '{printf " Broker %s: %d 个Leader分区\n", $2, $1}'# 5. JMX指标分析（如果启用）
echo -e "\n5. JMX性能指标分析"
echo "================================"echo "检查JMX指标支持情况:"
JMX_PORT=9999 # 默认JMX端口，根据实际配置修改if netstat -an 2>/dev/null | grep -q ":$JMX_PORT"; thenecho " ✅ JMX端口 $JMX_PORT 已开启"# 这里可以添加具体的JMX指标查询echo " 关键JMX指标说明:"echo " - kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec # 消息接收速率"echo " - kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec # 字节接收速率"echo " - kafka.server:type=BrokerTopicMetrics,name=BytesOutPerSec # 字节发送速率"echo " - kafka.server:type=ReplicaManager,name=LeaderCount # Leader数量"echo " - kafka.server:type=ReplicaManager,name=PartitionCount # 分区数量"echo " - kafka.server:type=RequestMetrics,name=RequestsPerSec # 请求速率"echo " 💡 建议使用Prometheus + JMX Exporter或其他监控工具收集这些指标"
elseecho " ⚠️ JMX端口 $JMX_PORT 未开启，建议启用JMX监控"echo " 启用方法: 在Kafka启动脚本中添加:"echo " export JMX_PORT=9999"echo " export KAFKA_JMX_OPTS='-Dcom.sun.management.jmxremote'"
fiecho -e "\n=== Broker和集群性能分析完成 ==="
第四步：性能优化建议和最佳实践
#!/bin/bash
# Kafka性能优化建议生成脚本echo "=== Kafka性能优化建议和最佳实践 ==="echo "1. 生产者优化建议"
echo "================================"cat << 'EOF'
🚀 生产者性能优化配置:# 高吞吐量配置 (适用于批处理场景)
bootstrap.servers=broker1:9092,broker2:9092,broker3:9092
acks=1 # 平衡可靠性和性能
retries=3 # 重试次数
batch.size=32768 # 32KB批次大小
linger.ms=5 # 等待5ms收集更多消息
compression.type=lz4 # 使用LZ4压缩
buffer.memory=67108864 # 64MB缓冲区
max.in.flight.requests.per.connection=5
enable.idempotence=true # 启用幂等性# 低延迟配置 (适用于实时场景)
acks=1
batch.size=1024 # 较小的批次
linger.ms=0 # 不等待，立即发送
compression.type=none # 不压缩以减少CPU开销
buffer.memory=33554432 # 32MB缓冲区💡 生产者调优要点:
1. batch.size和linger.ms配合使用，平衡吞吐量和延迟
2. compression.type选择：lz4适合高吞吐，snappy平衡性能和压缩率
3. acks=all提供最强一致性，acks=1平衡性能和可靠性
4. enable.idempotence=true避免重复消息
EOFecho -e "\n2. 消费者优化建议"
echo "================================"cat << 'EOF'
🚀 消费者性能优化配置:# 高吞吐量消费配置
bootstrap.servers=broker1:9092,broker2:9092,broker3:9092
group.id=high-throughput-consumer
fetch.min.bytes=50000 # 50KB，减少网络往返
fetch.max.wait.ms=500 # 最大等待500ms
max.poll.records=1000 # 每次拉取1000条消息
max.poll.interval.ms=300000 # 5分钟处理超时
session.timeout.ms=10000 # 10秒会话超时
auto.offset.reset=earliest
enable.auto.commit=false # 手动提交offset# 低延迟消费配置
fetch.min.bytes=1 # 立即返回可用消息
fetch.max.wait.ms=100 # 最大等待100ms
max.poll.records=100 # 每次处理较少消息💡 消费者调优要点:
1. fetch.min.bytes和fetch.max.wait.ms控制消息拉取策略
2. max.poll.records平衡内存使用和处理效率
3. 手动提交offset提供更好的消息处理控制
4. 适当的session.timeout.ms避免频繁rebalance
EOFecho -e "\n3. Broker优化建议"
echo "================================"cat << 'EOF'
🚀 Broker性能优化配置:# server.properties关键配置
num.network.threads=8 # 网络线程数，通常为CPU核数
num.io.threads=16 # IO线程数，通常为CPU核数的2倍
socket.send.buffer.bytes=102400 # 100KB socket发送缓冲区
socket.receive.buffer.bytes=102400 # 100KB socket接收缓冲区
socket.request.max.bytes=104857600 # 100MB最大请求大小# 日志相关配置
num.partitions=12 # 默认分区数
default.replication.factor=3 # 默认副本因子
min.insync.replicas=2 # 最小同步副本数
unclean.leader.election.enable=false # 禁用不干净的leader选举# 日志清理和保留
log.retention.hours=168 # 7天保留期
log.segment.bytes=1073741824 # 1GB日志段大小
log.retention.check.interval.ms=300000 # 5分钟检查一次
log.cleaner.enable=true # 启用日志清理
log.cleanup.policy=delete # 删除策略# 性能相关
replica.fetch.max.bytes=1048576 # 1MB副本拉取大小
message.max.bytes=1000000 # 1MB最大消息大小
replica.socket.timeout.ms=30000 # 30秒副本socket超时
replica.socket.receive.buffer.bytes=65536 # 64KB副本接收缓冲💡 Broker调优要点:
1. 根据硬件配置调整线程数和缓冲区大小
2. 合理设置分区数，一般为消费者数的2-3倍
3. 副本因子至少为3，min.insync.replicas设为副本数-1
4. 根据业务需求调整消息保留策略
EOFecho -e "\n4. JVM优化建议"
echo "================================"cat << 'EOF'
🚀 Kafka JVM优化配置:# 内存配置
export KAFKA_HEAP_OPTS="-Xmx6G -Xms6G" # 堆内存，建议为系统内存的50%
export KAFKA_JVM_PERFORMANCE_OPTS="-XX:+UseG1GC # 使用G1垃圾收集器-XX:MaxGCPauseMillis=20 # 最大GC暂停时间20ms-XX:InitiatingHeapOccupancyPercent=35 # GC触发阈值-XX:+ExplicitGCInvokesConcurrent # 并发执行显式GC-XX:MaxInlineLevel=15 # 内联优化-XX:+UnlockExperimentalVMOptions # 解锁实验性选项-XX:+UseCompressedOops # 压缩对象指针-Djava.awt.headless=true # 无头模式-Dcom.sun.management.jmxremote # 启用JMX-Dcom.sun.management.jmxremote.authenticate=false-Dcom.sun.management.jmxremote.ssl=false-Dcom.sun.management.jmxremote.port=9999
"# GC日志配置
export KAFKA_GC_LOG_OPTS="-XX:+UseGCLogFileRotation-XX:NumberOfGCLogFiles=10-XX:GCLogFileSize=100M-XX:+PrintGC-XX:+PrintGCDetails-XX:+PrintGCTimeStamps-XX:+PrintGCApplicationStoppedTime-Xloggc:/kafka-logs/kafkaServer-gc.log
"💡 JVM调优要点:
1. 堆内存建议设置为系统内存的50%，最大不超过32GB
2. 使用G1GC，设置合理的MaxGCPauseMillis
3. 启用GC日志便于性能分析
4. 定期监控GC频率和暂停时间
EOFecho -e "\n5. 系统级优化建议"
echo "================================"cat << 'EOF'
🚀 操作系统级优化:# 内核参数优化 (/etc/sysctl.conf)
vm.swappiness=1 # 减少swap使用
vm.dirty_background_ratio=5 # 后台刷新dirty page比例
vm.dirty_ratio=60 # 强制刷新dirty page比例
vm.dirty_expire_centisecs=12000 # dirty page过期时间
vm.max_map_count=262144 # 增加内存映射限制net.core.wmem_default=131072 # 默认发送缓冲区
net.core.rmem_default=131072 # 默认接收缓冲区
net.core.wmem_max=2097152 # 最大发送缓冲区
net.core.rmem_max=2097152 # 最大接收缓冲区
net.ipv4.tcp_wmem=4096 12582912 16777216
net.ipv4.tcp_rmem=4096 12582912 16777216
net.core.netdev_max_backlog=5000 # 网络设备队列长度
net.ipv4.tcp_max_syn_backlog=8096 # TCP SYN队列长度# 文件系统优化
# 对于XFS文件系统，在/etc/fstab中添加:
# /dev/sdb1 /kafka-logs xfs defaults,noatime,largeio,inode64 0 0# 对于EXT4文件系统:
# /dev/sdb1 /kafka-logs ext4 defaults,noatime 0 0# ulimit配置 (/etc/security/limits.conf)
kafka soft nofile 100000
kafka hard nofile 100000
kafka soft nproc 32768
kafka hard nproc 32768💡 系统级调优要点:
1. 禁用swap或设置vm.swappiness=1
2. 优化网络参数提高吞吐量
3. 使用noatime挂载选项减少磁盘IO
4. 适当增加文件描述符和进程限制
EOFecho -e "\n=== 性能优化建议完成 ===\n"
🔄 四、分区和副本问题处理
📖 分区和副本机制深入解析
分区（Partition）和副本（Replica）是Kafka实现高可用性和水平扩展的核心机制。理解这两个概念对于诊断和解决Kafka集群问题至关重要。
分区机制：
分区是Topic的基本存储单元，每个分区是一个有序的、不可变的消息序列。分区的数量决定了Topic的并发处理能力，因为每个分区只能被消费者组中的一个消费者实例消费。
副本机制：
每个分区可以有多个副本，分布在不同的Broker上。副本分为Leader和Follower：
- Leader副本：处理所有的读写请求
- Follower副本：从Leader同步数据，作为备份
ISR（In-Sync Replicas）机制：
ISR是与Leader保持同步的副本集合，只有ISR中的副本才能被选举为新的Leader。当Follower副本的延迟超过replica.lag.time.max.ms或消息落后超过replica.lag.max.messages时，会被从ISR中移除。
🔧 分区问题诊断和处理
第一步：分区状态全面检查
#!/bin/bash
# Kafka分区状态深度检查脚本echo "=== Kafka分区状态深度检查报告 ==="# 1. 全集群分区分布概览
echo "1. 集群分区分布分析"
echo "================================"# 获取所有Topic和分区信息
TOPICS=$(kafka-topics.sh --bootstrap-server localhost:9092 --list 2>/dev/null)
TOTAL_TOPICS=$(echo "$TOPICS" | wc -l)
echo "总Topic数量: $TOTAL_TOPICS"# 统计总分区数和副本数
TOTAL_PARTITIONS=0
TOTAL_REPLICAS=0for topic in $TOPICS; doTOPIC_INFO=$(kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic $topic 2>/dev/null)PARTITION_COUNT=$(echo "$TOPIC_INFO" | grep "PartitionCount:" | awk '{print $2}')REPLICATION_FACTOR=$(echo "$TOPIC_INFO" | grep "ReplicationFactor:" | awk '{print $4}')TOTAL_PARTITIONS=$((TOTAL_PARTITIONS + PARTITION_COUNT))TOTAL_REPLICAS=$((TOTAL_REPLICAS + PARTITION_COUNT * REPLICATION_FACTOR))
doneecho "总分区数: $TOTAL_PARTITIONS"
echo "总副本数: $TOTAL_REPLICAS"
echo "平均副本因子: $((TOTAL_REPLICAS / TOTAL_PARTITIONS))"# 2. 分区Leader分布均衡性分析
echo -e "\n2. 分区Leader分布均衡性分析"
echo "================================"echo "各Broker的Leader分区分布:"
# 创建临时文件存储分析结果
TEMP_LEADER_FILE="/tmp/kafka_leader_analysis_$(date +%s).txt"kafka-topics.sh --bootstrap-server localhost:9092 --describe 2>/dev/null | \
grep -E "Leader: [0-9]+" > $TEMP_LEADER_FILEif [ -f "$TEMP_LEADER_FILE" ]; then# 统计各Broker的Leader数量awk '{print $6}' $TEMP_LEADER_FILE | sort | uniq -c | \awk '{broker_leaders[$2] = $1total_leaders += $1broker_count++}END {printf " Broker节点 | Leader数量 | 占比\n"printf " -----------|-----------|------\n"for (broker in broker_leaders) {percentage = (broker_leaders[broker] / total_leaders) * 100printf " %-10s | %-9d | %.1f%%\n", "Broker-" broker, broker_leaders[broker], percentage}# 计算分布均衡度avg_leaders = total_leaders / broker_countmax_deviation = 0for (broker in broker_leaders) {deviation = (broker_leaders[broker] - avg_leaders) / avg_leaders * 100if (deviation < 0) deviation = -deviationif (deviation > max_deviation) max_deviation = deviation}printf "\n 分布均衡度分析:\n"printf " 平均Leader数: %.1f\n", avg_leadersprintf " 最大偏差: %.1f%%\n", max_deviationif (max_deviation > 20) {print " ⚠️ Leader分布不均衡，建议执行分区重新分配"} else {print " ✅ Leader分布相对均衡"}}'rm -f $TEMP_LEADER_FILE
fi# 3. 分区大小和消息分布分析
echo -e "\n3. 分区大小和消息分布分析"
echo "================================"echo "Topic分区大小分析 (Top 10):"
# 分析各Topic的磁盘使用情况
if [ -d "/kafka-logs" ]; thendu -sh /kafka-logs/* 2>/dev/null | sort -hr | head -10 | \awk '{size = $1path = $2gsub(/.*\//, "", path) # 提取文件夹名gsub(/-[0-9]+$/, "", path) # 移除分区号topic_size[path] += $1}END {printf " %-30s 磁盘使用\n", "Topic名称"printf " %s\n", sprintf("%*s", 40, "") | "tr ' ' '-'"for (topic in topic_size) {printf " %-30s %s\n", topic, size}}'
fi# 4. 分区消息数量分析
echo -e "\n分区消息数量分析 (随机抽样5个Topic):"
SAMPLE_TOPICS=$(echo "$TOPICS" | shuf | head -5)for topic in $SAMPLE_TOPICS; doecho " Topic: $topic"# 获取每个分区的消息数量kafka-run-class.sh kafka.tools.GetOffsetShell \--broker-list localhost:9092 --topic $topic --time -1 2>/dev/null | \awk -F: '{partition = $1gsub(/.*-/, "", partition)offset = $2total_messages += offsetpartition_messages[partition] = offsetpartition_count++}END {if (partition_count > 0) {printf " 总消息数: %d, 平均每分区: %.0f\n", total_messages, total_messages/partition_count# 找出消息分布不均衡的分区avg_messages = total_messages / partition_countfor (p in partition_messages) {deviation = (partition_messages[p] - avg_messages) / avg_messages * 100if (deviation > 50 || deviation < -50) {printf " 分区 %s: %d 条消息 (偏差 %.1f%%)\n", p, partition_messages[p], deviation}}}}'
done
第二步：副本同步状态检查
#!/bin/bash
# Kafka副本同步状态深度检查脚本echo "=== Kafka副本同步状态检查报告 ==="# 1. ISR状态全面分析
echo "1. ISR (In-Sync Replicas) 状态分析"
echo "================================"# 创建临时文件存储分析结果
ISR_ANALYSIS_FILE="/tmp/kafka_isr_analysis_$(date +%s).txt"# 获取所有Topic的分区和ISR信息
kafka-topics.sh --bootstrap-server localhost:9092 --describe 2>/dev/null > $ISR_ANALYSIS_FILEif [ -f "$ISR_ANALYSIS_FILE" ]; thenecho "ISR状态统计:"# 分析ISR状态awk '/\tTopic:/ {partition_count++# 提取副本信息replicas_match = match($0, /Replicas: ([0-9,]+)/, replicas_arr)isr_match = match($0, /Isr: ([0-9,]+)/, isr_arr)if (replicas_match && isr_match) {replicas_str = replicas_arr[1]isr_str = isr_arr[1]# 计算副本数和ISR数replicas_count = gsub(/,/, ",", replicas_str) + 1isr_count = gsub(/,/, ",", isr_str) + 1total_replicas += replicas_counttotal_isr += isr_count# 统计ISR不完整的分区if (isr_count < replicas_count) {out_of_sync++if (isr_count == 1) single_replica++} else {in_sync++}}}END {printf " 总分区数: %d\n", partition_countprintf " ISR完整的分区: %d (%.1f%%)\n", in_sync, (in_sync/partition_count)*100printf " ISR不完整的分区: %d (%.1f%%)\n", out_of_sync, (out_of_sync/partition_count)*100printf " 只有单个副本的分区: %d\n", single_replicaprintf " 平均副本数: %.1f\n", total_replicas/partition_countprintf " 平均ISR数: %.1f\n", total_isr/partition_countif (out_of_sync > 0) {print "\n ⚠️ 发现ISR不完整的分区，需要关注！"} else {print "\n ✅ 所有分区ISR状态正常"}}' $ISR_ANALYSIS_FILE# 2. 详细列出ISR异常的分区echo -e "\n2. ISR异常分区详细信息"echo "================================"echo "ISR不完整的分区列表:"awk '/\tTopic:/ {topic_match = match($0, /Topic: ([^\t]+)/, topic_arr)partition_match = match($0, /Partition: ([0-9]+)/, partition_arr)replicas_match = match($0, /Replicas: ([0-9,]+)/, replicas_arr)isr_match = match($0, /Isr: ([0-9,]+)/, isr_arr)if (topic_match && partition_match && replicas_match && isr_match) {topic = topic_arr[1]partition = partition_arr[1]replicas = replicas_arr[1]isr = isr_arr[1]replicas_count = gsub(/,/, ",", replicas) + 1isr_count = gsub(/,/, ",", isr) + 1if (isr_count < replicas_count) {printf " Topic: %-25s Partition: %2s | 副本: [%s] | ISR: [%s]\n", topic, partition, replicas, isrif (isr_count == 1) {printf " ❌ 警告: 只有一个副本在ISR中，存在数据丢失风险\n"}}}}' $ISR_ANALYSIS_FILErm -f $ISR_ANALYSIS_FILE
fi# 3. 副本延迟分析
echo -e "\n3. 副本延迟分析"
echo "================================"echo "检查副本同步延迟情况:"# 分析消费者组的副本延迟（通过UnderReplicatedPartitions指标）
echo " 正在检查UnderReplicatedPartitions指标..."# 这里需要JMX支持，检查JMX是否可用
JMX_PORT=9999
if netstat -an 2>/dev/null | grep -q ":$JMX_PORT"; thenecho " ✅ JMX端口可用，可以获取详细的副本指标"echo " 关键指标说明:"echo " - UnderReplicatedPartitions: 副本数量不足的分区数"echo " - ReplicaMaxLag: 副本最大延迟"echo " - ReplicaMinFetchRate: 副本最小拉取速率"
elseecho " ⚠️ JMX端口 $JMX_PORT 不可用，无法获取详细的副本延迟指标"echo " 建议启用JMX监控以获取更详细的副本状态信息"
fi# 通过日志分析副本同步问题
echo " 检查Kafka日志中的副本同步相关错误:"
if [ -f "/kafka-logs/server.log" ]; thenREPLICA_ERRORS=$(grep -i "replica\|isr\|sync" /kafka-logs/server.log | \grep -E "(ERROR|WARN)" | tail -10)if [ -n "$REPLICA_ERRORS" ]; thenecho " 最近的副本相关错误/警告:"echo "$REPLICA_ERRORS" | while read line; doecho " $line"doneelseecho " ✅ 未发现副本相关的错误信息"fi
elseecho " ❌ 无法找到Kafka日志文件"
fi# 4. 副本分布优化分析
echo -e "\n4. 副本分布优化分析"
echo "================================"echo "副本分布优化建议:"# 检查副本是否均匀分布在所有Broker上
BROKER_COUNT=$(kafka-broker-api-versions.sh --bootstrap-server localhost:9092 2>/dev/null | wc -l)
echo " 集群中Broker数量: $BROKER_COUNT"# 分析副本分布
if [ $BROKER_COUNT -gt 0 ]; thenkafka-topics.sh --bootstrap-server localhost:9092 --describe 2>/dev/null | \grep -E "Replicas: [0-9,]+" | awk '{replicas_match = match($0, /Replicas: ([0-9,]+)/, replicas_arr)if (replicas_match) {replicas = replicas_arr[1]split(replicas, broker_array, ",")for (i in broker_array) {broker_replica_count[broker_array[i]]++total_replicas++}}}END {printf " 各Broker的副本分布:\n"for (broker in broker_replica_count) {percentage = (broker_replica_count[broker] / total_replicas) * 100printf " Broker %s: %d 个副本 (%.1f%%)\n", broker, broker_replica_count[broker], percentage}# 计算分布均衡度broker_count = length(broker_replica_count)avg_replicas = total_replicas / broker_countmax_deviation = 0for (broker in broker_replica_count) {deviation = (broker_replica_count[broker] - avg_replicas) / avg_replicas * 100if (deviation < 0) deviation = -deviationif (deviation > max_deviation) max_deviation = deviation}printf "\n 副本分布均衡度: %.1f%% 最大偏差\n", max_deviationif (max_deviation > 25) {print " ⚠️ 副本分布不均衡，建议执行副本重新分配"} else {print " ✅ 副本分布相对均衡"}}'
fiecho -e "\n=== 副本同步状态检查完成 ==="
第三步：分区重新分配和优化
#!/bin/bash
# Kafka分区重新分配和优化脚本echo "=== Kafka分区重新分配和优化工具 ==="echo "1. 分区重新分配准备"
echo "================================"# 生成分区重新分配建议
cat << 'EOF'
📋 分区重新分配准备工作:1. 分析当前分区分布不均衡的原因:- 新增Broker导致负载不均- 某些Broker下线后分区集中- Topic创建时分区分配不合理2. 制定重新分配方案:- 确定需要迁移的Topic列表- 制定迁移的优先级和时间窗口- 评估迁移对系统性能的影响3. 准备重新分配的JSON配置文件
EOFecho -e "\n2. 自动生成分区重新分配方案"
echo "================================"# 自动生成重新分配建议的脚本
cat << 'EOF'
#!/bin/bash
# 自动生成分区重新分配方案TOPIC_NAME="${1:-all}" # 可以指定特定Topic或处理所有Topic
REBALANCE_FILE="/tmp/partition-reassignment-$(date +%s).json"if [ "$TOPIC_NAME" = "all" ]; thenecho "生成所有Topic的重新分配方案..."# 获取所有Topic列表TOPICS=$(kafka-topics.sh --bootstrap-server localhost:9092 --list)# 创建Topic列表文件TOPIC_LIST_FILE="/tmp/topics-to-move.json"echo '{"topics": [' > $TOPIC_LIST_FILEfirst=truefor topic in $TOPICS; doif [ "$first" = "true" ]; thenecho " {\"topic\": \"$topic\"}" >> $TOPIC_LIST_FILEfirst=falseelseecho " ,{\"topic\": \"$topic\"}" >> $TOPIC_LIST_FILEfidoneecho ']}' >> $TOPIC_LIST_FILE# 生成重新分配方案kafka-reassign-partitions.sh \--bootstrap-server localhost:9092 \--topics-to-move-json-file $TOPIC_LIST_FILE \--broker-list "0,1,2" \--throttle 50000000 \--generate > $REBALANCE_FILEecho "重新分配方案已生成: $REBALANCE_FILE"echo "请查看并确认分配方案后再执行"elseecho "生成Topic '$TOPIC_NAME' 的重新分配方案..."# 单个Topic的重新分配TOPIC_JSON="{\"topics\": [{\"topic\": \"$TOPIC_NAME\"}]}"echo $TOPIC_JSON > /tmp/single-topic.jsonkafka-reassign-partitions.sh \--bootstrap-server localhost:9092 \--topics-to-move-json-file /tmp/single-topic.json \--broker-list "0,1,2" \--throttle 25000000 \--generate > $REBALANCE_FILEecho "Topic '$TOPIC_NAME' 重新分配方案: $REBALANCE_FILE"
fi# 显示生成的分配方案摘要
echo -e "\n分配方案摘要:"
if [ -f "$REBALANCE_FILE" ]; thengrep -A 20 "Proposed partition reassignment configuration" $REBALANCE_FILE | head -20
fiecho -e "\n下一步操作:"
echo "1. 检查生成的分配方案是否合理"
echo "2. 执行命令: kafka-reassign-partitions.sh --execute --reassignment-json-file $REBALANCE_FILE"
echo "3. 监控重新分配进度"
EOFchmod +x /tmp/generate-reassignment.sh
echo "分区重新分配方案生成脚本: /tmp/generate-reassignment.sh"echo -e "\n3. 执行分区重新分配"
echo "================================"cat << 'EOF'
📝 分区重新分配执行步骤:# 步骤1: 执行重新分配
kafka-reassign-partitions.sh \--bootstrap-server localhost:9092 \--execute \--reassignment-json-file /path/to/reassignment.json \--throttle 50000000 # 限制带宽防止影响正常业务# 步骤2: 监控重新分配进度
kafka-reassign-partitions.sh \--bootstrap-server localhost:9092 \--verify \--reassignment-json-file /path/to/reassignment.json# 步骤3: 监控集群性能
watch -n 10 'kafka-topics.sh --bootstrap-server localhost:9092 --describe | grep "Leader:"'⚠️ 注意事项:
1. 重新分配期间会消耗额外的网络带宽和磁盘IO
2. 建议在业务低峰期执行
3. 设置合理的throttle值限制迁移速度
4. 密切监控集群性能指标
EOFecho -e "\n4. 分区重新分配监控脚本"
echo "================================"# 创建重新分配监控脚本
cat > /tmp/monitor-reassignment.sh << 'EOF'
#!/bin/bash
# 分区重新分配监控脚本REASSIGNMENT_FILE="$1"if [ -z "$REASSIGNMENT_FILE" ]; thenecho "用法: $0 <reassignment-json-file>"exit 1
fiecho "开始监控分区重新分配进度..."while true; doecho "=== $(date) ==="# 检查重新分配状态STATUS=$(kafka-reassign-partitions.sh \--bootstrap-server localhost:9092 \--verify \--reassignment-json-file "$REASSIGNMENT_FILE" 2>/dev/null)echo "$STATUS"# 检查是否完成if echo "$STATUS" | grep -q "successfully"; thenecho "✅ 分区重新分配已完成！"breakfi# 检查是否有错误if echo "$STATUS" | grep -q "error\|failed"; thenecho "❌ 分区重新分配遇到错误，请检查！"breakfi# 显示集群状态echo "当前集群分区分布:"kafka-topics.sh --bootstrap-server localhost:9092 --describe 2>/dev/null | \grep "Leader:" | awk '{print $6}' | sort | uniq -c | \awk '{printf " Broker %s: %d 个Leader分区\n", $2, $1}'echo "等待30秒后继续监控..."sleep 30
doneecho "分区重新分配监控结束。"
EOFchmod +x /tmp/monitor-reassignment.sh
echo "重新分配监控脚本: /tmp/monitor-reassignment.sh"echo -e "\n=== 分区和副本问题处理完成 ==="
👥 五、消费者问题排查指南
📖 消费者组和重平衡机制深入解析
Kafka消费者组（Consumer Group）是Kafka实现水平扩展消费的核心机制。理解消费者组的工作原理对于诊断和解决消费相关问题至关重要。
消费者组工作机制：
- 组成员管理：消费者组由一个或多个消费者实例组成，每个消费者实例都有唯一的member.id
- 分区分配：每个分区只能被组内的一个消费者实例消费，但一个消费者可以消费多个分区
- 协调器机制：每个消费者组都有一个Group Coordinator（组协调器）负责管理组成员和分区分配
重平衡（Rebalance）触发条件：
- 新的消费者加入组
- 现有消费者离开组（正常关闭或异常断开）
- 消费者无法在session.timeout.ms时间内发送心跳
- Topic的分区数发生变化
- 消费者调用unsubscribe()方法
重平衡的影响：
- 重平衡期间，整个消费者组停止消费
- 可能导致消息处理的短暂中断
- 频繁的重平衡会严重影响消费性能
🔧 消费者问题全面诊断
第一步：消费者组状态全面检查
#!/bin/bash
# Kafka消费者组状态深度分析脚本echo "=== Kafka消费者组深度分析报告 ==="# 1. 消费者组基础信息收集
echo "1. 消费者组基础信息收集"
echo "================================"# 获取所有消费者组列表
CONSUMER_GROUPS=$(kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list 2>/dev/null)
GROUP_COUNT=$(echo "$CONSUMER_GROUPS" | wc -l)echo "发现消费者组数量: $GROUP_COUNT"if [ $GROUP_COUNT -eq 0 ]; thenecho "❌ 未发现任何消费者组，请检查："echo " - Kafka服务是否正常运行"echo " - 网络连接是否正常"echo " - 是否有消费者应用在运行"exit 1
fi# 2. 详细分析每个消费者组
echo -e "\n2. 消费者组详细状态分析"
echo "================================"for group in $CONSUMER_GROUPS; doecho "--- 消费者组: $group ---"# 获取组的详细信息GROUP_DETAIL=$(kafka-consumer-groups.sh --bootstrap-server localhost:9092 \--describe --group $group 2>/dev/null)if [ -n "$GROUP_DETAIL" ]; then# 分析组状态echo " 基本信息分析:"# 统计分区和消费者信息PARTITION_COUNT=$(echo "$GROUP_DETAIL" | awk 'NR>1' | wc -l)ACTIVE_CONSUMERS=$(echo "$GROUP_DETAIL" | awk 'NR>1 && $7 != "-"' | wc -l)TOTAL_LAG=$(echo "$GROUP_DETAIL" | awk 'NR>1 && $5 != "-" {sum += $5} END {print sum+0}')echo " 总分区数: $PARTITION_COUNT"echo " 活跃消费者: $ACTIVE_CONSUMERS"echo " 总消息堆积: $TOTAL_LAG"# 计算消费者组的健康状态if [ $ACTIVE_CONSUMERS -eq 0 ]; thenecho " 🔴 状态: 无活跃消费者 - 所有分区未被消费"elif [ $ACTIVE_CONSUMERS -lt $PARTITION_COUNT ]; thenecho " 🟡 状态: 部分分区无消费者"elseecho " 🟢 状态: 所有分区都有消费者"fi# 分析消息堆积情况if [ $TOTAL_LAG -gt 100000 ]; thenecho " 🔴 堆积状态: 严重堆积 ($TOTAL_LAG 条消息)"elif [ $TOTAL_LAG -gt 10000 ]; thenecho " 🟡 堆积状态: 中等堆积 ($TOTAL_LAG 条消息)"elif [ $TOTAL_LAG -gt 0 ]; thenecho " 🟢 堆积状态: 轻微堆积 ($TOTAL_LAG 条消息)"elseecho " ✅ 堆积状态: 无堆积"fi# 分析Topic分布TOPICS=$(echo "$GROUP_DETAIL" | awk 'NR>1 {print $1}' | sort | uniq)TOPIC_COUNT=$(echo "$TOPICS" | wc -l)echo " 订阅Topic数: $TOPIC_COUNT"echo " Topic列表: $(echo $TOPICS | tr '\n' ' ')"# 检查是否有异常情况echo " 异常情况检查:"# 检查无消费者的分区NO_CONSUMER_PARTITIONS=$(echo "$GROUP_DETAIL" | awk 'NR>1 && $7 == "-"' | wc -l)if [ $NO_CONSUMER_PARTITIONS -gt 0 ]; thenecho " ⚠️ 发现 $NO_CONSUMER_PARTITIONS 个分区没有消费者"echo "$GROUP_DETAIL" | awk 'NR>1 && $7 == "-" {printf " %s-%s\n", $1, $2}' | head -5fi# 检查消息堆积异常的分区HIGH_LAG_PARTITIONS=$(echo "$GROUP_DETAIL" | awk 'NR>1 && $5 != "-" && $5 > 10000' | wc -l)if [ $HIGH_LAG_PARTITIONS -gt 0 ]; thenecho " ⚠️ 发现 $HIGH_LAG_PARTITIONS 个分区消息堆积严重 (>10K)"echo "$GROUP_DETAIL" | awk 'NR>1 && $5 != "-" && $5 > 10000 {printf " %s-%s: %s 条消息\n", $1, $2, $5}' | head -5fi# 检查消费者分布不均if [ $ACTIVE_CONSUMERS -gt 0 ]; thenCONSUMERS_PER_PARTITION=$(echo "scale=2; $PARTITION_COUNT / $ACTIVE_CONSUMERS" | bc 2>/dev/null || echo "N/A")echo " 平均每个消费者处理分区数: $CONSUMERS_PER_PARTITION"fielseecho " ❌ 无法获取消费者组详细信息"fiecho ""
done# 3. 消费者组重平衡分析
echo "3. 消费者组重平衡分析"
echo "================================"echo "分析Kafka日志中的重平衡记录..."if [ -f "/kafka-logs/server.log" ]; then# 查找重平衡相关日志REBALANCE_LOGS=$(grep -i "rebalance\|joining\|leaving" /kafka-logs/server.log | tail -20)if [ -n "$REBALANCE_LOGS" ]; thenecho "最近的重平衡活动:"echo "$REBALANCE_LOGS" | while read line; doecho " $line"done# 统计重平衡频率TODAY=$(date +%Y-%m-%d)REBALANCE_COUNT=$(grep "$TODAY" /kafka-logs/server.log | grep -i "rebalance" | wc -l)echo -e "\n今日重平衡次数: $REBALANCE_COUNT"if [ $REBALANCE_COUNT -gt 10 ]; thenecho "⚠️ 重平衡过于频繁，可能的原因："echo " - 消费者实例不稳定，频繁上下线"echo " - session.timeout.ms配置过小"echo " - max.poll.interval.ms配置不当"echo " - 网络连接不稳定"fielseecho "✅ 未发现最近的重平衡活动"fi
elseecho "❌ 无法找到Kafka日志文件"
fiecho -e "\n=== 消费者组状态检查完成 ==="
第二步：消费者性能和配置分析
#!/bin/bash
# Kafka消费者性能和配置深度分析脚本echo "=== Kafka消费者性能和配置分析 ==="# 1. 消费者配置分析
echo "1. 消费者配置分析和建议"
echo "================================"cat << 'EOF'
🔧 关键消费者配置参数分析:核心配置参数及其影响:1. fetch.min.bytes (默认: 1)- 作用: 控制消费者从broker拉取的最小数据量- 建议: 高吞吐场景设置为1024-10240，低延迟场景保持默认值1- 影响: 值越大，吞吐量越高但延迟增加2. fetch.max.wait.ms (默认: 500)- 作用: 等待fetch.min.bytes数据的最大时间- 建议: 与fetch.min.bytes配合调整，通常100-1000ms- 影响: 控制延迟的上限3. max.poll.records (默认: 500)- 作用: 单次poll()调用返回的最大记录数- 建议: 根据消息处理能力调整，100-2000- 影响: 过大可能导致超时，过小影响吞吐量4. max.poll.interval.ms (默认: 300000 = 5分钟)- 作用: 两次poll()调用的最大间隔- 建议: 根据业务处理时间调整，确保有足够的处理时间- 影响: 超时会触发重平衡5. session.timeout.ms (默认: 10000 = 10秒)- 作用: 消费者会话超时时间- 建议: 网络稳定时可以设置较小值(6000-15000)- 影响: 过小导致频繁重平衡，过大影响故障检测6. heartbeat.interval.ms (默认: 3000 = 3秒)- 作用: 心跳间隔，通常设置为session.timeout.ms的1/3- 建议: session.timeout.ms / 3- 影响: 影响故障检测的敏感度
EOFecho -e "\n2. 消费者性能测试"
echo "================================"# 创建消费者性能测试脚本
cat > /tmp/consumer-performance-test.sh << 'EOF'
#!/bin/bash
# 消费者性能测试脚本TEST_TOPIC="${1:-performance-test-topic}"
TEST_DURATION="${2:-60}" # 测试持续时间（秒）echo "开始消费者性能测试..."
echo "测试Topic: $TEST_TOPIC"
echo "测试时长: ${TEST_DURATION}秒"# 创建测试Topic（如果不存在）
kafka-topics.sh --bootstrap-server localhost:9092 \--create --topic $TEST_TOPIC \--partitions 12 --replication-factor 3 \--if-not-exists# 先生成测试数据
echo "生成测试数据..."
kafka-producer-perf-test.sh \--topic $TEST_TOPIC \--num-records 100000 \--record-size 1024 \--throughput -1 \--producer-props bootstrap.servers=localhost:9092 \> /dev/null 2>&1echo "开始消费者性能测试..."# 测试不同配置的消费性能
echo "1. 默认配置性能测试:"
timeout ${TEST_DURATION}s kafka-consumer-perf-test.sh \--bootstrap-server localhost:9092 \--topic $TEST_TOPIC \--messages 50000 \--consumer-props group.id=perf-test-1echo -e "\n2. 高吞吐配置性能测试:"
timeout ${TEST_DURATION}s kafka-consumer-perf-test.sh \--bootstrap-server localhost:9092 \--topic $TEST_TOPIC \--messages 50000 \--consumer-props group.id=perf-test-2 \fetch.min.bytes=10240 \fetch.max.wait.ms=1000 \max.poll.records=1000echo -e "\n3. 低延迟配置性能测试:"
timeout ${TEST_DURATION}s kafka-consumer-perf-test.sh \--bootstrap-server localhost:9092 \--topic $TEST_TOPIC \--messages 50000 \--consumer-props group.id=perf-test-3 \fetch.min.bytes=1 \fetch.max.wait.ms=100 \max.poll.records=100# 清理测试Topic
echo -e "\n清理测试数据..."
kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic $TEST_TOPICecho "消费者性能测试完成"
EOFchmod +x /tmp/consumer-performance-test.sh
echo "消费者性能测试脚本: /tmp/consumer-performance-test.sh"echo -e "\n3. 消费者异常诊断"
echo "================================"# 消费者异常诊断脚本
cat << 'EOF'
🔍 常见消费者异常及诊断方法:1. 消费者无法连接到Kafka症状: 连接超时、无法加入消费者组诊断:- 检查网络连接: telnet kafka-broker 9092- 验证bootstrap.servers配置- 检查防火墙和安全组设置- 查看consumer日志中的连接错误2. 消费者频繁重平衡症状: 日志中频繁出现rebalance、rejoining group诊断:- 检查session.timeout.ms是否过小- 验证max.poll.interval.ms是否足够- 分析消费者处理时间是否过长- 检查网络稳定性3. 消费者消息处理超时症状: 消费者被踢出组、重新加入诊断:- 调整max.poll.interval.ms- 优化消息处理逻辑- 考虑异步处理- 减少max.poll.records4. 消费者消息丢失症状: 消息被消费但业务处理失败诊断:- 检查enable.auto.commit设置- 验证手动提交offset的逻辑- 分析异常处理机制- 检查事务处理5. 消费者重复消费症状: 同一消息被处理多次诊断:- 检查offset提交机制- 验证消费者唯一性- 分析重平衡影响- 实现幂等性处理
EOF# 4. 消费者监控脚本
echo -e "\n4. 消费者监控和告警"
echo "================================"cat > /tmp/consumer-monitor.sh << 'EOF'
#!/bin/bash
# 消费者组实时监控脚本ALERT_LAG_THRESHOLD=${1:-50000} # 告警阈值
MONITOR_INTERVAL=${2:-30} # 监控间隔(秒)echo "启动消费者组监控..."
echo "告警阈值: $ALERT_LAG_THRESHOLD 条消息"
echo "监控间隔: $MONITOR_INTERVAL 秒"while true; doclearecho "=== Kafka消费者组监控 $(date) ==="# 获取所有消费者组CONSUMER_GROUPS=$(kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list 2>/dev/null)for group in $CONSUMER_GROUPS; doecho "--- 消费者组: $group ---"# 获取消费者组详情GROUP_INFO=$(kafka-consumer-groups.sh --bootstrap-server localhost:9092 \--describe --group $group 2>/dev/null)if [ -n "$GROUP_INFO" ]; then# 计算总lagTOTAL_LAG=$(echo "$GROUP_INFO" | awk 'NR>1 && $5 != "-" {sum += $5} END {print sum+0}')# 统计分区状态TOTAL_PARTITIONS=$(echo "$GROUP_INFO" | awk 'NR>1' | wc -l)ACTIVE_PARTITIONS=$(echo "$GROUP_INFO" | awk 'NR>1 && $7 != "-"' | wc -l)printf " 分区状态: %d/%d 活跃, 总lag: %d\n" $ACTIVE_PARTITIONS $TOTAL_PARTITIONS $TOTAL_LAG# 检查是否需要告警if [ $TOTAL_LAG -gt $ALERT_LAG_THRESHOLD ]; thenecho " 🚨 警告: 消息堆积严重! ($TOTAL_LAG > $ALERT_LAG_THRESHOLD)"# 显示堆积最严重的分区echo "$GROUP_INFO" | awk 'NR>1 && $5 != "-" && $5 > 1000 {printf " %s-%s: %s\n", $1, $2, $5}' | \sort -k3 -nr | head -3elif [ $ACTIVE_PARTITIONS -lt $TOTAL_PARTITIONS ]; thenecho " ⚠️ 注意: 有分区缺少消费者"elseecho " ✅ 状态正常"fielseecho " ❌ 无法获取消费者组信息"fiecho ""doneecho "按 Ctrl+C 停止监控"sleep $MONITOR_INTERVAL
done
EOFchmod +x /tmp/consumer-monitor.sh
echo "消费者监控脚本: /tmp/consumer-monitor.sh"echo -e "\n=== 消费者问题排查指南完成 ==="
🌐 六、网络和连接问题诊断
📖 Kafka网络架构和连接机制
Kafka作为分布式系统，网络通信是其正常运行的基础。理解Kafka的网络架构对于诊断连接问题至关重要。
Kafka网络组件：
- Broker网络接口：每个Broker监听指定端口（默认9092）接收客户端连接
- 客户端连接池：生产者和消费者维护与Broker的长连接
- 内部通信：Broker之间的副本同步、Controller协调等内部通信
- Zookeeper连接：Broker与Zookeeper之间的协调连接
常见网络问题类型：
- 连接超时：客户端无法建立与Broker的连接
- 连接中断：已建立的连接异常断开
- 网络分区：部分节点间无法通信
- 带宽限制：网络带宽不足影响数据传输
- 防火墙配置：安全策略阻止正常通信
🔧 网络问题全面诊断
第一步：网络连接状态检查
#!/bin/bash
# Kafka网络连接深度诊断脚本echo "=== Kafka网络连接深度诊断报告 ==="# 1. Kafka服务端口状态检查
echo "1. Kafka服务端口状态检查"
echo "================================"KAFKA_PORTS="9092 9093 2181" # 常用端口
KAFKA_HOST="localhost"for port in $KAFKA_PORTS; doecho "检查端口 $port:"# 检查端口监听状态if netstat -tlnp 2>/dev/null | grep -q ":$port "; thenecho " ✅ 端口 $port 正在监听"# 获取监听进程信息PROCESS_INFO=$(netstat -tlnp 2>/dev/null | grep ":$port " | awk '{print $7}')echo " 进程信息: $PROCESS_INFO"# 检查连接数CONNECTION_COUNT=$(netstat -an 2>/dev/null | grep ":$port " | wc -l)ESTABLISHED_COUNT=$(netstat -an 2>/dev/null | grep ":$port " | grep ESTABLISHED | wc -l)echo " 总连接数: $CONNECTION_COUNT"echo " 已建立连接: $ESTABLISHED_COUNT"# 连接状态分布echo " 连接状态分布:"netstat -an 2>/dev/null | grep ":$port " | awk '{print $6}' | sort | uniq -c | \awk '{printf " %-15s %d\n", $2, $1}'elseecho " ❌ 端口 $port 未在监听"# 检查端口被占用情况if lsof -i :$port 2>/dev/null; thenecho " 端口被其他进程占用:"lsof -i :$port 2>/dev/nullelseecho " 端口未被占用，可能是服务未启动"fifi# 测试端口连通性echo " 连通性测试:"if timeout 3 bash -c "</dev/tcp/$KAFKA_HOST/$port" 2>/dev/null; thenecho " ✅ 端口 $port 连通正常"elseecho " ❌ 端口 $port 连接失败"fiecho ""
done# 2. 网络接口和路由检查
echo "2. 网络接口和路由检查"
echo "================================"echo "网络接口状态:"
ip addr show | grep -E "(inet |UP|DOWN)" | \
awk '/^[0-9]/ {printf " 接口: %s ", $2} /inet / {printf "IP: %s ", $2} /UP/ {print "状态: UP"} /DOWN/ {print "状态: DOWN"}'echo -e "\n路由表信息:"
route -n | awk 'NR==1 {print " " $0} NR>1 && $1=="0.0.0.0" {print " 默认网关: " $2}'echo -e "\nDNS配置:"
if [ -f "/etc/resolv.conf" ]; thengrep "nameserver" /etc/resolv.conf | awk '{print " DNS服务器: " $2}'
fi# 3. 防火墙状态检查
echo -e "\n3. 防火墙状态检查"
echo "================================"# 检查iptables
if command -v iptables >/dev/null 2>&1; thenecho "iptables状态:"IPTABLES_RULES=$(iptables -L 2>/dev/null | wc -l)if [ $IPTABLES_RULES -gt 10 ]; thenecho " ⚠️ 检测到iptables规则，可能影响Kafka连接"echo " 相关规则:"iptables -L | grep -E "(9092|9093|2181|REJECT|DROP)" | head -5 | sed 's/^/ /'elseecho " ✅ iptables规则较少，影响较小"fi
fi# 检查firewalld (CentOS/RHEL)
if command -v firewall-cmd >/dev/null 2>&1; thenif systemctl is-active firewalld >/dev/null 2>&1; thenecho "firewalld状态:"echo " 状态: 运行中"echo " 当前区域: $(firewall-cmd --get-active-zones | head -1)"# 检查Kafka端口是否开放for port in 9092 9093 2181; doif firewall-cmd --query-port=$port/tcp 2>/dev/null; thenecho " ✅ 端口 $port/tcp 已开放"elseecho " ❌ 端口 $port/tcp 未开放"fidoneelseecho "firewalld: 未运行"fi
fi# 检查ufw (Ubuntu)
if command -v ufw >/dev/null 2>&1; thenUFW_STATUS=$(ufw status 2>/dev/null | head -1)echo "ufw状态: $UFW_STATUS"if echo "$UFW_STATUS" | grep -q "active"; thenecho " ⚠️ ufw防火墙激活，请检查Kafka端口是否开放"fi
fiecho -e "\n=== 网络连接状态检查完成 ==="
第二步：网络性能和延迟分析
#!/bin/bash
# Kafka网络性能和延迟分析脚本echo "=== Kafka网络性能和延迟分析 ==="# 1. 网络延迟测试
echo "1. 网络延迟测试"
echo "================================"# 获取Kafka集群节点列表（这里以localhost为例，实际环境需要修改）
KAFKA_NODES="localhost" # 实际环境中应该是: "kafka1 kafka2 kafka3"echo "Kafka集群节点网络延迟测试:"
for node in $KAFKA_NODES; doecho " 测试到节点 $node 的延迟:"if ping -c 5 $node >/dev/null 2>&1; thenPING_RESULT=$(ping -c 5 $node 2>/dev/null | tail -1)echo " $PING_RESULT"# 提取平均延迟AVG_LATENCY=$(echo "$PING_RESULT" | awk -F'/' '{print $5}' 2>/dev/null)if [ -n "$AVG_LATENCY" ]; then# 延迟评估if awk "BEGIN {exit !($AVG_LATENCY < 1)}"; thenecho " ✅ 延迟优秀 (<1ms)"elif awk "BEGIN {exit !($AVG_LATENCY < 10)}"; thenecho " 🟢 延迟良好 (<10ms)"elif awk "BEGIN {exit !($AVG_LATENCY < 50)}"; thenecho " 🟡 延迟一般 (<50ms)"elseecho " 🔴 延迟较高 (>50ms)，可能影响性能"fifielseecho " ❌ 无法ping通节点 $node"fi# 测试TCP连接延迟echo " TCP连接延迟测试 (端口9092):"TCP_TEST_START=$(date +%s%N)if timeout 3 bash -c "</dev/tcp/$node/9092" 2>/dev/null; thenTCP_TEST_END=$(date +%s%N)TCP_LATENCY=$(((TCP_TEST_END - TCP_TEST_START) / 1000000))echo " TCP连接耗时: ${TCP_LATENCY}ms"if [ $TCP_LATENCY -lt 10 ]; thenecho " ✅ TCP连接速度优秀"elif [ $TCP_LATENCY -lt 50 ]; thenecho " 🟢 TCP连接速度良好"elseecho " 🟡 TCP连接速度较慢"fielseecho " ❌ TCP连接失败"fiecho ""
done# 2. 网络带宽测试
echo "2. 网络带宽和吞吐量分析"
echo "================================"echo "网络接口带宽使用情况:"
if command -v sar >/dev/null 2>&1; thenecho " 当前网络流量 (最近5秒平均):"sar -n DEV 1 5 | grep -E "(IFACE|eth|ens|eno)" | tail -n +2 | \awk 'NF>6 {if ($2 != "IFACE") {printf " 接口: %-8s 接收: %8.2f KB/s, 发送: %8.2f KB/s\n", $2, $5, $6}}'echo -e "\n 网络错误统计:"sar -n EDEV 1 3 | grep -E "(IFACE|eth|ens|eno)" | tail -n +2 | \awk 'NF>6 {if ($2 != "IFACE" && ($3+$4+$5+$6) > 0) {printf " 接口: %-8s 接收错误: %d, 发送错误: %d\n", $2, $3+$5, $4+$6}}'
elseecho " 请安装sysstat包以获取详细的网络统计信息"echo " 当前网络接口统计:"cat /proc/net/dev | awk 'NR>2 {gsub(/:/, " ", $1)if ($2 > 0 || $10 > 0) {printf " %-8s 接收: %10d bytes, 发送: %10d bytes\n", $1, $2, $10}}'
fi# 3. 网络连接池分析
echo -e "\n3. Kafka网络连接池分析"
echo "================================"echo "当前Kafka相关网络连接分析:"# 分析Kafka端口的连接情况
for port in 9092 9093; doif netstat -an 2>/dev/null | grep -q ":$port "; thenecho " 端口 $port 连接分析:"# 连接状态统计netstat -an 2>/dev/null | grep ":$port " | awk '{print $6}' | sort | uniq -c | \awk '{printf " %-15s %3d 个连接\n", $2, $1}'# 外部连接来源分析echo " 主要连接来源:"netstat -an 2>/dev/null | grep ":$port " | grep ESTABLISHED | \awk '{print $5}' | cut -d: -f1 | sort | uniq -c | sort -nr | head -5 | \awk '{printf " %-15s %3d 个连接\n", $2, $1}'# 检查连接数是否异常TOTAL_CONNECTIONS=$(netstat -an 2>/dev/null | grep ":$port " | wc -l)if [ $TOTAL_CONNECTIONS -gt 1000 ]; thenecho " ⚠️ 连接数较多 ($TOTAL_CONNECTIONS)，请注意监控"elif [ $TOTAL_CONNECTIONS -gt 5000 ]; thenecho " 🔴 连接数过多 ($TOTAL_CONNECTIONS)，可能存在连接泄露"fifi
done# 4. 系统网络参数检查
echo -e "\n4. 系统网络参数检查"
echo "================================"echo "关键网络内核参数:"
NETWORK_PARAMS=("net.core.rmem_max""net.core.wmem_max""net.core.rmem_default""net.core.wmem_default""net.ipv4.tcp_rmem""net.ipv4.tcp_wmem""net.core.netdev_max_backlog""net.ipv4.tcp_max_syn_backlog""net.core.somaxconn"
)for param in "${NETWORK_PARAMS[@]}"; doif [ -f "/proc/sys/${param//./\/}" ]; thenVALUE=$(cat "/proc/sys/${param//./\/}" 2>/dev/null)printf " %-30s %s\n" "$param:" "$VALUE"fi
done# 网络参数优化建议
echo -e "\n网络参数优化建议:"
echo " 对于高并发Kafka集群，建议调整以下参数:"
echo " net.core.rmem_max = 134217728 # 128MB"
echo " net.core.wmem_max = 134217728 # 128MB"
echo " net.core.netdev_max_backlog = 5000 # 网络设备队列长度"
echo " net.ipv4.tcp_max_syn_backlog = 8096 # TCP SYN队列长度"
echo " net.core.somaxconn = 32768 # 监听队列长度"echo -e "\n=== 网络性能和延迟分析完成 ==="
第三步：网络问题解决方案
#!/bin/bash
# Kafka网络问题解决方案和修复工具echo "=== Kafka网络问题解决方案 ==="# 1. 网络连接问题自动诊断和修复
echo "1. 网络连接问题自动诊断"
echo "================================"cat << 'EOF'
🔧 网络连接问题诊断和解决流程:常见网络问题及解决方案:1. 连接超时问题症状: 客户端连接Kafka超时诊断步骤:a) 检查网络连通性: ping kafka-brokerb) 检查端口开放: telnet kafka-broker 9092c) 验证防火墙配置d) 检查Kafka服务状态解决方案:- 调整客户端connection.timeout.ms参数- 检查和修复网络配置- 更新防火墙规则开放必要端口- 重启Kafka服务2. 频繁连接断开症状: 连接建立后异常断开，客户端重连诊断步骤:a) 查看Kafka server.log中的连接错误b) 检查网络稳定性c) 分析客户端日志d) 监控系统资源使用解决方案:- 调整socket.connection.setup.timeout.ms- 优化网络配置和路由- 增加客户端重试机制- 调整系统文件描述符限制3. 网络分区问题症状: 部分节点无法通信，集群脑裂诊断:- 检查网络设备状态- 验证路由配置- 测试节点间连通性解决方案:- 修复网络基础设施- 调整网络拓扑- 配置网络冗余
EOF# 2. 自动化网络修复脚本
echo -e "\n2. 自动化网络修复工具"
echo "================================"# 创建网络问题自动修复脚本
cat > /tmp/kafka-network-fix.sh << 'EOF'
#!/bin/bash
# Kafka网络问题自动修复脚本echo "开始 Kafka 网络问题诊断和修复..."# 1. 检查和修复Kafka服务
echo "1. 检查Kafka服务状态..."
if ! pgrep -f kafka.Kafka >/dev/null; thenecho "⚠️ Kafka进程未运行，尝试启动..."# 这里需要根据实际环境修改启动命令# systemctl start kafkaecho "请手动启动Kafka服务"
elseecho "✅ Kafka进程正常运行"
fi# 2. 检查端口状态
echo "2. 检查端口状态..."
for port in 9092 2181; doif netstat -tlnp 2>/dev/null | grep -q ":$port "; thenecho "✅ 端口 $port 正常监听"elseecho "❌ 端口 $port 未监听，检查服务配置"fi
done# 3. 检查防火墙配置
echo "3. 检查防火墙配置..."
if command -v firewall-cmd >/dev/null 2>&1 && systemctl is-active firewalld >/dev/null 2>&1; thenecho "检查firewalld配置..."for port in 9092 9093 2181; doif ! firewall-cmd --query-port=$port/tcp 2>/dev/null; thenecho "正在开放端口 $port..."firewall-cmd --permanent --add-port=$port/tcpfirewall-cmd --reloadfidone
elif command -v ufw >/dev/null 2>&1; thenecho "检查ufw配置..."if ufw status | grep -q "Status: active"; thenfor port in 9092 9093 2181; doufw allow $port/tcpdonefi
fi# 4. 优化网络内核参数
echo "4. 检查网络内核参数..."
SYSCTL_OPTIMIZATIONS=("net.core.rmem_max=134217728""net.core.wmem_max=134217728""net.core.netdev_max_backlog=5000""net.ipv4.tcp_max_syn_backlog=8096"
)for param in "${SYSCTL_OPTIMIZATIONS[@]}"; dokey=$(echo $param | cut -d'=' -f1)value=$(echo $param | cut -d'=' -f2)current_value=$(sysctl -n $key 2>/dev/null)if [ "$current_value" != "$value" ]; thenecho "优化参数: $param"echo $param >> /etc/sysctl.confsysctl -p >/dev/null 2>&1fi
done# 5. 检查文件描述符限制
echo "5. 检查文件描述符限制..."
CURRENT_LIMIT=$(ulimit -n)
if [ $CURRENT_LIMIT -lt 65536 ]; thenecho "⚠️ 文件描述符限制过小: $CURRENT_LIMIT"echo "建议在/etc/security/limits.conf中添加:"echo "* soft nofile 65536"echo "* hard nofile 65536"
fiecho "网络问题修复完成！"
EOFchmod +x /tmp/kafka-network-fix.sh
echo "网络问题修复脚本: /tmp/kafka-network-fix.sh"echo -e "\n3. 网络监控和告警"
echo "================================"# 创建网络监控脚本
cat > /tmp/kafka-network-monitor.sh << 'EOF'
#!/bin/bash
# Kafka网络实时监控脚本MONITOR_INTERVAL=${1:-30} # 监控间隔（秒）
LOG_FILE="/var/log/kafka-network-monitor.log"echo "启动Kafka网络监控..."
echo "监控间隔: $MONITOR_INTERVAL 秒"
echo "日志文件: $LOG_FILE"while true; doTIMESTAMP=$(date '+%Y-%m-%d %H:%M:%S')# 检查Kafka端口状态for port in 9092 2181; doif ! netstat -tlnp 2>/dev/null | grep -q ":$port "; thenMESSAGE="[$TIMESTAMP] ALERT: Kafka端口 $port 不可用"echo "$MESSAGE" | tee -a $LOG_FILE# 这里可以集成告警系统fidone# 检查网络连接数CONNECTION_COUNT=$(netstat -an 2>/dev/null | grep ":9092 " | wc -l)if [ $CONNECTION_COUNT -gt 5000 ]; thenMESSAGE="[$TIMESTAMP] WARNING: Kafka连接数过多: $CONNECTION_COUNT"echo "$MESSAGE" | tee -a $LOG_FILEfi# 检查网络延迟PING_LATENCY=$(ping -c 1 localhost 2>/dev/null | tail -1 | awk -F'/' '{print $5}' 2>/dev/null)if [ -n "$PING_LATENCY" ] && awk "BEGIN {exit !($PING_LATENCY > 100)}"; thenMESSAGE="[$TIMESTAMP] WARNING: 网络延迟过高: ${PING_LATENCY}ms"echo "$MESSAGE" | tee -a $LOG_FILEfi# 输出当前状态echo "[$TIMESTAMP] Kafka网络状态正常 - 连接数: $CONNECTION_COUNT"sleep $MONITOR_INTERVAL
done
EOFchmod +x /tmp/kafka-network-monitor.sh
echo "网络监控脚本: /tmp/kafka-network-monitor.sh"echo -e "\n=== 网络和连接问题诊断完成 ==="---## 📊 七、专业监控脚本工具### 📖 Kafka监控体系架构Kafka生产环境需要完善的监控体系来保证系统的稳定运行。一个完整的Kafka监控体系应该包含以下几个层面：**系统层监控：**
- **硬件资源**：CPU、内存、磁盘、网络的使用情况
- **操作系统**：进程状态、文件描述符、网络连接数
- **JVM状态**：堆内存使用、GC活动、线程数量**Kafka应用层监控：**
- **Broker指标**：消息吞吐量、分区状态、副本同步
- **Topic指标**：消息产生速率、消费速率、分区分布
- **消费者组**：消息堆积、重平衡次数、消费延迟**业务层监控：**
- **数据质量**：消息丢失率、重复消费率
- **服务水平**：处理延迟、错误率、可用性
- **容量规划**：存储增长趋势、流量预测### 🔧 综合监控仪表板#### 第一部分：集群健康状态监控```bash
#!/bin/bash
# Kafka集群健康状态综合监控仪表板echo "=== Kafka集群健康状态仪表板 ==="# 面板配置
DASHBOARD_REFRESH_INTERVAL=10 # 刷新间隔（秒）
ALERT_LAG_THRESHOLD=50000 # 消息堆积告警阈值
ALERT_DISK_THRESHOLD=80 # 磁盘使用率告警阈值# 颜色定义
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
BLUE='\033[0;34m'
NC='\033[0m' # No Color# 显示函数
show_status() {local status=$1local message=$2case $status in"OK") echo -e "${GREEN}✓${NC} $message" ;;"WARNING") echo -e "${YELLOW}⚠${NC} $message" ;;"ERROR") echo -e "${RED}✗${NC} $message" ;;"INFO") echo -e "${BLUE}ⓘ${NC} $message" ;;esac
}# 显示仪表板
show_dashboard() {clearecho -e "${BLUE}"echo " _ __ __ _ ____ _ _ _ "echo " | |/ / __ _ / _|| | __ __ _ | _ \\ __ _ ___ | |__ | |__ ___ __ _ _ __ __| | "echo " | ' / / _\` || |_ | |/ / / _\` || | | | / _\` |/ __|| '_ \\| '_ \\ / _ \\ / _\` || '__|/ _\` | "echo " | . \\| (_| || _|| < | (_| || |_| || (_| |\\__ \\| | | || |_) || (_) || (_| || | | (_| | "echo " |_|\\_\\\\__,_||_| |_|\\_\\\\__,_||____/ \\__,_||___/|_| |_||_.__/ \\___/ \\__,_||_| \\__,_| "echo -e "${NC}"echo -e "\n✨ Kafka集群实时状态监控 - $(date '+%Y-%m-%d %H:%M:%S')"echo "================================================================================="# 1. 集群基本信息echo -e "\n🏠 集群基本信息"echo "--------------------------------------------------------------------------------"# Broker数量和状态BROKER_COUNT=$(kafka-broker-api-versions.sh --bootstrap-server localhost:9092 2>/dev/null | wc -l)if [ $BROKER_COUNT -gt 0 ]; thenshow_status "OK" "Broker节点数量: $BROKER_COUNT"# 检查每个Broker的状态kafka-broker-api-versions.sh --bootstrap-server localhost:9092 2>/dev/null | while read line; dobroker_id=$(echo $line | awk '{print $1}')echo " • Broker $broker_id: 连接正常"doneelseshow_status "ERROR" "无法连接到Kafka集群"fi# Topic和分区统计TOPIC_COUNT=$(kafka-topics.sh --bootstrap-server localhost:9092 --list 2>/dev/null | wc -l)TOTAL_PARTITIONS=$(kafka-topics.sh --bootstrap-server localhost:9092 --describe 2>/dev/null | \grep "PartitionCount" | awk '{sum += $2} END {print sum+0}')show_status "INFO" "Topic数量: $TOPIC_COUNT, 总分区数: $TOTAL_PARTITIONS"# 2. 系统资源状态echo -e "\n📊 系统资源状态"echo "--------------------------------------------------------------------------------"# CPU使用率CPU_USAGE=$(top -bn1 | grep "Cpu(s)" | awk '{print $2}' | cut -d'%' -f1)if awk "BEGIN {exit !($CPU_USAGE < 80)}"; thenshow_status "OK" "CPU使用率: ${CPU_USAGE}%"elseshow_status "WARNING" "CPU使用率较高: ${CPU_USAGE}%"fi# 内存使用率MEMORY_INFO=$(free | grep Mem)MEMORY_TOTAL=$(echo $MEMORY_INFO | awk '{print $2}')MEMORY_USED=$(echo $MEMORY_INFO | awk '{print $3}')MEMORY_USAGE=$((MEMORY_USED * 100 / MEMORY_TOTAL))if [ $MEMORY_USAGE -lt 80 ]; thenshow_status "OK" "内存使用率: ${MEMORY_USAGE}%"elseshow_status "WARNING" "内存使用率较高: ${MEMORY_USAGE}%"fi# 磁盘使用率if [ -d "/kafka-logs" ]; thenDISK_USAGE=$(df /kafka-logs | tail -1 | awk '{print $5}' | cut -d'%' -f1)if [ $DISK_USAGE -lt $ALERT_DISK_THRESHOLD ]; thenshow_status "OK" "Kafka数据目录磁盘使用率: ${DISK_USAGE}%"elseshow_status "WARNING" "Kafka数据目录磁盘使用率较高: ${DISK_USAGE}%"fifi# 3. Kafka服务状态echo -e "\n🚀 Kafka服务状态"echo "--------------------------------------------------------------------------------"# Kafka进程状态if pgrep -f kafka.Kafka >/dev/null; thenKAFKA_PID=$(pgrep -f kafka.Kafka | head -1)KAFKA_UPTIME=$(ps -o etime= -p $KAFKA_PID | tr -d ' ')show_status "OK" "Kafka进程运行正常 (PID: $KAFKA_PID, 运行时间: $KAFKA_UPTIME)"# JVM内存状态if command -v jstat >/dev/null 2>&1; thenJVM_HEAP=$(jstat -gc $KAFKA_PID | awk 'NR==2 {printf "%.1f", ($3+$4+$6+$8)/($2+$5+$7)*100}')if awk "BEGIN {exit !($JVM_HEAP < 80)}"; thenshow_status "OK" "JVM堆内存使用率: ${JVM_HEAP}%"elseshow_status "WARNING" "JVM堆内存使用率较高: ${JVM_HEAP}%"fifielseshow_status "ERROR" "Kafka进程未运行"fi# 网络连接状态KAFKA_CONNECTIONS=$(netstat -an 2>/dev/null | grep ":9092 " | grep ESTABLISHED | wc -l)show_status "INFO" "Kafka当前连接数: $KAFKA_CONNECTIONS"# 4. 消费者组状态echo -e "\n👥 消费者组状态"echo "--------------------------------------------------------------------------------"CONSUMER_GROUPS=$(kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list 2>/dev/null)GROUP_COUNT=$(echo "$CONSUMER_GROUPS" | wc -l)if [ $GROUP_COUNT -gt 0 ]; thenshow_status "INFO" "消费者组数量: $GROUP_COUNT"# 分析每个消费者组的状态for group in $(echo "$CONSUMER_GROUPS" | head -5); do # 只显示前5个组GROUP_LAG=$(kafka-consumer-groups.sh --bootstrap-server localhost:9092 \--describe --group $group 2>/dev/null | \awk 'NR>1 && $5 != "-" {sum += $5} END {print sum+0}')if [ $GROUP_LAG -eq 0 ]; thenshow_status "OK" "组 $group: 无堆积"elif [ $GROUP_LAG -lt $ALERT_LAG_THRESHOLD ]; thenshow_status "INFO" "组 $group: 堆积 ${GROUP_LAG} 条消息"elseshow_status "WARNING" "组 $group: 严重堆积 ${GROUP_LAG} 条消息"fidoneif [ $GROUP_COUNT -gt 5 ]; thenecho " ... 还有 $((GROUP_COUNT - 5)) 个消费者组"fielseshow_status "INFO" "暂无消费者组"fi# 5. 最近的错误日志echo -e "\n📝 最近的系统日志"echo "--------------------------------------------------------------------------------"if [ -f "/kafka-logs/server.log" ]; thenRECENT_ERRORS=$(grep -E "(ERROR|WARN)" /kafka-logs/server.log | tail -3)if [ -n "$RECENT_ERRORS" ]; thenecho "$RECENT_ERRORS" | while read line; doshow_status "WARNING" "$line"doneelseshow_status "OK" "最近无错误日志"fielseshow_status "INFO" "日志文件不存在"fi# 底部信息echo -e "\n⚙️ 控制: Ctrl+C 退出 | 自动刷新间隔: ${DASHBOARD_REFRESH_INTERVAL}秒"echo "================================================================================="
}# 主循环
echo "启动Kafka集群健康状态仪表板..."while true; doshow_dashboardsleep $DASHBOARD_REFRESH_INTERVAL
done
第二部分：消息流量监控
#!/bin/bash
# Kafka消息流量实时监控脚本echo "=== Kafka消息流量实时监控 ==="# 初始化参数
MONITOR_DURATION=${1:-300} # 监控时长（秒）
SAMPLE_INTERVAL=${2:-10} # 采样间隔（秒）
STATISTICS_FILE="/tmp/kafka_traffic_stats_$(date +%Y%m%d_%H%M%S).csv"echo "监控参数: 时长=${MONITOR_DURATION}秒, 采样间隔=${SAMPLE_INTERVAL}秒"
echo "统计数据将保存到: $STATISTICS_FILE"# 创建 CSV 文件头
echo "Timestamp,Topic,Partition,StartOffset,EndOffset,MessageCount,GrowthRate" > $STATISTICS_FILE# 获取所有Topic列表
TOPICS=$(kafka-topics.sh --bootstrap-server localhost:9092 --list 2>/dev/null)if [ -z "$TOPICS" ]; thenecho "❌ 无法获取Topic列表，请检查Kafka服务状态"exit 1
fiecho "检测到 $(echo "$TOPICS" | wc -l) 个Topic"# 初始化上一次的offset记录
declare -A LAST_OFFSETS# 获取初始 offset
echo "正在初始化基线数据..."
for topic in $TOPICS; dokafka-run-class.sh kafka.tools.GetOffsetShell \--broker-list localhost:9092 --topic $topic --time -1 2>/dev/null | \while IFS=':' read partition offset; dokey="${topic}_${partition}"LAST_OFFSETS[$key]=$offsetdone
doneecho "基线数据初始化完成，开始监控..."# 监控主循环
START_TIME=$(date +%s)
CYCLE_COUNT=0while [ $(($(date +%s) - START_TIME)) -lt $MONITOR_DURATION ]; doCYCLE_COUNT=$((CYCLE_COUNT + 1))CURRENT_TIME=$(date '+%Y-%m-%d %H:%M:%S')clearecho "=== Kafka消息流量实时监控 - $CURRENT_TIME (第${CYCLE_COUNT}次采样) ==="echo "================================================================================="# 表头printf "%-25s %-5s %12s %12s %12s %15s\n" "Topic" "Part" "StartOffset" "EndOffset" "MsgCount" "Rate(msg/s)"echo "--------------------------------------------------------------------------------"TOTAL_MESSAGES=0TOTAL_RATE=0for topic in $TOPICS; do# 获取当前 offsetkafka-run-class.sh kafka.tools.GetOffsetShell \--broker-list localhost:9092 --topic $topic --time -1 2>/dev/null | \while IFS=':' read partition current_offset; dokey="${topic}_${partition}"# 获取上次的 offsetlast_offset=${LAST_OFFSETS[$key]:-0}# 计算消息增量和速率message_count=$((current_offset - last_offset))rate=$((message_count / SAMPLE_INTERVAL))# 保存数据到CSVecho "$CURRENT_TIME,$topic,$partition,$last_offset,$current_offset,$message_count,$rate" >> $STATISTICS_FILE# 显示数据if [ $message_count -gt 0 ] || [ $current_offset -gt 0 ]; thenprintf "%-25s %-5s %12s %12s %12s %15s\n" \"$topic" "$partition" "$last_offset" "$current_offset" "$message_count" "$rate"TOTAL_MESSAGES=$((TOTAL_MESSAGES + message_count))TOTAL_RATE=$((TOTAL_RATE + rate))fi# 更新上次offsetLAST_OFFSETS[$key]=$current_offsetdonedoneecho "--------------------------------------------------------------------------------"printf "%-25s %-5s %12s %12s %12s %15s\n" "TOTAL" "-" "-" "-" "$TOTAL_MESSAGES" "$TOTAL_RATE"# 显示统计信息echo -e "\n📊 实时统计:"echo " 本轮采样消息增量: $TOTAL_MESSAGES 条"echo " 消息产生率: $TOTAL_RATE 条/秒"echo " 累计采样次数: $CYCLE_COUNT"echo " 剩余监控时间: $((MONITOR_DURATION - ($(date +%s) - START_TIME)))秒"# 等待下一次采样sleep $SAMPLE_INTERVAL
doneecho -e "\n监控完成！统计数据已保存到: $STATISTICS_FILE"# 生成统计报告
echo -e "\n正在生成统计报告..."# TOP 10 消息产生最活跃的Topic
echo -e "\n🏆 消息产生最活跃的Topic (TOP 10):"
awk -F',' 'NR>1 {topic_msg[$2] += $6} END {for (topic in topic_msg) {print topic_msg[topic], topic}
}' $STATISTICS_FILE | sort -nr | head -10 | \
awk '{printf " %-25s %10d 条消息\n", $2, $1}'# 平均消息产生速率
echo -e "\n📊 平均消息产生速率:"
AVG_RATE=$(awk -F',' 'NR>1 {sum += $7; count++} END {if(count>0) printf "%.2f", sum/count; else print 0}' $STATISTICS_FILE)
echo " 全集群平均消息产生速率: $AVG_RATE 条/秒"echo -e "\n详细统计数据请查看: $STATISTICS_FILE"
🆘 八、紧急故障处理预案
📖 紧急故障处理流程
Kafka生产环境中的紧急故障需要快速、准确的处理。有效的应急预案可以显著减少故障影响范围和恢复时间。
故障等级定义：
- P0 - 致命故障：整个集群不可用，对业务造成严重影响
- P1 - 重大故障：部分功能受影响，但系统基本可用
- P2 - 一般故障：性能降低或部分功能异常
- P3 - 轻微故障：对业务影响很小，可正常维护时间处理
🚑 紧急故障处理工具包
第一部分：快速诊断工具
#!/bin/bash
# Kafka紧急故障快速诊断工具echo "=== Kafka紧急故障快速诊断工具 ==="# 设置紧急模式
EMERGENCY_MODE=true
QUICK_CHECK_TIMEOUT=5 # 快速检查超时时间（秒）echo "🚑 紧急模式启动 - $(date)"
echo "================================================================================="# 快速故障等级评估
echo "1. 快速故障等级评估"
echo "--------------------------------------------------------------------------------"FAULT_LEVEL="P3" # 默认轻微故障
FAULT_ISSUES=()# 检查Kafka进程状态
if ! pgrep -f kafka.Kafka >/dev/null; thenFAULT_LEVEL="P0"FAULT_ISSUES+=("Kafka进程未运行")echo "🔴 P0 致命: Kafka进程未运行"
elseecho "✅ Kafka进程运行正常"
fi# 检查端口状态
for port in 9092 2181; doif ! timeout $QUICK_CHECK_TIMEOUT bash -c "</dev/tcp/localhost/$port" 2>/dev/null; thenif [ "$port" = "9092" ]; thenFAULT_LEVEL="P0"FAULT_ISSUES+=("Kafka主端口$port不可访问")echo "🔴 P0 致命: Kafka主端口$port不可访问"elseif [ "$FAULT_LEVEL" != "P0" ]; thenFAULT_LEVEL="P1"fiFAULT_ISSUES+=("Zookeeper端口$port不可访问")echo "🟠 P1 重大: Zookeeper端口$port不可访问"fielseecho "✅ 端口$port可访问"fi
done# 检查Broker状态
if [ "$FAULT_LEVEL" != "P0" ]; thenBROKER_COUNT=$(timeout $QUICK_CHECK_TIMEOUT kafka-broker-api-versions.sh --bootstrap-server localhost:9092 2>/dev/null | wc -l)if [ $BROKER_COUNT -eq 0 ]; thenFAULT_LEVEL="P0"FAULT_ISSUES+=("无法连接到任何Broker")echo "🔴 P0 致命: 无法连接到任何Broker"elseecho "✅ 检测到 $BROKER_COUNT 个Broker节点"fi
fi# 检查磁盘空间
if [ -d "/kafka-logs" ]; thenDISK_USAGE=$(df /kafka-logs | tail -1 | awk '{print $5}' | cut -d'%' -f1)if [ $DISK_USAGE -gt 95 ]; thenif [ "$FAULT_LEVEL" != "P0" ]; thenFAULT_LEVEL="P1"fiFAULT_ISSUES+=("Kafka数据目录磁盘使用率$DISK_USAGE%")echo "🟠 P1 重大: 磁盘空间严重不足 ($DISK_USAGE%)"elif [ $DISK_USAGE -gt 85 ]; thenif [ "$FAULT_LEVEL" = "P3" ]; thenFAULT_LEVEL="P2"fiFAULT_ISSUES+=("Kafka数据目录磁盘使用率$DISK_USAGE%")echo "🟡 P2 一般: 磁盘空间不足 ($DISK_USAGE%)"elseecho "✅ 磁盘空间充足 ($DISK_USAGE%)"fi
fi# 显示故障等级和建议
echo -e "\n2. 故障等级评估结果"
echo "--------------------------------------------------------------------------------"
echo "故障等级: $FAULT_LEVEL"
echo "问题数量: ${#FAULT_ISSUES[@]}"if [ ${#FAULT_ISSUES[@]} -gt 0 ]; thenecho "发现的问题:"for issue in "${FAULT_ISSUES[@]}"; doecho " • $issue"done
fi# 根据故障等级提供建议
echo -e "\n3. 紧急处理建议"
echo "--------------------------------------------------------------------------------"case $FAULT_LEVEL in"P0")echo "🆘 致命故障 - 立即处理:"echo " 1. 立即通知相关人员和上级领导"echo " 2. 检查服务器硬件和网络状态"echo " 3. 尝试重启 Kafka 服务"echo " 4. 如果重启失败，考虑切换到备用集群"echo " 5. 记录所有操作和日志信息";;"P1")echo "🟠 重大故障 - 一小时内处理:"echo " 1. 通知相关技术人员"echo " 2. 分析故障原因和影响范围"echo " 3. 制定并执行修复方案"echo " 4. 监控系统恢复情况";;"P2")echo "🟡 一般故障 - 一天内处理:"echo " 1. 评估对业务的影响"echo " 2. 安排合适的维护窗口"echo " 3. 制定详细的解决方案";;"P3")echo "✅ 轻微故障 - 正常维护时间处理:"echo " 1. 正常排期进行检查和修复"echo " 2. 记录问题和解决方案";;
esacecho -e "\n4. 快速恢复操作指令"
echo "--------------------------------------------------------------------------------"# 根据具体问题提供快速恢复指令
if [[ " ${FAULT_ISSUES[@]} " =~ "Kafka进程未运行" ]]; thenecho "Kafka服务重启指令:"echo " sudo systemctl start kafka"echo " # 或者"echo " cd /kafka && bin/kafka-server-start.sh config/server.properties &"
fiif [[ " ${FAULT_ISSUES[@]} " =~ "磁盘" ]]; thenecho "\n磁盘空间清理指令:"echo " # 清理过期日志段"echo " find /kafka-logs -name '*.log' -mtime +7 -delete"echo " # 压缩旧日志"echo " find /kafka-logs -name '*.log' -mtime +3 -exec gzip {} \;"
fiecho -e "\n⚠️ 注意: 执行任何恢复操作前，请确保备份重要数据和配置文件！"
echo "================================================================================="
第二部分：自动恢复脚本
#!/bin/bash
# Kafka自动恢复脚本echo "=== Kafka自动恢复脚本 ==="# 自动恢复参数
RECOVERY_TIMEOUT=300 # 恢复超时时间（秒）
MAX_RETRY_ATTEMPTS=3 # 最大重试次数
HEALTH_CHECK_INTERVAL=10 # 健康检查间隔（秒）
BACKUP_DIR="/backup/kafka-$(date +%Y%m%d_%H%M%S)"echo "自动恢复参数:"
echo " 恢复超时: $RECOVERY_TIMEOUT 秒"
echo " 最大重试: $MAX_RETRY_ATTEMPTS 次"
echo " 健康检查间隔: $HEALTH_CHECK_INTERVAL 秒"
echo " 备份目录: $BACKUP_DIR"# 创建备份目录
mkdir -p "$BACKUP_DIR"# 恢复函数定义
log_message() {echo "[$(date '+%Y-%m-%d %H:%M:%S')] $1" | tee -a "$BACKUP_DIR/recovery.log"
}check_kafka_health() {local max_wait=${1:-30}local start_time=$(date +%s)while [ $(($(date +%s) - start_time)) -lt $max_wait ]; doif pgrep -f kafka.Kafka >/dev/null; thenif timeout 5 kafka-topics.sh --bootstrap-server localhost:9092 --list >/dev/null 2>&1; thenreturn 0fifisleep 2donereturn 1
}backup_critical_data() {log_message "开始备份关键数据..."# 备份配置文件if [ -f "/kafka/config/server.properties" ]; thencp /kafka/config/server.properties "$BACKUP_DIR/"fi# 备份Topic配置if check_kafka_health 10; thenkafka-topics.sh --bootstrap-server localhost:9092 --describe > "$BACKUP_DIR/topics_backup.txt" 2>/dev/nullfi# 备份消费者组信息if check_kafka_health 10; thenkafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --all-groups > "$BACKUP_DIR/consumer_groups_backup.txt" 2>/dev/nullfilog_message "关键数据备份完成"
}restart_kafka_service() {local attempt=$1log_message "开始第 $attempt 次Kafka服务重启尝试..."# 停止现有进程if pgrep -f kafka.Kafka >/dev/null; thenlog_message "停止现有Kafka进程..."pkill -f kafka.Kafkasleep 10# 强制杀死残留进程if pgrep -f kafka.Kafka >/dev/null; thenpkill -9 -f kafka.Kafkasleep 5fifi# 检查必要的目录和文件if [ ! -d "/kafka-logs" ]; thenlog_message "创建缺失的日志目录..."mkdir -p /kafka-logschown kafka:kafka /kafka-logs 2>/dev/null || truefi# 检查配置文件if [ ! -f "/kafka/config/server.properties" ]; thenlog_message "错误: Kafka配置文件不存在"return 1fi# 启动Kafka服务log_message "启动Kafka服务..."# 尝试使用systemctlif command -v systemctl >/dev/null 2>&1; thensystemctl start kafka && sleep 15else# 手动启动cd /kafkanohup bin/kafka-server-start.sh config/server.properties > /kafka-logs/kafka.out 2>&1 &sleep 15fi# 检查启动结果if check_kafka_health 60; thenlog_message "Kafka服务启动成功"return 0elselog_message "Kafka服务启动失败"return 1fi
}cleanup_disk_space() {log_message "开始清理磁盘空间..."if [ -d "/kafka-logs" ]; then# 清理过期日志段（7天前）find /kafka-logs -name "*.log" -mtime +7 -delete 2>/dev/null# 压缩旧日志（3天前）find /kafka-logs -name "*.log" -mtime +3 -exec gzip {} \; 2>/dev/null# 清理临时文件find /kafka-logs -name "*.tmp" -delete 2>/dev/nullfind /kafka-logs -name "*.swap" -delete 2>/dev/nullDISK_USAGE_AFTER=$(df /kafka-logs | tail -1 | awk '{print $5}' | cut -d'%' -f1)log_message "磁盘清理完成，当前使用率: ${DISK_USAGE_AFTER}%"fi
}fix_permissions() {log_message "修复文件权限..."if [ -d "/kafka-logs" ]; thenchown -R kafka:kafka /kafka-logs 2>/dev/null || \chown -R $(whoami):$(whoami) /kafka-logs 2>/dev/null || truechmod -R 755 /kafka-logs 2>/dev/null || truefiif [ -d "/kafka/config" ]; thenchmod 644 /kafka/config/*.properties 2>/dev/null || truefilog_message "文件权限修复完成"
}# 主恢复流程
log_message "=== 开始 Kafka 自动恢复流程 ==="# 步骤1: 备份关键数据
backup_critical_data# 步骤2: 检查磁盘空间
if [ -d "/kafka-logs" ]; thenDISK_USAGE=$(df /kafka-logs | tail -1 | awk '{print $5}' | cut -d'%' -f1)if [ $DISK_USAGE -gt 90 ]; thencleanup_disk_spacefi
fi# 步骤3: 修复文件权限
fix_permissions# 步骤4: 尝试重启Kafka服务
RECOVERY_SUCCESS=falsefor attempt in $(seq 1 $MAX_RETRY_ATTEMPTS); doif restart_kafka_service $attempt; thenRECOVERY_SUCCESS=truebreakfiif [ $attempt -lt $MAX_RETRY_ATTEMPTS ]; thenlog_message "等待 30 秒后重试..."sleep 30fi
done# 步骤5: 恢复结果验证
if [ "$RECOVERY_SUCCESS" = true ]; thenlog_message "✅ Kafka自动恢复成功！"# 进行全面健康检查log_message "进行全面健康检查..."# 检查Topic状态if kafka-topics.sh --bootstrap-server localhost:9092 --list >/dev/null 2>&1; thenTOPIC_COUNT=$(kafka-topics.sh --bootstrap-server localhost:9092 --list 2>/dev/null | wc -l)log_message "检测到 $TOPIC_COUNT 个Topic"fi# 检查消费者组状态if kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list >/dev/null 2>&1; thenGROUP_COUNT=$(kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list 2>/dev/null | wc -l)log_message "检测到 $GROUP_COUNT 个消费者组"filog_message "健康检查完成，系统恢复正常"elselog_message "❌ Kafka自动恢复失败！"log_message "请手动检查以下项目:"log_message "1. 服务器硬件状态"log_message "2. 操作系统资源"log_message "3. Kafka配置文件"log_message "4. 网络连接状态"log_message "5. Zookeeper服务状态"# 生成诊断报告log_message "正在生成详细诊断报告..."{echo "=== Kafka恢复失败诊断报告 ==="echo "生成时间: $(date)"echo ""echo "1. 系统资源状态:"echo "CPU: $(top -bn1 | grep 'Cpu(s)' | awk '{print $2}')"echo "内存: $(free -h | grep Mem | awk '{printf "%s/%s (%.1f%%)", $3, $2, $3/$2*100}')"echo "磁盘: $(df -h /kafka-logs 2>/dev/null | tail -1 | awk '{print $5}' || echo 'N/A')"echo ""echo "2. 进程状态:"ps aux | grep -E "(kafka|zookeeper)" | grep -v grepecho ""echo "3. 网络状态:"netstat -tlnp | grep -E ":(9092|2181) "echo ""echo "4. 最近的错误日志:"tail -20 /kafka-logs/server.log 2>/dev/null || echo "日志文件不存在"} > "$BACKUP_DIR/diagnostic_report.txt"log_message "诊断报告已保存到: $BACKUP_DIR/diagnostic_report.txt"
filog_message "=== Kafka 自动恢复流程结束 ==="
log_message "所有日志和备份文件保存在: $BACKUP_DIR"
⚙️ 九、生产环境配置优化
📖 生产环境配置最佳实践
Kafka在生产环境中的正确配置对于系统的性能、稳定性和可靠性至关重要。以下配置模板基于多年的生产环境实践经验总结而成。
🔧 生产环境配置生成器
#!/bin/bash
# Kafka生产环境配置生成器echo "=== Kafka生产环境配置生成器 ==="# 配置参数收集
read -p "输入Broker ID [0]: " BROKER_ID
BROKER_ID=${BROKER_ID:-0}read -p "输入监听地址 [localhost]: " LISTENERS_HOST
LISTENERS_HOST=${LISTENERS_HOST:-localhost}read -p "输入监听端口 [9092]: " LISTENERS_PORT
LISTENERS_PORT=${LISTENERS_PORT:-9092}read -p "输入Zookeeper地址 [localhost:2181]: " ZOOKEEPER_CONNECT
ZOOKEEPER_CONNECT=${ZOOKEEPER_CONNECT:-localhost:2181}read -p "输入日志目录 [/kafka-logs]: " LOG_DIRS
LOG_DIRS=${LOG_DIRS:-/kafka-logs}read -p "输入默认分区数 [12]: " NUM_PARTITIONS
NUM_PARTITIONS=${NUM_PARTITIONS:-12}read -p "输入默认副本因子 [3]: " DEFAULT_REPLICATION_FACTOR
DEFAULT_REPLICATION_FACTOR=${DEFAULT_REPLICATION_FACTOR:-3}read -p "选择环境类型 [1-高吞吐量, 2-低延迟, 3-平衡]: " ENV_TYPE
ENV_TYPE=${ENV_TYPE:-3}# 生成配置文件
CONFIG_FILE="server-production-broker${BROKER_ID}.properties"echo "正在生成配置文件: $CONFIG_FILE"cat > $CONFIG_FILE << EOF
# Kafka生产环境配置文件
# 生成时间: $(date)
# Broker ID: $BROKER_ID
# 环境类型: $([[ $ENV_TYPE == 1 ]] && echo "高吞吐量" || [[ $ENV_TYPE == 2 ]] && echo "低延迟" || echo "平衡")############################# Server Basics ############################## Broker的唯一标识符
broker.id=$BROKER_ID# 删除topic功能使能
delete.topic.enable=true# 不允许自动创建topic
auto.create.topics.enable=false############################# Socket Server Settings ############################## 监听地址配置
listeners=PLAINTEXT://$LISTENERS_HOST:$LISTENERS_PORT# 对外公布的地址
advertised.listeners=PLAINTEXT://$LISTENERS_HOST:$LISTENERS_PORT# 网络请求处理的线程数
num.network.threads=8# 磁盘IO线程数
num.io.threads=16# Socket发送缓冲区大小
socket.send.buffer.bytes=102400# Socket接收缓冲区大小
socket.receive.buffer.bytes=102400# Socket请求的最大字节数
socket.request.max.bytes=104857600############################# Log Basics ############################## 日志存储目录
log.dirs=$LOG_DIRS# 默认分区数
num.partitions=$NUM_PARTITIONS# 默认副本因子
default.replication.factor=$DEFAULT_REPLICATION_FACTOR# 最小同步副本数
min.insync.replicas=2# 禁用unclean leader选举
unclean.leader.election.enable=false############################# Internal Topic Settings ############################## 内部topic的副本因子
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2############################# Log Flush Policy #############################EOF# 根据环境类型添加特定配置
case $ENV_TYPE in1) # 高吞吐量配置cat >> $CONFIG_FILE << 'EOF'
# 高吞吐量环境配置
# 刷盘策略：性能优先
log.flush.interval.messages=10000
log.flush.interval.ms=3000# 批处理优化
replica.fetch.max.bytes=1048576
message.max.bytes=1000000# 网络优化
replica.socket.timeout.ms=30000
replica.socket.receive.buffer.bytes=65536EOF;;2) # 低延迟配置cat >> $CONFIG_FILE << 'EOF'
# 低延迟环境配置
# 刷盘策略：立即刷盘
log.flush.interval.messages=1
log.flush.interval.ms=100# 减少批处理大小
replica.fetch.max.bytes=65536
message.max.bytes=100000# 网络优化
replica.socket.timeout.ms=3000
replica.socket.receive.buffer.bytes=32768EOF;;3) # 平衡配置（默认）cat >> $CONFIG_FILE << 'EOF'
# 平衡性能配置
# 适中的刷盘策略
log.flush.interval.messages=5000
log.flush.interval.ms=1000# 适中的批处理大小
replica.fetch.max.bytes=524288
message.max.bytes=500000# 网络优化
replica.socket.timeout.ms=10000
replica.socket.receive.buffer.bytes=32768EOF;;
esac# 添加通用的生产环境配置
cat >> $CONFIG_FILE << EOF############################# Log Retention Policy ############################## 日志保留时间（7天）
log.retention.hours=168# 日志保留大小（1GB）
log.retention.bytes=1073741824# 日志段文件的最大大小（1GB）
log.segment.bytes=1073741824# 检查日志保留的间隔时间
log.retention.check.interval.ms=300000# 日志清理策略
log.cleanup.policy=delete# 启用日志压缩清理器
log.cleaner.enable=true############################# Zookeeper ############################## Zookeeper连接字符串
zookeeper.connect=$ZOOKEEPER_CONNECT# Zookeeper连接超时时间
zookeeper.connection.timeout.ms=18000# Zookeeper会话超时时间
zookeeper.session.timeout.ms=18000############################# Group Coordinator Settings ############################## 消费者组初始重平衡延迟
group.initial.rebalance.delay.ms=3000# 消费者组会话超时时间
group.max.session.timeout.ms=1800000# 消费者组最小会话超时时间
group.min.session.timeout.ms=6000############################# Producer/Consumer Defaults ############################## 生产者默认批次大小
batch.size=16384# 生产者缓冲区大小
buffer.memory=33554432# 消费者获取大小
fetch.min.bytes=1# 消费者等待时间
fetch.max.wait.ms=500############################# Metrics Settings ############################## JMX端口设置
# 在启动脚本中设置: export JMX_PORT=9999# 指标报告间隔
metric.reporters=
metrics.num.samples=2
metrics.recording.level=INFO
metrics.sample.window.ms=30000############################# Security Settings ############################## SSL配置（如果需要）
# ssl.keystore.location=/path/to/kafka.server.keystore.jks
# ssl.keystore.password=test1234
# ssl.key.password=test1234
# ssl.truststore.location=/path/to/kafka.server.truststore.jks
# ssl.truststore.password=test1234# SASL配置（如果需要）
# sasl.enabled.mechanisms=PLAIN
# sasl.mechanism.inter.broker.protocol=PLAIN# ACL配置（如果需要）
# authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
# super.users=User:admin############################# Compression Settings ############################## 默认压缩类型
compression.type=producer# 启用压缩
log.compression.type=producer############################# Replication Settings ############################## 副本同步超时
replica.lag.time.max.ms=30000# 副本获取超时
replica.socket.timeout.ms=30000# Leader选举超时
controller.socket.timeout.ms=30000# 副本高水位检查点间隔
replica.high.watermark.checkpoint.interval.ms=5000# 副本获取响应最大大小
replica.fetch.response.max.bytes=10485760############################# Quota Settings ############################## 生产者配额（字节/秒）
# quota.producer.default=1048576# 消费者配额（字节/秒）
# quota.consumer.default=2097152# 请求配额（请求/秒）
# quota.window.num=11
# quota.window.size.seconds=1EOFecho "✅ 配置文件生成完成: $CONFIG_FILE"# 生成JVM配置
JVM_CONFIG_FILE="kafka-jvm-production.sh"cat > $JVM_CONFIG_FILE << 'EOF'
#!/bin/bash
# Kafka生产环境JVM配置脚本# 基础内存配置（根据实际服务器内存调整）
export KAFKA_HEAP_OPTS="-Xmx6G -Xms6G"# G1垃圾收集器配置
export KAFKA_JVM_PERFORMANCE_OPTS="-server \-XX:+UseG1GC \-XX:MaxGCPauseMillis=20 \-XX:InitiatingHeapOccupancyPercent=35 \-XX:+ExplicitGCInvokesConcurrent \-XX:MaxInlineLevel=15 \-XX:+UseCompressedOops \-Djava.awt.headless=true"# GC日志配置
export KAFKA_GC_LOG_OPTS="-XX:+UseGCLogFileRotation \-XX:NumberOfGCLogFiles=10 \-XX:GCLogFileSize=100M \-XX:+PrintGC \-XX:+PrintGCDetails \-XX:+PrintGCTimeStamps \-XX:+PrintGCApplicationStoppedTime \-Xloggc:/kafka-logs/kafkaServer-gc.log"# JMX配置
export KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote \-Dcom.sun.management.jmxremote.authenticate=false \-Dcom.sun.management.jmxremote.ssl=false \-Dcom.sun.management.jmxremote.port=9999"# 其他JVM优化参数
export KAFKA_OPTS="-Djava.security.auth.login.config=/kafka/config/kafka_server_jaas.conf"echo "Kafka JVM配置已加载"
echo "堆内存: $KAFKA_HEAP_OPTS"
echo "JMX端口: 9999"
EOFchmod +x $JVM_CONFIG_FILE
echo "✅ JVM配置脚本生成完成: $JVM_CONFIG_FILE"# 生成启动脚本
STARTUP_SCRIPT="kafka-production-start.sh"cat > $STARTUP_SCRIPT << EOF
#!/bin/bash
# Kafka生产环境启动脚本KAFKA_HOME="/kafka"
CONFIG_FILE="$PWD/$CONFIG_FILE"
JVM_CONFIG="$PWD/$JVM_CONFIG_FILE"echo "=== Kafka生产环境启动脚本 ==="
echo "Kafka主目录: \$KAFKA_HOME"
echo "配置文件: \$CONFIG_FILE"
echo "JVM配置: \$JVM_CONFIG"# 检查必要文件
if [ ! -f "\$CONFIG_FILE" ]; thenecho "❌ 配置文件不存在: \$CONFIG_FILE"exit 1
fiif [ ! -f "\$JVM_CONFIG" ]; thenecho "❌ JVM配置文件不存在: \$JVM_CONFIG"exit 1
fi# 创建必要目录
mkdir -p $LOG_DIRS
mkdir -p /kafka-logs# 加载JVM配置
source \$JVM_CONFIG# 检查端口是否被占用
if netstat -tlnp 2>/dev/null | grep -q ":$LISTENERS_PORT "; thenecho "⚠️ 端口 $LISTENERS_PORT 已被占用，请检查"exit 1
fi# 启动Kafka
echo "🚀 启动Kafka服务..."
cd \$KAFKA_HOME# 后台启动
nohup bin/kafka-server-start.sh \$CONFIG_FILE > /kafka-logs/kafka-startup.log 2>&1 &KAFKA_PID=\$!
echo "Kafka进程ID: \$KAFKA_PID"# 等待启动
echo "等待Kafka服务启动..."
sleep 15# 检查启动状态
if ps -p \$KAFKA_PID > /dev/null; thenecho "✅ Kafka服务启动成功"echo "进程ID: \$KAFKA_PID"echo "监听地址: $LISTENERS_HOST:$LISTENERS_PORT"echo "日志目录: $LOG_DIRS"# 简单健康检查sleep 10if timeout 10 bin/kafka-topics.sh --bootstrap-server $LISTENERS_HOST:$LISTENERS_PORT --list >/dev/null 2>&1; thenecho "✅ Kafka服务健康检查通过"elseecho "⚠️ Kafka服务可能未完全启动，请查看日志"fi
elseecho "❌ Kafka服务启动失败"echo "请查看启动日志: /kafka-logs/kafka-startup.log"exit 1
fi
EOFchmod +x $STARTUP_SCRIPT
echo "✅ 启动脚本生成完成: $STARTUP_SCRIPT"# 生成系统优化建议
SYSTEM_TUNING_FILE="system-tuning-recommendations.md"cat > $SYSTEM_TUNING_FILE << 'EOF'
# Kafka生产环境系统优化建议## 操作系统内核参数优化在 `/etc/sysctl.conf` 中添加以下配置：```bash
# 虚拟内存设置
vm.swappiness=1
vm.dirty_background_ratio=5
vm.dirty_ratio=60
vm.dirty_expire_centisecs=12000
vm.max_map_count=262144# 网络参数优化
net.core.wmem_default=131072
net.core.rmem_default=131072
net.core.wmem_max=2097152
net.core.rmem_max=2097152
net.ipv4.tcp_wmem=4096 12582912 16777216
net.ipv4.tcp_rmem=4096 12582912 16777216
net.core.netdev_max_backlog=5000
net.ipv4.tcp_max_syn_backlog=8096
net.core.somaxconn=32768# 文件系统优化
fs.file-max=2097152
应用配置：
sysctl -p
用户限制配置
在 /etc/security/limits.conf 中添加：
kafka soft nofile 100000
kafka hard nofile 100000
kafka soft nproc 32768
kafka hard nproc 32768
kafka soft memlock unlimited
kafka hard memlock unlimited
文件系统优化
XFS文件系统（推荐）
# 挂载选项
/dev/sdb1 /kafka-logs xfs defaults,noatime,largeio,inode64 0 0
EXT4文件系统
# 挂载选项
/dev/sdb1 /kafka-logs ext4 defaults,noatime 0 0
Java环境优化
时区设置
export TZ=Asia/Shanghai
文件编码
export LANG=zh_CN.UTF-8
监控和日志配置
日志轮转配置 (/etc/logrotate.d/kafka)
/kafka-logs/*.log {dailymissingokrotate 7compressdelaycompresscopytruncatenotifemptycreate 644 kafka kafka
}
Crontab定时任务
# 每小时检查磁盘使用情况
0 * * * * /usr/local/bin/kafka-disk-check.sh# 每天凌晨2点进行日志清理
0 2 * * * /usr/local/bin/kafka-log-cleanup.sh# 每5分钟检查Kafka服务状态
*/5 * * * * /usr/local/bin/kafka-health-check.sh
安全配置建议
防火墙配置
# 开放必要端口
firewall-cmd --permanent --add-port=9092/tcp
firewall-cmd --permanent --add-port=9093/tcp
firewall-cmd --permanent --add-port=2181/tcp
firewall-cmd --permanent --add-port=9999/tcp
firewall-cmd --reload
用户权限配置
# 创建专用用户
useradd -r -s /bin/false kafka
chown -R kafka:kafka /kafka
chown -R kafka:kafka /kafka-logs
硬件建议
CPU
- 推荐：16核心或更多
- 最少：8核心
内存
- 推荐：64GB或更多
- 最少：32GB
- JVM堆内存建议为物理内存的25-50%
存储
- 推荐：SSD存储，RAID10配置
- 最少：7200转机械硬盘，RAID1配置
- 独立的日志存储磁盘
网络
- 推荐：万兆网卡
- 最少：千兆网卡
- 低延迟网络交换机
容量规划
分区数规划
- 单个Topic分区数 = 目标吞吐量 / 单分区最大吞吐量
- 集群总分区数建议不超过4000-6000个
副本数规划
- 生产环境建议副本因子为3
- 重要数据可以设置为5
存储容量规划
- 预留30%的存储空间用于数据增长
- 考虑压缩比例（通常可以达到2:1-4:1）
EOF
echo “✅ 系统优化建议文档生成完成: $SYSTEM_TUNING_FILE”
echo -e “\n=== 配置生成总结 ===”
echo “1. Kafka配置文件: $CONFIG_FILE”
echo “2. JVM配置脚本: $JVM_CONFIG_FILE”
echo “3. 启动脚本: $STARTUP_SCRIPT”
echo “4. 系统优化建议: $SYSTEM_TUNING_FILE”
echo -e “\n🚀 下一步操作：”
echo “1. 检查生成的配置文件”
echo “2. 根据实际环境调整参数”
echo “3. 执行系统优化建议”
echo “4. 使用启动脚本启动Kafka”
echo “5. 进行功能和性能验证”
---## 🎯 十、最佳实践和预防措施### 📖 Kafka运维最佳实践总结基于多年的Kafka生产环境运维经验，以下最佳实践可以有效预防常见问题，提高系统的稳定性和可维护性。### 🛡️ 预防性维护策略#### 第一部分：监控和告警体系```bash
#!/bin/bash
# Kafka预防性监控策略实施脚本echo "=== Kafka预防性监控策略实施 ==="# 创建监控脚本目录
MONITOR_DIR="/usr/local/kafka-monitor"
mkdir -p $MONITOR_DIRecho "1. 部署核心监控指标收集"
echo "================================"# 核心指标监控脚本
cat > $MONITOR_DIR/kafka-metrics-collector.sh << 'EOF'
#!/bin/bash
# Kafka核心指标收集脚本METRICS_FILE="/var/log/kafka-metrics-$(date +%Y%m%d).json"
TIMESTAMP=$(date '+%Y-%m-%d %H:%M:%S')collect_metrics() {# Broker基础指标BROKER_COUNT=$(kafka-broker-api-versions.sh --bootstrap-server localhost:9092 2>/dev/null | wc -l)# Topic和分区指标TOPIC_COUNT=$(kafka-topics.sh --bootstrap-server localhost:9092 --list 2>/dev/null | wc -l)PARTITION_COUNT=$(kafka-topics.sh --bootstrap-server localhost:9092 --describe 2>/dev/null | grep "PartitionCount" | awk '{sum += $2} END {print sum+0}')# 消费者组指标CONSUMER_GROUPS=$(kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list 2>/dev/null | wc -l)TOTAL_LAG=$(kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --all-groups 2>/dev/null | awk 'NR>1 && $5 != "-" {sum += $5} END {print sum+0}')# 系统资源指标CPU_USAGE=$(top -bn1 | grep "Cpu(s)" | awk '{print $2}' | cut -d'%' -f1)MEMORY_USAGE=$(free | grep Mem | awk '{printf "%.1f", $3/$2 * 100.0}')DISK_USAGE=$(df /kafka-logs 2>/dev/null | tail -1 | awk '{print $5}' | cut -d'%' -f1 || echo "0")# 网络连接数KAFKA_CONNECTIONS=$(netstat -an 2>/dev/null | grep ":9092 " | grep ESTABLISHED | wc -l)# 构建JSON格式指标cat << JSON >> $METRICS_FILE
{"timestamp": "$TIMESTAMP","broker_count": $BROKER_COUNT,"topic_count": $TOPIC_COUNT,"partition_count": $PARTITION_COUNT,"consumer_groups": $CONSUMER_GROUPS,"total_lag": $TOTAL_LAG,"cpu_usage": $CPU_USAGE,"memory_usage": $MEMORY_USAGE,"disk_usage": $DISK_USAGE,"kafka_connections": $KAFKA_CONNECTIONS
},
JSON
}# 执行指标收集
collect_metrics# 告警检查
check_alerts() {# 高延迟告警if [ $TOTAL_LAG -gt 100000 ]; thenecho "ALERT: High message lag detected: $TOTAL_LAG" | logger -t kafka-monitorfi# 磁盘空间告警if [ $DISK_USAGE -gt 85 ]; thenecho "ALERT: High disk usage: ${DISK_USAGE}%" | logger -t kafka-monitorfi# CPU使用率告警if awk "BEGIN {exit !($CPU_USAGE > 80)}"; thenecho "ALERT: High CPU usage: ${CPU_USAGE}%" | logger -t kafka-monitorfi# 内存使用率告警if awk "BEGIN {exit !($MEMORY_USAGE > 85)}"; thenecho "ALERT: High memory usage: ${MEMORY_USAGE}%" | logger -t kafka-monitorfi
}check_alerts
EOFchmod +x $MONITOR_DIR/kafka-metrics-collector.shecho "2. 设置预防性检查任务"
echo "================================"# 预防性健康检查脚本
cat > $MONITOR_DIR/kafka-preventive-checks.sh << 'EOF'
#!/bin/bash
# Kafka预防性健康检查脚本REPORT_FILE="/var/log/kafka-health-report-$(date +%Y%m%d_%H%M).txt"echo "=== Kafka预防性健康检查报告 - $(date) ===" > $REPORT_FILE# 1. 检查磁盘空间趋势
echo "1. 磁盘空间趋势分析" >> $REPORT_FILE
echo "================================" >> $REPORT_FILEif [ -d "/kafka-logs" ]; thenCURRENT_USAGE=$(df /kafka-logs | tail -1 | awk '{print $5}' | cut -d'%' -f1)AVAILABLE_SPACE=$(df -h /kafka-logs | tail -1 | awk '{print $4}')echo "当前磁盘使用率: ${CURRENT_USAGE}%" >> $REPORT_FILEecho "剩余可用空间: $AVAILABLE_SPACE" >> $REPORT_FILE# 预测磁盘满的时间（基于过去7天的增长）if [ -f "/var/log/disk-usage-history.log" ]; then# 简单的线性预测（实际应用中可以使用更复杂的算法）GROWTH_RATE=$(tail -7 /var/log/disk-usage-history.log | awk '{sum+=$1} END {print sum/7}' 2>/dev/null || echo "0")if [ "$GROWTH_RATE" != "0" ] && awk "BEGIN {exit !($GROWTH_RATE > 0)}"; thenDAYS_TO_FULL=$(awk "BEGIN {printf \"%.0f\", (100-$CURRENT_USAGE)/$GROWTH_RATE}")echo "预计磁盘满的时间: ${DAYS_TO_FULL}天后（基于平均增长率${GROWTH_RATE}%/天）" >> $REPORT_FILEif [ $DAYS_TO_FULL -lt 30 ]; thenecho "⚠️ 警告: 磁盘可能在30天内用满，请及时清理或扩容" >> $REPORT_FILEfififi# 记录当前使用率到历史文件echo "$CURRENT_USAGE" >> /var/log/disk-usage-history.log
fi# 2. 检查分区分布均衡性
echo -e "\n2. 分区分布均衡性检查" >> $REPORT_FILE
echo "================================" >> $REPORT_FILEPARTITION_BALANCE=$(kafka-topics.sh --bootstrap-server localhost:9092 --describe 2>/dev/null | \grep -E "Leader: [0-9]+" | awk '{print $6}' | sort | uniq -c | \awk '{leader_count[$2] = $1total += $1count++}END {if (count > 0) {avg = total / countmax_dev = 0for (broker in leader_count) {dev = (leader_count[broker] - avg) / avg * 100if (dev < 0) dev = -devif (dev > max_dev) max_dev = dev}printf "最大偏差: %.1f%%", max_dev}}')echo "分区Leader分布均衡度: $PARTITION_BALANCE" >> $REPORT_FILE# 3. 检查消费者组健康状态
echo -e "\n3. 消费者组健康状态检查" >> $REPORT_FILE
echo "================================" >> $REPORT_FILECONSUMER_GROUPS=$(kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list 2>/dev/null)
UNHEALTHY_GROUPS=0for group in $CONSUMER_GROUPS; doGROUP_LAG=$(kafka-consumer-groups.sh --bootstrap-server localhost:9092 \--describe --group $group 2>/dev/null | \awk 'NR>1 && $5 != "-" {sum += $5} END {print sum+0}')INACTIVE_PARTITIONS=$(kafka-consumer-groups.sh --bootstrap-server localhost:9092 \--describe --group $group 2>/dev/null | \awk 'NR>1 && $7 == "-"' | wc -l)if [ $GROUP_LAG -gt 50000 ] || [ $INACTIVE_PARTITIONS -gt 0 ]; thenecho "⚠️ 消费者组 $group: 堆积 $GROUP_LAG, 无活跃消费者分区 $INACTIVE_PARTITIONS" >> $REPORT_FILE((UNHEALTHY_GROUPS++))fi
doneecho "异常消费者组数量: $UNHEALTHY_GROUPS" >> $REPORT_FILE# 4. 检查集群配置一致性
echo -e "\n4. 集群配置一致性检查" >> $REPORT_FILE
echo "================================" >> $REPORT_FILE# 检查关键配置参数是否一致（这里简化处理）
echo "检查各Broker关键配置参数一致性..." >> $REPORT_FILE
echo "注意: 需要在所有Broker节点上运行此检查" >> $REPORT_FILE# 5. 性能趋势分析
echo -e "\n5. 性能趋势分析" >> $REPORT_FILE
echo "================================" >> $REPORT_FILE# 分析最近的性能指标趋势（需要历史数据支持）
if [ -f "/var/log/kafka-metrics-$(date +%Y%m%d).json" ]; thenecho "基于今日指标数据的趋势分析:" >> $REPORT_FILE# 简单的趋势分析（实际应用中需要更复杂的分析）AVG_CPU=$(grep '"cpu_usage"' "/var/log/kafka-metrics-$(date +%Y%m%d).json" | \awk -F': ' '{gsub(/,/, "", $2); sum+=$2; count++} END {if(count>0) printf "%.1f", sum/count}')AVG_MEMORY=$(grep '"memory_usage"' "/var/log/kafka-metrics-$(date +%Y%m%d).json" | \awk -F': ' '{gsub(/,/, "", $2); sum+=$2; count++} END {if(count>0) printf "%.1f", sum/count}')echo "今日平均CPU使用率: ${AVG_CPU}%" >> $REPORT_FILEecho "今日平均内存使用率: ${AVG_MEMORY}%" >> $REPORT_FILE
fi# 6. 生成改进建议
echo -e "\n6. 改进建议" >> $REPORT_FILE
echo "================================" >> $REPORT_FILEif [ $CURRENT_USAGE -gt 70 ]; thenecho "• 考虑增加磁盘容量或实施数据清理策略" >> $REPORT_FILE
fiif [ $UNHEALTHY_GROUPS -gt 0 ]; thenecho "• 检查并修复异常的消费者组" >> $REPORT_FILE
fiecho "• 定期检查和优化Topic分区数量" >> $REPORT_FILE
echo "• 监控并调整JVM参数以优化GC性能" >> $REPORT_FILE
echo "• 定期更新Kafka版本以获得性能改进和安全修复" >> $REPORT_FILEecho "报告生成完成: $REPORT_FILE"
EOFchmod +x $MONITOR_DIR/kafka-preventive-checks.shecho "3. 创建定时任务配置"
echo "================================"# 生成crontab配置建议
cat > $MONITOR_DIR/kafka-crontab-template.txt << 'EOF'
# Kafka监控和维护定时任务配置
# 请根据实际需求调整时间设置# 每5分钟收集一次核心指标
*/5 * * * * /usr/local/kafka-monitor/kafka-metrics-collector.sh# 每小时进行一次预防性健康检查
0 * * * * /usr/local/kafka-monitor/kafka-preventive-checks.sh# 每天凌晨2点进行日志清理
0 2 * * * find /kafka-logs -name "*.log" -mtime +7 -delete# 每周日凌晨3点进行数据压缩清理
0 3 * * 0 /usr/local/kafka-monitor/kafka-weekly-cleanup.sh# 每月1号生成月度报告
0 0 1 * * /usr/local/kafka-monitor/kafka-monthly-report.sh
EOFecho "定时任务模板已生成: $MONITOR_DIR/kafka-crontab-template.txt"
echo "请使用 'crontab -e' 添加需要的任务"echo -e "\n=== 预防性监控策略部署完成 ==="
echo "监控脚本目录: $MONITOR_DIR"
echo "请确保定期查看生成的报告和日志文件"
第二部分：容量规划和扩容策略
#!/bin/bash
# Kafka容量规划和自动扩容建议脚本echo "=== Kafka容量规划和扩容策略 ==="# 容量分析脚本
cat > /usr/local/kafka-monitor/kafka-capacity-planner.sh << 'EOF'
#!/bin/bash
# Kafka容量规划分析脚本REPORT_FILE="/var/log/kafka-capacity-report-$(date +%Y%m%d).txt"echo "=== Kafka容量规划分析报告 - $(date) ===" > $REPORT_FILE# 1. 当前容量使用分析
echo "1. 当前容量使用情况" >> $REPORT_FILE
echo "================================" >> $REPORT_FILE# 存储容量分析
if [ -d "/kafka-logs" ]; thenTOTAL_SIZE=$(du -sh /kafka-logs | awk '{print $1}')AVAILABLE_SPACE=$(df -h /kafka-logs | tail -1 | awk '{print $4}')USAGE_PERCENT=$(df /kafka-logs | tail -1 | awk '{print $5}')echo "当前数据大小: $TOTAL_SIZE" >> $REPORT_FILEecho "可用空间: $AVAILABLE_SPACE" >> $REPORT_FILEecho "使用率: $USAGE_PERCENT" >> $REPORT_FILE
fi# Topic容量分布分析
echo -e "\nTopic存储分布 (Top 10):" >> $REPORT_FILE
du -sh /kafka-logs/* 2>/dev/null | sort -hr | head -10 >> $REPORT_FILE# 2. 流量分析
echo -e "\n2. 消息流量分析" >> $REPORT_FILE
echo "================================" >> $REPORT_FILE# 分析消息生产速率（需要历史数据）
TOPICS=$(kafka-topics.sh --bootstrap-server localhost:9092 --list 2>/dev/null)
TOTAL_MESSAGES=0for topic in $TOPICS; doMESSAGES=$(kafka-run-class.sh kafka.tools.GetOffsetShell \--broker-list localhost:9092 --topic $topic --time -1 2>/dev/null | \awk -F: '{sum += $2} END {print sum+0}')TOTAL_MESSAGES=$((TOTAL_MESSAGES + MESSAGES))
doneecho "当前总消息数: $TOTAL_MESSAGES" >> $REPORT_FILE# 3. 性能瓶颈预测
echo -e "\n3. 性能瓶颈预测" >> $REPORT_FILE
echo "================================" >> $REPORT_FILE# CPU瓶颈分析
CPU_CORES=$(nproc)
CURRENT_CPU=$(top -bn1 | grep "Cpu(s)" | awk '{print $2}' | cut -d'%' -f1)echo "CPU核心数: $CPU_CORES" >> $REPORT_FILE
echo "当前CPU使用率: ${CURRENT_CPU}%" >> $REPORT_FILEif awk "BEGIN {exit !($CURRENT_CPU > 70)}"; thenecho "⚠️ CPU使用率较高，建议监控CPU瓶颈" >> $REPORT_FILE
fi# 内存瓶颈分析
TOTAL_MEMORY=$(free -h | grep Mem | awk '{print $2}')
USED_MEMORY=$(free -h | grep Mem | awk '{print $3}')
MEMORY_PERCENT=$(free | grep Mem | awk '{printf "%.1f", $3/$2 * 100.0}')echo "总内存: $TOTAL_MEMORY" >> $REPORT_FILE
echo "已用内存: $USED_MEMORY (${MEMORY_PERCENT}%)" >> $REPORT_FILE# 4. 扩容建议
echo -e "\n4. 扩容建议" >> $REPORT_FILE
echo "================================" >> $REPORT_FILE# 存储扩容建议
USAGE_NUM=$(df /kafka-logs | tail -1 | awk '{print $5}' | cut -d'%' -f1)
if [ $USAGE_NUM -gt 70 ]; thenecho "🔴 存储扩容建议: 磁盘使用率 ${USAGE_NUM}%" >> $REPORT_FILEecho " - 短期: 清理过期日志，调整retention设置" >> $REPORT_FILEecho " - 中期: 增加磁盘容量" >> $REPORT_FILEecho " - 长期: 考虑增加Broker节点进行横向扩展" >> $REPORT_FILE
elif [ $USAGE_NUM -gt 50 ]; thenecho "🟡 存储关注: 磁盘使用率 ${USAGE_NUM}%" >> $REPORT_FILEecho " - 开始制定存储扩容计划" >> $REPORT_FILE
elseecho "🟢 存储充足: 磁盘使用率 ${USAGE_NUM}%" >> $REPORT_FILE
fi# Broker扩容建议
BROKER_COUNT=$(kafka-broker-api-versions.sh --bootstrap-server localhost:9092 2>/dev/null | wc -l)
TOPIC_COUNT=$(kafka-topics.sh --bootstrap-server localhost:9092 --list 2>/dev/null | wc -l)
PARTITION_COUNT=$(kafka-topics.sh --bootstrap-server localhost:9092 --describe 2>/dev/null | grep "PartitionCount" | awk '{sum += $2} END {print sum+0}')echo -e "\n集群规模分析:" >> $REPORT_FILE
echo "Broker数量: $BROKER_COUNT" >> $REPORT_FILE
echo "Topic数량: $TOPIC_COUNT" >> $REPORT_FILE
echo "总分区数: $PARTITION_COUNT" >> $REPORT_FILE
echo "平均每Broker分区数: $((PARTITION_COUNT / BROKER_COUNT))" >> $REPORT_FILEif [ $((PARTITION_COUNT / BROKER_COUNT)) -gt 1000 ]; thenecho "🔴 建议增加Broker节点: 平均分区数过高" >> $REPORT_FILE
elif [ $((PARTITION_COUNT / BROKER_COUNT)) -gt 500 ]; thenecho "🟡 考虑扩容: 平均分区数较高" >> $REPORT_FILE
fi# 5. 容量增长预测
echo -e "\n5. 容量增长预测" >> $REPORT_FILE
echo "================================" >> $REPORT_FILE# 基于历史数据的简单线性预测
if [ -f "/var/log/disk-usage-history.log" ]; thenRECENT_GROWTH=$(tail -30 /var/log/disk-usage-history.log | \awk 'NR==1{first=$1} END{if(NR>1) print ($1-first)/(NR-1); else print 0}')if awk "BEGIN {exit !($RECENT_GROWTH > 0)}"; thenMONTHS_TO_80=$(awk "BEGIN {printf \"%.1f\", (80-$USAGE_NUM)/$RECENT_GROWTH/30}")echo "基于最近趋势，预计 ${MONTHS_TO_80} 个月后达到80%使用率" >> $REPORT_FILEif awk "BEGIN {exit !($MONTHS_TO_80 < 6)}"; thenecho "⚠️ 建议尽快制定扩容计划" >> $REPORT_FILEfifi
fi# 6. 成本优化建议
echo -e "\n6. 成本优化建议" >> $REPORT_FILE
echo "================================" >> $REPORT_FILEecho "• 评估数据保留策略，删除不必要的历史数据" >> $REPORT_FILE
echo "• 启用消息压缩以减少存储需求" >> $REPORT_FILE
echo "• 优化分区数量以提高资源利用率" >> $REPORT_FILE
echo "• 考虑使用分层存储，将冷数据迁移到低成本存储" >> $REPORT_FILE
echo "• 定期审查Topic和消费者组，清理不用的资源" >> $REPORT_FILEecho -e "\n报告生成完成: $REPORT_FILE"
EOFchmod +x /usr/local/kafka-monitor/kafka-capacity-planner.shecho "✅ 容量规划脚本部署完成"
第三部分：最佳实践检查清单
# Kafka运维最佳实践检查清单## 🔧 配置管理最佳实践### Broker配置
- [ ] 设置合理的 `num.partitions` 默认值（建议12-50）
- [ ] 配置 `default.replication.factor=3` 确保数据冗余
- [ ] 设置 `min.insync.replicas=2` 平衡一致性和可用性
- [ ] 禁用 `unclean.leader.election.enable=false` 防止数据丢失
- [ ] 配置合适的 `log.retention.hours` 和 `log.retention.bytes`
- [ ] 启用 `log.cleaner.enable=true` 支持日志压缩### JVM配置
- [ ] 使用G1垃圾收集器 `-XX:+UseG1GC`
- [ ] 设置合理的堆内存大小（物理内存的25-50%）
- [ ] 配置GC日志便于问题诊断
- [ ] 启用JMX监控 `-Dcom.sun.management.jmxremote`### 系统配置
- [ ] 调整文件描述符限制 `ulimit -n 100000`
- [ ] 优化虚拟内存设置 `vm.swappiness=1`
- [ ] 配置网络参数优化吞吐量
- [ ] 使用合适的文件系统挂载选项（noatime）## 📊 监控和告警### 核心指标监控
- [ ] 消息堆积（Consumer Lag）
- [ ] 磁盘使用率和IO性能
- [ ] CPU和内存使用率
- [ ] 网络带宽使用情况
- [ ] JVM垃圾回收性能### 告警设置
- [ ] 消息堆积超过阈值（如50K消息）
- [ ] 磁盘使用率超过85%
- [ ] Broker节点下线
- [ ] 分区Leader不均衡
- [ ] 消费者组异常## 🛡️ 安全性配置### 访问控制
- [ ] 配置SSL/TLS加密传输
- [ ] 启用SASL认证机制
- [ ] 设置ACL权限控制
- [ ] 定期轮换证书和密钥### 网络安全
- [ ] 配置防火墙规则
- [ ] 使用VPC或专用网络
- [ ] 限制管理端口访问
- [ ] 定期安全扫描和更新## 📈 性能优化### 生产者优化
- [ ] 调整 `batch.size` 和 `linger.ms`
- [ ] 选择合适的压缩算法
- [ ] 配置 `acks` 参数平衡性能和可靠性
- [ ] 启用幂等性 `enable.idempotence=true`### 消费者优化
- [ ] 调整 `fetch.min.bytes` 和 `fetch.max.wait.ms`
- [ ] 设置合理的 `max.poll.records`
- [ ] 配置适当的 `session.timeout.ms`
- [ ] 实现手动offset提交控制### Topic设计
- [ ] 合理规划分区数量
- [ ] 选择适当的分区策略
- [ ] 设计合理的消息key
- [ ] 考虑消息顺序需求## 🔄 运维流程### 日常维护
- [ ] 定期检查集群健康状态
- [ ] 监控磁盘使用趋势
- [ ] 清理过期日志和快照
- [ ] 检查配置一致性### 变更管理
- [ ] 制定变更审批流程
- [ ] 准备回滚方案
- [ ] 在测试环境验证
- [ ] 分批次执行变更### 备份和恢复
- [ ] 定期备份配置文件
- [ ] 备份关键Topic数据
- [ ] 测试恢复流程
- [ ] 文档化恢复步骤## 🚨 故障预防### 容量规划
- [ ] 定期进行容量分析
- [ ] 预测存储增长趋势
- [ ] 制定扩容计划
- [ ] 测试扩容流程### 高可用性
- [ ] 部署多个Broker节点
- [ ] 配置合适的副本因子
- [ ] 实现跨机架部署
- [ ] 准备灾难恢复方案### 监控自动化
- [ ] 部署自动化监控系统
- [ ] 配置智能告警规则
- [ ] 实现自动化响应
- [ ] 定期测试告警有效性## 📚 文档和培训### 文档维护
- [ ] 维护系统架构文档
- [ ] 更新运维手册
- [ ] 记录故障处理案例
- [ ] 保持配置变更记录### 团队培训
- [ ] 定期进行技术培训
- [ ] 分享最佳实践经验
- [ ] 进行故障演练
- [ ] 建立知识库## ✅ 定期检查项目### 每日检查
- [ ] 集群整体状态
- [ ] 关键监控指标
- [ ] 告警处理情况
- [ ] 日志错误分析### 每周检查
- [ ] 性能趋势分析
- [ ] 容量使用情况
- [ ] 配置变更审查
- [ ] 备份完整性验证### 每月检查
- [ ] 安全漏洞扫描
- [ ] 性能基准测试
- [ ] 灾难恢复演练
- [ ] 文档更新维护### 每季度检查
- [ ] 系统架构审查
- [ ] 容量规划更新
- [ ] 成本优化分析
- [ ] 技术栈升级评估
🎯 总结和建议
通过本完整手册的学习和实践，您将掌握一套完整的Kafka线上问题排查方法和工具。这些内容基于大量的生产环境实践经验总结而成，可以帮助您：
- 快速定位问题：通过系统化的排查流程，快速定位各类问题的根因
- 防范于未然：通过预防性监控和维护，减少故障发生的概率
- 提高效率：使用自动化工具和脚本，提高问题解决效率
- 优化性能：通过配置优化和最佳实践，提升Kafka集群性能
🎓 学习路径建议
对于不同经验水平的读者，建议按以下路径学习：
初学者：
- 先阅读“快速诊断命令集合”，掌握基本命令
- 学习“集群状态全面检查”，理解Kafka架构
- 实践基础的监控和告警脚本
中级运维人员：
- 深入学习消息堆积和性能问题排查
- 掌握网络和连接问题诊断方法
- 实施全面的监控体系
高级专家：
- 研究紧急故障处理和自动恢复机制
- 优化生产环境配置和性能调优
- 建立完善的预防性维护体系
🔗 相关资源和工具
官方文档和工具：
- Apache Kafka 官方文档：[https://kafka.apache.org/documentation/]
- Kafka Manager：[https://github.com/yahoo/CMAK]
- Kafka Tools：[https://kafka.apache.org/downloads]
监控工具：
- Prometheus + Kafka Exporter
- Grafana 仪表板
- ElasticSearch + Kibana
- Zabbix 或 Nagios
学习资料：
- Kafka官方社区和论坛
- LinkedIn Kafka 最佳实践分享
- Confluent 技术博客和文档
📝 本手册使用说明
复制和使用
本手册中的所有脚本和配置都经过实际测试，可以直接复制使用。使用时请注意：
- 环境适配：根据你的实际环境修改相关参数
- 权限检查：确保脚本有正确的执行权限
- 备份重要数据：在执行修改操作前做好备份
- 渐进式部署：先在测试环境验证，再在生产环境使用
更新和维护
这个手册会根据Kafka版本更新和实践经验的积累持续改进。建议：
- 定期检查更新：关注Kafka新版本的功能和改进
- 反馈问题：如果发现问题或有改进建议，欢迎反馈
- 分享经验：将你的实践经验和改进分享给其他人
🚀 结语
Kafka作为现代大数据架构的核心组件，其稳定运行对业务系统至关重要。通过系统化的问题排查和预防性维护，我们可以：
- ⚙️ 提高效率：减少故障排查时间，快速定位和解决问题
- 🛡️ 降低风险：通过预防性监控，在问题发生前及时发现和处理
- 📈 优化性能：通过最佳实践和调优，持续提升系统性能
- 📚 经验传承：将实践经验文档化，促进团队知识共享
希望这个手册能够成为您日常Kafka运维工作中的得力助手，帮助您构建更加稳定、高效的Kafka集群。
记住：最好的问题解决方案是预防问题的发生。