1.创建monitor_kafka脚本，并赋予对应权限

#在/etc/zabbix/scripts创建脚本，并赋予权限
vim monitor_kafka.sh
chmod +x /etc/zabbix/scripts/monitor_kafka.sh

监控current_offset、log_end_offset、lag、ratio(消息积压占比情况)

#!/bin/bash
#Global Variables
VERSION="0.2"
PROGNAME=`basename $0`
PROGPATH=`dirname $0`
KAFKA_BIN_PATH="/home/kafka_2.12-3.6.0/bin"
KAFKA_HOST="ip地址:9092"
CONFIG_FILE=""
CG=""
TN=""

#Help function
print_help() {
  echo "This zabbix script can discovery all consumer groups in a Kafka server and calculate the global lag for a specific consumer group"
  echo "Usage: $PROGNAME"
  echo "    -H (--bootstrap-server)  <host>  Hostname or IP address of Kafka server"
  echo "    -B (--path)      <path> Path to kafka-consumer-groups.sh script"
  echo "    -C (--config)    <config_file> Configuration parameters file (with jaas if needed)"
  echo "    -G (--group)     <consumer_group> Name of the consumer group for lag"
  echo "    -T (--topic)     <topic> Name of the topic"
  echo "    -d (--discovery) List all the consumer group for topic discovery"
  echo "    -l (--lag)       Calculate the global lag for a specific consumer group"
  echo "    -o (--offset)    Calculate the offset of this consumer group on this topic"
  echo "    -s (--size)      Calculate the log size of this topic"
  echo "    -r (--ratio)     Calculate the backlog ratio of message for this consumer group and topic"
  echo "    -v (--version)   Script version"
  echo "    -h (--help)      Script usage"
}

#Check presence of required parameter's number
if [ "$#" -lt 1 ]; then
  echo "PROGNAME: requires at least three parameters"
  print_help
  exit 1
fi

#Getting Parameters options
OPTS=$(getopt -o B:C:G:T:H:P:dghlsorv -l bootstrap-server:,path:,config:,discovery,group:,topic:,help,lag,size,offset,ratio,version -n "$(basename $0)" -- "$@")
eval set -- "$OPTS"
while true
do
  case $1 in
    -H|--bootstrap-server)
      KAFKA_HOST="$2"
      shift 2
      ;;
    -B|--path)
      KAFKA_BIN_PATH="$2"
      shift 2
      ;;
    -C|--config)
      CONFIG_FILE="$2"
      shift 2
      ;;
    -G|--group)
      CG="$2"
      shift 2
      ;;
    -T|--topic)
      TN="$2"
      shift 2
      ;;
    -d|--discovery)
      CG_DISCOVERY="true"
      shift
      ;;
    -h|--help)
      print_help
      exit 0
      ;;
    -l|--lag)
      CG_LAG="true"
      shift
      ;;
    -s|--size)
      CG_LOGSIZE="true"
      shift
      ;;
	-r|--ratio)
	  CG_RATIO="true"
	  shift
	  ;;
    -o|--offset)
      CG_OFFSET="true"
      shift
      ;;
    -v|--version)
      print_version
      exit 0
      ;;
    --)
      shift ; break
      ;;
    *)
      echo "Unknown argument: $1"
      print_help
      exit 1
      ;;
  esac
done

if [ "$CG_LAG" ] && [ "$CG" == "" ]; then
  echo "Consumer group must to be declare."
  exit 1
fi

if [ "$CG_OFFSET" ] && ( [ "$CG" == "" ] || [ "$TN" == "" ]); then
  echo "Consumer group and topic name must to be declare."
  exit 1
fi

if [ "$CG_LOGSIZE" ] && [ "$TN" == "" ]; then
  echo "Topic name must to be declare."
  exit 1
fi

if [ "$CG_RATIO" ] && ( [ "$CG" == "" ] || [ "$TN" == "" ]); then
  echo "Consumer group and topic name must to be declare."
  exit 1
fi

source /etc/profile 2>/dev/null

#Zabbix's discovery
if [ "$CG_DISCOVERY" ]; then
  CG_LIST=`${KAFKA_BIN_PATH}/kafka-consumer-groups.sh --bootstrap-server ${KAFKA_HOST} --list`
  CG_TN_LIST=`for i in ${CG_LIST}; do ${KAFKA_BIN_PATH}/kafka-consumer-groups.sh --bootstrap-server ${KAFKA_HOST} --group $i --describe|grep $i|awk -v OFS="_" '{print $1,$2}'|sort -u; done`
  ZBX_DISCO_LIST=`for i in ${CG_TN_LIST}; do echo -en "{"; echo -en "\"{#TOPIC_GROUP}\":\"$i\""; echo -en "},"; done`
  ZBX_DISCO_LIST=${ZBX_DISCO_LIST%?};
  echo -e "{\"data\":[${ZBX_DISCO_LIST}]}"
fi

#Consumer group lag
if [ "$CG_LAG" ]; then
	${KAFKA_BIN_PATH}/kafka-consumer-groups.sh --bootstrap-server ${KAFKA_HOST} --describe --group ${CG} |tail -n +3 |grep ${TN} |awk '{sum+=$6} END {print sum}'
fi

#Consumer group logsize
if [ "$CG_LOGSIZE" ]; then
        # ${KAFKA_BIN_PATH}/kafka-consumer-groups.sh --bootstrap-server ${KAFKA_HOST} --describe --group ${CG} |tail -n +3 |grep ${TN} |awk '{sum+=$5} END {print sum}'
   ${KAFKA_BIN_PATH}/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list ${KAFKA_HOST} --topic ${TN} --time -1| awk -F ":" '{sum+=$3} END {print sum}'
fi

#Consumer group offset
if [ "$CG_OFFSET" ]; then
   ${KAFKA_BIN_PATH}/kafka-consumer-groups.sh --bootstrap-server ${KAFKA_HOST} --describe --group ${CG} |tail -n +3 |grep ${TN} |awk '{sum+=$4} END {print sum}'
fi

#Consumert group message backlog ratio
if [ "$CG_RATIO" ]; then
	#${KAFKA_BIN_PATH}/kafka-consumer-groups.sh --bootstrap-server ${KAFKA_HOST} --describe --group ${CG} |tail -n +3 |grep ${TN} |awk 'if($5 != 0) { var ratio = ($6 / $5) * 100; printf("%.2f",$ratio)}else {print  0};'
	# 1. 首先执行 kafka-consumer-groups.sh 命令获取消费者组相关信息，并去除前两行（表头等信息）
consumer_groups_info=$(${KAFKA_BIN_PATH}/kafka-consumer-groups.sh --bootstrap-server ${KAFKA_HOST} --describe --group ${CG} |tail -n +3)

# 2. 从上述信息中筛选出与指定主题（${TN}）相关的行
		filtered_info=$(echo "$consumer_groups_info" | grep ${TN})

# 3. 使用 awk 命令对筛选后的每一行进行处理
		while read line; do
		    # 这里使用 awk 处理每一行数据，注意双引号的使用，以确保变量能正确替换
		    result=$(echo "$line" | awk "{
				if (\$5!= 0) {
					ratio = (\$6 / \$5) * 100;
					printf(\"%.2f\", ratio);
				} else {
					print 0;
				}
			}")
    echo "$result"
	done <<< "$filtered_info"
fi

2.在linux系统测试脚本是否能否正常输出信息

#运行脚本信息
#获取对应分组下对应主题的消息积压数量(lag)
sh /etc/zabbix/scripts/monitor_kafka.sh -T jq_weather -G test0512 -l
#获取对应分组下对应主题的消息积压占比情况(lag/log-end-offset)
sh /etc/zabbix/scripts/monitor_kafka.sh -T jq_weather -G test0512 -r
#获取对应分组下对应主题的消息current-offset
sh /etc/zabbix/scripts/monitor_kafka.sh -T jq_weather -G test0512 -o
#获取对应分组下对应主题的消息log-end-offset
sh /etc/zabbix/scripts/monitor_kafka.sh -T jq_weather -s 

3.修改zabbix中agent2的配置文件

vim /etc/zabbix/zabbix_agent2.conf
#加入下面的配置信息
UserParameter=kafka.lag[*],/etc/zabbix/scripts/monitor_kafka.sh -T $1 -G $2 -l
UserParameter=kafka.ratio[*],/etc/zabbix/scripts/monitor_kafka.sh -T $1 -G $2 -r
UserParameter=kafka.offset[*],/etc/zabbix/scripts/monitor_kafka.sh -T $1 -G $2 -o
UserParameter=kafka.logsize[*],/etc/zabbix/scripts/monitor_kafka.sh -T $1 -s

4.在zabbix中创建指标

5.创建指标完成后，点击下面测试按钮，看是否能正常输出

6.问题分析

1.在Window里面创建好monitor_kafka.sh脚本上传到linux进行运行时，报错信息如下：

运行如下命令

sed -i 's/\r$//' monitor_kafka_lag.sh