监控消息中间件状态的脚本如何写

从零构建高可用运维体系

目录导读

1. 为什么需要监控消息中间件？ —— 核心价值与常见风险
2. 监控脚本的基本要素 —— 指标、频率、报警策略
3. 实战脚本编写（以Kafka + RabbitMQ为例） —— 代码拆解与原理
4. 进阶技巧：分布式环境下的监控自动化 —— 集成Prometheus与钉钉通知
5. 常见问题与解答（Q&A） —— 覆盖90%运维场景的纠偏

为什么需要监控消息中间件？

消息中间件（如Kafka、RabbitMQ、RocketMQ）是现代微服务架构的“神经中枢”，一旦其状态异常，轻则消息积压、重则服务雪崩。脚本化监控能实现：

• 提前发现异常：磁盘写满、连接数暴增、消费延迟通常有前兆信号。
• 自动化故障恢复：脚本可配合重启服务、清理队列等操作。
• 降低MTTR（平均修复时间）：无需人工登录服务器逐一检查。

常见故障场景：

• RabbitMQ内存水位超过40%会触发阻塞（数据来源：官方文档）。
• Kafka分区副本ISR（同步副本）数量异常,导致消息丢失。

监控脚本的基本要素

一个合格的监控脚本应包含四个维度：

代码结构建议：

# 通用伪代码逻辑
while true:
    获取中间件状态()
    if 状态异常:
        触发报警()
        执行修复动作()  # 可选
    sleep(检测周期)

实战脚本编写（以Kafka + RabbitMQ为例）

场景1：Kafka消费者组延迟监控

原理：通过Kafka自带的kafka-consumer-groups.sh工具获取每个分区的LAG（滞后消息数）。
脚本核心：

#!/bin/bash
# kafka_monitor.sh
BOOTSTRAP_SERVER="localhost:9092"
GROUP_ID="my-consumer-group"
# 获取当前LAG值
LAG=$(./kafka-consumer-groups.sh \
  --bootstrap-server $BOOTSTRAP_SERVER \
  --group $GROUP_ID \
  --describe 2>/dev/null \
  | awk '{print $5}' | grep -v -e '^$' | awk '{sum+=$1} END {print sum}')
# 判断并报警
if [ "$LAG" -gt 10000 ]; then
  echo "警告：消费者组 $GROUP_ID 积压 $LAG 条消息"
  # 发送报警到企业微信/钉钉
  curl -X POST "你的报警webhook" -d "{\"msgtype\":\"text\",\"text\":{\"content\":\"Kafka积压严重:$LAG\"}}"
fi

场景2：RabbitMQ队列深度与连接数

原理：RabbitMQ Management API返回JSON格式的硬件指标。
脚本核心：

import requests, json, sys
def check_rabbitmq(host='localhost', port=15672, user='guest', passwd='guest'):
    url = f'http://{host}:{port}/api/queues'
    resp = requests.get(url, auth=(user, passwd), timeout=5)
    for queue in resp.json():
        if queue['messages'] > 5000:  # 阈值可调
            print(f"队列 {queue['name']} 积压 {queue['messages']} 条")
            # 可触发API清空队列或扩容消费者
    # 检查连接数
    conn_url = f'http://{host}:{port}/api/connections'
    conn_count = len(requests.get(conn_url, auth=(user, passwd)).json())
    if conn_count < 10:  # 低于基线告警
        print("连接数异常偏低")

场景3：进程与端口保活检测

核心命令：

# 检查进程是否存在
if ! pgrep -x "kafka" > /dev/null; then
    systemctl restart kafka
    echo "Kafka进程挂掉，已自动重启" | mail -s "故障通知" ops@example.com
fi
# 检查端口（适用于任何中间件）
if ! nc -z localhost 9092; then
    echo "端口9092不可达" | logger
fi

进阶技巧：分布式环境下的监控自动化

集成Prometheus指标体系

对于大型集群，建议用脚本输出Prometheus Exporter格式，

# 输出Kafka分区状态
echo "# HELP kafka_partition_leader_count leader分区数"
echo "# TYPE kafka_partition_leader_count gauge"
echo "kafka_partition_leader_count $(head /tmp/kafka_metrics.txt)"

然后采集到Prometheus+Grafana,实现可视化。

报警整合：钉钉/企业微信机器人

# alarm.sh 通用通知函数
send_alarm() {
    WEBHOOK="https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY"
    curl -s -X POST $WEBHOOK \
        -H "Content-Type: application/json" \
        -d "{\"msgtype\":\"text\",\"text\":{\"content\":\"$1\"}}"
}

定时任务部署

# crontab -e 每5分钟执行
*/5 * * * * /opt/scripts/monitor_kafka.sh >> /var/log/mq_monitor.log 2>&1

常见问题与解答（Q&A）

Q1：脚本直接执行正常，但crontab运行无效？
A：crontab环境变量缺失，解决方法：脚本内使用绝对路径，或加载用户profile，例如在脚本开头添加source /etc/profile。

Q2：RabbitMQ API返回401鉴权失败
A：确认guest账号是否有远程访问权限，生产环境建议创建专用监控用户并授权monitoring

Q3：如何避免重复报警？
A：在脚本中实现状态记录文件，第一次发现异常时记录/tmp/.last_alert_time,下次检查如果时间差小于300秒则不重复通知。