1、获取Kafka消息堆积情况

vi check-kafka-lag.sh

#！/bin/bash

TOPIC="total_random"
GROUP_ID="etl-dw"
BOOTSTRAP_SERVER="node-01:9092,node-02:9092,node-03:9092"

# 检查第一个参数是否为数字
if ! [[ $1 =~ ^[0-9]+$ ]]; then
    echo "错误: 传入参数必须是数字" >&2
    exit 1
fi

# 获取LAG信息
LAG_INFO=`sh /data/app/kafka/bin/kafka-consumer-groups.sh --bootstrap-server $BOOTSTRAP_SERVER --group $GROUP_ID --describe | grep $TOPIC | awk -F " " '{print $6}'`

# 初始化LAG总和为0
total_lag=0

for lag in ${$LAG_INFO[@]};
do
  total_lag=$(($total_lag + $lag))
done

# 判断消息积压情况，是否重启程序
if [ $total_lag -gt $1 ]; then 
  echo "$(date +"%Y-%m-%d %H:%M:%S") Kafka Topic " $TOPIC " IN GROUP " $GROUP_ID " 消息堆积总量为：" $total_lag " 大于 " $1 "，停止DW模块"
  sh ./stop-dwkudu.sh
  
  # 判断是否执行成功
  if [ $? -eq 0 ]; then 
    sh ./start-dwkudu.sh
	echo "$(date +"%Y-%m-%d %H:%M:%S") 启动DW模块完成"
  else
    echo "$(date +"%Y-%m-%d %H:%M:%S") 停止DW模块失败"
  fi

else
  echo "$(date +"%Y-%m-%d %H:%M:%S") Kafka Topic " $TOPIC " IN GROUP " $GROUP_ID " 消息堆积总量为：" $total_lag " 小于 " $1
fi

2、创建定时任务

每隔1小时，执行一次

0 */1 * * * /data/app/etl-dw/shell/check-kafka-lag.sh 100000 >> /data/logs/etl-dw/check-kafka-lag.log 2>&1 &