未记录事实核验 · 状态：维护中
记录仅说明核验时间与维护状态，不保证适用于所有环境；使用前请留意正文中的版本、适用条件与风险提示。
warning 适用边界与维护说明
本文适合高级Kafka运维人员梳理监控、故障响应和灾难恢复任务，代码为未完成的示例骨架，不能直接作为生产恢复预案。灾难恢复按ZooKeeper模式展开，不能直接照搬到KRaft集群；数据与元数据恢复须以一致性备份、匹配的集群身份及明确的RPO/RTO为前提。当前副本迁移未提取并审查生成的执行计划，目录查询被误标为索引重建且在本机服务停止后执行，preferred选举不保证恢复离线分区；真实指标采集、多个恢复函数、部分通知和流量切换尚未实现，备份日期未传入，失败也未可靠阻断后续流程。偏移量重置前须确认消费组不活跃并预览目标偏移量，回退可能重复消费，前移可能跳过消息。事故金额、耗时区间及恢复报告均为示例，脚本结束、目录查询和报告生成不代表恢复成功；实际恢复结果须通过数据与消费状态核对及集群恢复验证确定。本次仅作全文结构审阅，未作外部事实核验或执行验证。
故障处理与恢复 - Kafka灾难恢复策略
以下为假设性事故场景，老赵为示例人物，损失金额并非实际案例数据：周六上午10点，正在逛街的运维专家老赵收到告警电话："Kafka集群彻底宕机了！整个数据中心断电，所有服务器都关机了！"公司所有业务系统瘫痪，损失每分钟数万元。"这就是最坏的情况，"老赵冷静地说，"启动灾难恢复预案，按照我们设计的步骤，一步步把集群恢复起来。"关键时刻，完善的故障恢复策略就是救命稻草。
🚨 故障监控和预警
📊 关键监控指标
# !/bin/bash
echo "=== Kafka故障监控指标配置 ==="
# 1. 系统级监控指标
setup_system_monitoring() {
echo "配置系统级监控..."
cat > /opt/monitoring/system-metrics.yaml << 'EOF'
# 系统监控指标配置
# CPU监控
cpu_metrics:
- name: cpu_usage_percent
threshold_warning: 70
threshold_critical: 90
check_interval: 30s
# 内存监控
memory_metrics:
- name: memory_usage_percent
threshold_warning: 80
threshold_critical: 95
check_interval: 30s
- name: jvm_heap_usage
threshold_warning: 75
threshold_critical: 90
check_interval: 30s
# 磁盘监控
disk_metrics:
- name: disk_usage_percent
path: /var/kafka-logs
threshold_warning: 80
threshold_critical: 90
check_interval: 60s
- name: disk_io_wait
threshold_warning: 20
threshold_critical: 40
check_interval: 30s
# 网络监控
network_metrics:
- name: network_connections
threshold_warning: 10000
threshold_critical: 15000
check_interval: 30s
EOF
echo "系统级监控配置完成"
}
# 2. Kafka服务监控
setup_kafka_monitoring() {
echo "配置Kafka服务监控..."
cat > /opt/monitoring/kafka-metrics.yaml << 'EOF'
# Kafka服务监控指标
# Broker监控
broker_metrics:
- name: broker_online_status
jmx_attribute: "kafka.server:type=KafkaServer,name=BrokerState"
expected_value: 3 # RUNNING状态
check_interval: 30s
- name: under_replicated_partitions
jmx_attribute: "kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions"
threshold_critical: 0 # 任何分区副本不足都是严重问题
check_interval: 30s
- name: offline_partitions
jmx_attribute: "kafka.controller:type=KafkaController,name=OfflinePartitionsCount"
threshold_critical: 0 # 任何离线分区都是严重问题
check_interval: 30s
# 生产者监控
producer_metrics:
- name: producer_request_rate
jmx_attribute: "kafka.producer:type=producer-metrics,client-id=*,name=request-rate"
threshold_warning: 1000
check_interval: 60s
- name: producer_error_rate
jmx_attribute: "kafka.producer:type=producer-metrics,client-id=*,name=record-error-rate"
threshold_warning: 0.01 # 每秒错误记录数阈值，不是1%错误率
threshold_critical: 0.05 # 每秒错误记录数阈值，不是5%错误率
check_interval: 30s
# 消费者监控
consumer_metrics:
- name: consumer_lag
jmx_attribute: "kafka.consumer:type=consumer-fetch-manager-metrics,client-id=*,name=records-lag-max"
threshold_warning: 10000
threshold_critical: 50000
check_interval: 60s
- name: consumer_heartbeat
jmx_attribute: "kafka.consumer:type=consumer-coordinator-metrics,client-id=*,name=heartbeat-rate"
threshold_critical: 0.1 # 心跳频率过低
check_interval: 30s
EOF
echo "Kafka服务监控配置完成"
}
# 3. 业务级监控
setup_business_monitoring() {
echo "配置业务级监控..."
cat > /opt/monitoring/business-metrics.yaml << 'EOF'
# 业务监控指标
# 消息量监控
message_metrics:
- name: daily_message_volume
topic_pattern: "business-.*"
threshold_warning: -20% # 日消息量下降20%
threshold_critical: -50% # 日消息量下降50%
check_interval: 1h
- name: message_processing_latency
topic_pattern: "realtime-.*"
threshold_warning: 1000ms
threshold_critical: 5000ms
check_interval: 5m
# 数据质量监控
data_quality_metrics:
- name: duplicate_message_rate
threshold_warning: 0.001 # 0.1%重复率
threshold_critical: 0.01 # 1%重复率
check_interval: 15m
- name: schema_validation_error_rate
threshold_warning: 0.01
threshold_critical: 0.05
check_interval: 5m
EOF
echo "业务级监控配置完成"
}
# 执行监控配置
setup_system_monitoring
setup_kafka_monitoring
setup_business_monitoring
🔔 告警机制配置
# !/usr/bin/env python3
"""
Kafka故障告警系统
"""
import json
import time
import requests
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from kafka import KafkaConsumer
from kafka.errors import KafkaError
class KafkaAlertManager:
"""Kafka告警管理器"""
def __init__(self, config_file):
with open(config_file, 'r') as f:
self.config = json.load(f)
self.alert_history = {}
self.setup_notification_channels()
def setup_notification_channels(self):
"""配置通知渠道"""
self.notification_channels = {
'email': self.send_email_alert,
'webhook': self.send_webhook_alert,
'sms': self.send_sms_alert
}
def check_kafka_health(self):
"""检查Kafka集群健康状态"""
alerts = []
for cluster in self.config['clusters']:
try:
# 检查连接性
consumer = KafkaConsumer(
bootstrap_servers=cluster['bootstrap_servers'],
consumer_timeout_ms=5000
)
consumer.close()
# 检查关键指标
cluster_alerts = self.check_cluster_metrics(cluster)
alerts.extend(cluster_alerts)
except KafkaError as e:
alert = {
'level': 'CRITICAL',
'cluster': cluster['name'],
'type': 'CONNECTION_FAILED',
'message': f"无法连接到Kafka集群: {str(e)}",
'timestamp': time.time()
}
alerts.append(alert)
return alerts
def check_cluster_metrics(self, cluster):
"""检查集群指标"""
alerts = []
try:
# 获取JMX指标
metrics = self.get_jmx_metrics(cluster)
# 检查副本不足分区
under_replicated = metrics.get('under_replicated_partitions', 0)
if under_replicated > 0:
alerts.append({
'level': 'CRITICAL',
'cluster': cluster['name'],
'type': 'UNDER_REPLICATED_PARTITIONS',
'message': f"发现{under_replicated}个副本不足的分区",
'timestamp': time.time()
})
# 检查离线分区
offline_partitions = metrics.get('offline_partitions', 0)
if offline_partitions > 0:
alerts.append({
'level': 'CRITICAL',
'cluster': cluster['name'],
'type': 'OFFLINE_PARTITIONS',
'message': f"发现{offline_partitions}个离线分区",
'timestamp': time.time()
})
# 检查磁盘使用率
disk_usage = metrics.get('disk_usage_percent', 0)
if disk_usage > 90:
alerts.append({
'level': 'CRITICAL',
'cluster': cluster['name'],
'type': 'HIGH_DISK_USAGE',
'message': f"磁盘使用率达到{disk_usage}%",
'timestamp': time.time()
})
elif disk_usage > 80:
alerts.append({
'level': 'WARNING',
'cluster': cluster['name'],
'type': 'HIGH_DISK_USAGE',
'message': f"磁盘使用率达到{disk_usage}%",
'timestamp': time.time()
})
except Exception as e:
alerts.append({
'level': 'WARNING',
'cluster': cluster['name'],
'type': 'METRICS_COLLECTION_FAILED',
'message': f"指标收集失败: {str(e)}",
'timestamp': time.time()
})
return alerts
def get_jmx_metrics(self, cluster):
"""获取JMX指标"""
# 简化示例，实际应该通过JMX API获取
return {
'under_replicated_partitions': 0,
'offline_partitions': 0,
'disk_usage_percent': 65
}
def process_alerts(self, alerts):
"""处理告警"""
for alert in alerts:
# 检查告警去重
if self.should_suppress_alert(alert):
continue
# 发送告警
self.send_alert(alert)
# 记录告警历史
self.record_alert(alert)
def should_suppress_alert(self, alert):
"""告警去重逻辑"""
alert_key = f"{alert['cluster']}:{alert['type']}"
if alert_key in self.alert_history:
last_alert_time = self.alert_history[alert_key]['last_time']
suppress_duration = self.config.get('suppress_duration', 300) # 5分钟
if time.time() - last_alert_time < suppress_duration:
return True
return False
def send_alert(self, alert):
"""发送告警"""
channels = self.config['notification_channels']
if alert['level'] == 'CRITICAL':
# 严重告警发送到所有渠道
for channel in channels:
self.notification_channels[channel](alert)
else:
# 警告级别只发送邮件
self.send_email_alert(alert)
def send_email_alert(self, alert):
"""发送邮件告警"""
try:
msg = MIMEMultipart()
msg['From'] = self.config['email']['from']
msg['To'] = ', '.join(self.config['email']['to'])
msg['Subject'] = f"[{alert['level']}] Kafka集群告警 - {alert['cluster']}"
body = f"""
告警级别: {alert['level']}
集群名称: {alert['cluster']}
告警类型: {alert['type']}
告警消息: {alert['message']}
发生时间: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(alert['timestamp']))}
"""
msg.attach(MIMEText(body, 'plain'))
server = smtplib.SMTP(self.config['email']['smtp_server'])
server.starttls()
server.login(self.config['email']['username'],
self.config['email']['password'])
server.send_message(msg)
server.quit()
except Exception as e:
print(f"邮件发送失败: {str(e)}")
def send_webhook_alert(self, alert):
"""发送Webhook告警"""
try:
webhook_url = self.config['webhook']['url']
payload = {
'text': f"🚨 Kafka告警\n"
f"级别: {alert['level']}\n"
f"集群: {alert['cluster']}\n"
f"消息: {alert['message']}"
}
response = requests.post(webhook_url, json=payload, timeout=10)
response.raise_for_status()
except Exception as e:
print(f"Webhook发送失败: {str(e)}")
def send_sms_alert(self, alert):
"""发送短信告警"""
# 集成短信服务API
pass
def record_alert(self, alert):
"""记录告警历史"""
alert_key = f"{alert['cluster']}:{alert['type']}"
self.alert_history[alert_key] = {
'last_time': alert['timestamp'],
'count': self.alert_history.get(alert_key, {}).get('count', 0) + 1
}
def run(self):
"""运行告警监控"""
print("启动Kafka故障监控...")
while True:
try:
alerts = self.check_kafka_health()
if alerts:
self.process_alerts(alerts)
print(f"处理了{len(alerts)}个告警")
time.sleep(self.config.get('check_interval', 60))
except KeyboardInterrupt:
print("监控程序停止")
break
except Exception as e:
print(f"监控过程中发生错误: {str(e)}")
time.sleep(60)
if __name__ == "__main__":
# 配置文件示例
config = {
"clusters": [
{
"name": "production",
"bootstrap_servers": ["prod-kafka-01:9092", "prod-kafka-02:9092"]
}
],
"notification_channels": ["email", "webhook"],
"email": {
"smtp_server": "smtp.company.com",
"from": "kafka-alerts@company.com",
"to": ["ops-team@company.com"],
"username": "alerts",
"password": "password"
},
"webhook": {
"url": "https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK"
},
"check_interval": 30,
"suppress_duration": 300
}
# 保存配置
with open('alert_config.json', 'w') as f:
json.dump(config, f, indent=2)
# 启动告警管理器
alert_manager = KafkaAlertManager('alert_config.json')
alert_manager.run()
🔧 常见故障处理
💾 数据恢复策略
# !/bin/bash
echo "=== Kafka数据恢复工具集 ==="
# 1. 分区数据恢复
recover_partition_data() {
local topic=$1
local partition=$2
local backup_path=$3
echo "恢复分区数据: $topic-$partition"
# 停止相关broker
echo "停止Broker服务..."
systemctl stop kafka
# 备份当前损坏的数据
echo "备份损坏数据..."
mv /var/kafka-logs/$topic-$partition /var/kafka-logs/$topic-$partition.corrupted.$(date +%s)
# 从备份恢复数据
echo "从备份恢复数据..."
cp -r $backup_path/$topic-$partition /var/kafka-logs/
# 修复权限
chown -R kafka:kafka /var/kafka-logs/$topic-$partition
# 重建索引
echo "重建索引文件..."
kafka-log-dirs.sh --bootstrap-server localhost:9092 \
--describe --json | jq .
# 启动broker
echo "启动Broker服务..."
systemctl start kafka
echo "分区数据恢复完成"
}
# 2. 索引文件重建
rebuild_index_files() {
local log_dir=$1
echo "重建索引文件: $log_dir"
# 停止Kafka服务
systemctl stop kafka
# 删除损坏的索引文件
find $log_dir -name "*.index" -delete
find $log_dir -name "*.timeindex" -delete
# 重建索引
kafka-log-dirs.sh --bootstrap-server localhost:9092 \
--describe --json > /tmp/log_dirs_before.json
# 启动Kafka（会自动重建索引）
systemctl start kafka
# 验证索引重建
sleep 30
kafka-log-dirs.sh --bootstrap-server localhost:9092 \
--describe --json > /tmp/log_dirs_after.json
echo "索引文件重建完成"
}
# 3. 消费者偏移量恢复
recover_consumer_offsets() {
local group_id=$1
local topic=$2
local target_timestamp=$3
echo "恢复消费者偏移量: $group_id"
# 停止消费者
echo "停止消费者组: $group_id"
# 需要应用程序配合停止消费者
# 重置偏移量到指定时间
kafka-consumer-groups.sh \
--bootstrap-server localhost:9092 \
--group $group_id \
--topic $topic \
--reset-offsets \
--to-datetime $target_timestamp \
--execute
echo "消费者偏移量恢复完成"
}
# 4. 集群状态恢复
recover_cluster_state() {
echo "恢复集群状态..."
# 检查集群状态
echo "=== 当前集群状态 ==="
kafka-topics.sh --bootstrap-server localhost:9092 --list
# 检查副本分布
echo "=== 副本分布检查 ==="
kafka-topics.sh --bootstrap-server localhost:9092 \
--describe | grep "ReplicationFactor\|Replicas"
# 重新分配副本
echo "=== 生成副本重新分配计划 ==="
cat > /tmp/topics-to-move.json << 'EOF'
{
"topics": [
{"topic": "important-topic"}
],
"version": 1
}
EOF
kafka-reassign-partitions.sh \
--bootstrap-server localhost:9092 \
--topics-to-move-json-file /tmp/topics-to-move.json \
--broker-list "1,2,3" \
--generate > /tmp/reassignment-plan.json
# 执行重新分配
echo "=== 执行副本重新分配 ==="
kafka-reassign-partitions.sh \
--bootstrap-server localhost:9092 \
--reassignment-json-file /tmp/reassignment-plan.json \
--execute
# 验证重新分配
echo "=== 验证重新分配结果 ==="
kafka-reassign-partitions.sh \
--bootstrap-server localhost:9092 \
--reassignment-json-file /tmp/reassignment-plan.json \
--verify
echo "集群状态恢复完成"
}
# 5. 自动化故障恢复脚本
automated_recovery() {
local cluster_id=$1
echo "启动自动化故障恢复: $cluster_id"
# 检查故障类型
if ! check_broker_failure; then
echo "检测到Broker故障，执行Broker恢复..."
recover_broker_failure
fi
if check_partition_failure; then
echo "检测到分区故障，执行分区恢复..."
recover_partition_failure
fi
if ! check_network_partition; then
echo "检测到网络分区，执行网络恢复..."
recover_network_partition
fi
echo "自动化故障恢复完成"
}
# 故障检测函数
check_broker_failure() {
# 检查broker是否在线
kafka-broker-api-versions.sh --bootstrap-server localhost:9092 \
--timeout 5000 > /dev/null 2>&1
return $?
}
check_partition_failure() {
# 检查是否有离线分区
offline_partitions=$(kafka-topics.sh --bootstrap-server localhost:9092 \
--describe | grep "Leader: -1" | wc -l)
[ $offline_partitions -gt 0 ]
}
check_network_partition() {
# 检查网络连通性
# 简化检查，实际应该检查所有broker间的连通性
ping -c 1 -W 3 kafka-broker-2 > /dev/null 2>&1
return $?
}
# 恢复函数
recover_broker_failure() {
echo "执行Broker故障恢复..."
systemctl restart kafka
sleep 30
# 验证恢复
if kafka-broker-api-versions.sh --bootstrap-server localhost:9092 \
--timeout 5000 > /dev/null 2>&1; then
echo "Broker恢复成功"
else
echo "Broker恢复失败，需要人工介入"
fi
}
recover_partition_failure() {
echo "执行分区故障恢复..."
# 触发领导者选举
kafka-leader-election.sh --bootstrap-server localhost:9092 \
--election-type preferred --all-topic-partitions
}
recover_network_partition() {
echo "执行网络分区恢复..."
# 重启网络服务
systemctl restart network
sleep 10
# 重启Kafka服务
systemctl restart kafka
}
# 命令行参数处理
case "$1" in
"recover-partition")
recover_partition_data "$2" "$3" "$4"
;;
"rebuild-index")
rebuild_index_files "$2"
;;
"recover-offsets")
recover_consumer_offsets "$2" "$3" "$4"
;;
"recover-cluster")
recover_cluster_state
;;
"auto-recover")
automated_recovery "$2"
;;
*)
echo "用法: $0 {recover-partition|rebuild-index|recover-offsets|recover-cluster|auto-recover}"
echo "示例:"
echo " $0 recover-partition user-events 0 /backup/kafka-data"
echo " $0 rebuild-index /var/kafka-logs"
echo " $0 recover-offsets my-consumer-group user-events 2023-01-01T00:00:00.000"
echo " $0 recover-cluster"
echo " $0 auto-recover production"
exit 1
;;
esac
🎯 灾难恢复计划
📋 灾难恢复流程
# !/bin/bash
echo "=== Kafka灾难恢复执行计划 ==="
# 1. 灾难评估
assess_disaster() {
echo "=== 第一阶段：灾难评估 ==="
local disaster_type=$1
local affected_components=$2
echo "灾难类型: $disaster_type"
echo "影响组件: $affected_components"
# 评估数据损失程度
echo "评估数据损失程度..."
# 检查可用的broker
available_brokers=$(kafka-broker-api-versions.sh \
--bootstrap-server localhost:9092 --timeout 5000 2>/dev/null | \
grep "broker" | wc -l)
echo "可用Broker数量: $available_brokers"
# 检查数据完整性
echo "检查数据完整性..."
check_data_integrity
# 评估恢复时间
estimate_recovery_time $disaster_type $affected_components
}
# 2. 紧急响应
emergency_response() {
echo "=== 第二阶段：紧急响应 ==="
# 通知相关人员
echo "发送紧急通知..."
send_disaster_notification
# 启动应急集群
echo "启动应急集群..."
start_emergency_cluster
# 切换流量
echo "切换业务流量..."
switch_traffic_to_emergency
# 保护现场
echo "保护故障现场..."
preserve_failure_site
}
# 3. 数据恢复
data_recovery() {
echo "=== 第三阶段：数据恢复 ==="
# 从备份恢复
echo "从备份恢复数据..."
restore_from_backup
# 重建索引
echo "重建索引和元数据..."
rebuild_metadata
# 验证数据完整性
echo "验证数据完整性..."
verify_data_integrity
# 同步增量数据
echo "同步增量数据..."
sync_incremental_data
}
# 4. 服务恢复
service_recovery() {
echo "=== 第四阶段：服务恢复 ==="
# 启动生产集群
echo "启动生产集群..."
start_production_cluster
# 验证集群状态
echo "验证集群状态..."
verify_cluster_health
# 恢复客户端连接
echo "恢复客户端连接..."
restore_client_connections
# 切回生产流量
echo "切回生产流量..."
switch_back_to_production
}
# 5. 验证和监控
verification_and_monitoring() {
echo "=== 第五阶段：验证和监控 ==="
# 全面健康检查
echo "执行全面健康检查..."
comprehensive_health_check
# 性能基准测试
echo "执行性能基准测试..."
performance_benchmark
# 强化监控
echo "强化监控..."
enhance_monitoring
# 生成恢复报告
echo "生成恢复报告..."
generate_recovery_report
}
# 辅助函数实现
check_data_integrity() {
# 检查Topic完整性
echo "检查Topic完整性..."
kafka-topics.sh --bootstrap-server localhost:9092 \
--describe 2>/dev/null | grep -c "Topic:"
# 检查分区分布
echo "检查分区分布..."
kafka-topics.sh --bootstrap-server localhost:9092 \
--describe 2>/dev/null | grep "PartitionCount\|ReplicationFactor"
}
estimate_recovery_time() {
local disaster_type=$1
local affected_components=$2
echo "以下时间区间仅为情景假设，不是恢复承诺；实际耗时取决于数据量、备份状态、带宽和演练结果。"
case $disaster_type in
"single_broker_failure")
echo "预估恢复时间: 30-60分钟"
;;
"multiple_broker_failure")
echo "预估恢复时间: 2-4小时"
;;
"datacenter_failure")
echo "预估恢复时间: 4-8小时"
;;
"complete_disaster")
echo "预估恢复时间: 8-24小时"
;;
*)
echo "预估恢复时间: 需要进一步评估"
;;
esac
}
send_disaster_notification() {
# 发送紧急通知
cat > /tmp/disaster_notification.txt << 'EOF'
🚨 KAFKA集群灾难告警 🚨
发生时间: $(date)
灾难级别: CRITICAL
影响范围: 生产环境Kafka集群
预计恢复时间: 评估中
应急措施:
1. 已启动应急响应流程
2. 正在评估数据损失程度
3. 准备启动备用集群
请相关人员立即响应！
EOF
# 发送到各个通知渠道
echo "发送邮件通知..."
echo "发送短信通知..."
echo "发送即时消息通知..."
}
start_emergency_cluster() {
echo "启动应急Kafka集群..."
# 启动备用ZooKeeper
for i in {1..3}; do
ssh backup-zk-0$i "sudo systemctl start zookeeper"
done
sleep 10
# 启动备用Kafka集群
for i in {1..3}; do
ssh backup-kafka-0$i "sudo systemctl start kafka"
done
# 验证应急集群状态
sleep 30
kafka-topics.sh --bootstrap-server backup-kafka-01:9092 --list
}
switch_traffic_to_emergency() {
echo "切换流量到应急集群..."
# 更新DNS记录
echo "更新DNS记录..."
# 更新负载均衡器配置
echo "更新负载均衡器配置..."
# 通知应用程序切换连接
echo "通知应用程序切换连接..."
}
restore_from_backup() {
echo "从备份恢复数据..."
local backup_date=$1
local backup_path="/backup/kafka/$backup_date"
# 停止目标集群
echo "停止目标集群..."
# 清理现有数据
echo "清理现有数据目录..."
# 恢复数据文件
echo "恢复数据文件..."
for broker in {1..6}; do
rsync -av $backup_path/broker-$broker/ \
kafka-0$broker:/var/kafka-logs/
done
# 恢复ZooKeeper数据
echo "恢复ZooKeeper数据..."
for zk in {1..3}; do
rsync -av $backup_path/zookeeper-$zk/ \
zk-0$zk:/var/zookeeper/
done
}
comprehensive_health_check() {
echo "执行全面健康检查..."
# 检查所有broker状态
echo "检查Broker状态..."
for i in {1..6}; do
echo "检查kafka-0$i..."
kafka-broker-api-versions.sh \
--bootstrap-server kafka-0$i:9092 --timeout 5000
done
# 检查Topic状态
echo "检查Topic状态..."
kafka-topics.sh --bootstrap-server localhost:9092 \
--describe | grep -E "(Topic:|PartitionCount|ReplicationFactor)"
# 检查消费者组状态
echo "检查消费者组状态..."
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
--list
# 检查复制状态
echo "检查复制状态..."
kafka-topics.sh --bootstrap-server localhost:9092 \
--describe --under-replicated-partitions
}
generate_recovery_report() {
echo "生成灾难恢复报告..."
cat > /tmp/disaster_recovery_report.md << EOF
# Kafka灾难恢复报告（模板，非实际恢复记录）
以下事故类型、影响范围和阶段耗时为示例假设；发生及完成时间应采用实际记录，不应以报告生成时间代替。
## 基本信息
- 灾难发生时间: $(date)
- 恢复完成时间: $(date)
- 总恢复耗时: XXX小时XX分钟
- 影响范围: 生产环境Kafka集群
## 灾难详情
- 灾难类型: 数据中心断电
- 受影响组件: 所有Kafka broker和ZooKeeper节点
- 数据损失情况: 待验证；仅在恢复数据与源端记录、备份及消费状态核对通过后，才可填写“无数据丢失”
## 恢复过程
1. 紧急响应 (30分钟)
2. 应急集群启动 (60分钟)
3. 数据恢复 (4小时)
4. 服务恢复 (2小时)
5. 验证监控 (1小时)
## 经验教训
1. 备份验证需要加强
2. 自动化恢复脚本需要优化
3. 监控告警需要完善
## 改进建议
1. 增加跨区域备份
2. 完善自动故障切换
3. 加强运维培训
EOF
echo "恢复报告已生成: /tmp/disaster_recovery_report.md"
}
# 主流程
main() {
local disaster_type=$1
local affected_components=$2
echo "开始执行Kafka灾难恢复流程..."
assess_disaster "$disaster_type" "$affected_components"
emergency_response
data_recovery
service_recovery
verification_and_monitoring
echo "灾难恢复流程执行完成！"
}
# 脚本入口
if [ $# -eq 0 ]; then
echo "用法: $0 <disaster_type> [affected_components]"
echo "灾难类型:"
echo " single_broker_failure - 单个broker故障"
echo " multiple_broker_failure - 多个broker故障"
echo " datacenter_failure - 数据中心故障"
echo " complete_disaster - 完全灾难"
exit 1
fi
main "$1" "$2"
✅ 掌握检查
完成以下检查，确保你已经掌握了故障处理与恢复：
- [ ] 建立完善的监控和告警体系
- [ ] 掌握各种故障场景的处理方法
- [ ] 能够执行数据恢复和集群恢复
- [ ] 制定灾难恢复计划和流程
- [ ] 具备故障预防和应急响应能力
🔗 学习连接
前置知识: 多集群架构 - 企业级部署架构 下一步学习: Kafka - 生产环境经验总结 相关概念:
- 高可用性架构
- 灾难恢复规划
- 运维自动化
tip 💡 故障处理原则
构建可靠Kafka系统的关键要点：
- 🔍 预防为主：通过监控及时发现潜在问题
- ⚡ 快速响应：建立高效的故障响应机制
- 📋 标准化流程：制定详细的故障处理流程
- 🎯 持续改进：从每次故障中总结经验教训
记住：最好的故障恢复是永远不需要恢复！