监控消息队列积压的脚本如何写

wen 实用脚本 2026-07-10 1

本文目录导读：

监控消息队列积压的脚本如何写

RabbitMQ 监控脚本
Kafka 监控脚本
Python版本（支持多种MQ）
定时任务配置
Prometheus + Grafana 集成

我来提供一个通用的消息队列积压监控脚本,以常见的消息队列系统为例：

RabbitMQ 监控脚本

#!/bin/bash
# rabbitmq_monitor.sh - RabbitMQ队列积压监控
# 配置参数
RABBITMQ_HOST="localhost"
RABBITMQ_PORT=15672
RABBITMQ_USER="guest"
RABBITMQ_PASS="guest"
THRESHOLD=1000  # 告警阈值
WEBHOOK_URL="https://hooks.slack.com/services/xxx"  # 可选告警渠道
# 获取队列积压情况
get_queue_info() {
    curl -s -u "$RABBITMQ_USER:$RABBITMQ_PASS" \
        "http://$RABBITMQ_HOST:$RABBITMQ_PORT/api/queues" | \
        jq -r '.[] | "\(.name),\(.messages_ready),\(.messages_unacknowledged)"'
}
# 检查并告警
check_and_alert() {
    local queue_name=$1
    local total_messages=$2
    if [ "$total_messages" -gt "$THRESHOLD" ]; then
        echo "[ALERT] 队列 $queue_name 积压 ${total_messages} 条消息，超过阈值 ${THRESHOLD}"
        # 发送告警到Slack
        curl -X POST -H "Content-Type: application/json" \
            -d "{\"text\":\"队列 $queue_name 积压 ${total_messages} 条消息\"}" \
            "$WEBHOOK_URL"
    fi
}
# 主逻辑
echo "=== RabbitMQ 队列监控 ==="
echo "时间: $(date '+%Y-%m-%d %H:%M:%S')"
echo "阈值: $THRESHOLD"
echo "========================="
while read -r line; do
    queue_name=$(echo "$line" | cut -d',' -f1)
    ready=$(echo "$line" | cut -d',' -f2)
    unacked=$(echo "$line" | cut -d',' -f3)
    total=$((ready + unacked))
    check_and_alert "$queue_name" "$total"
    echo "队列: $queue_name | 待消费: $ready | 未确认: $unacked | 总计: $total"
done < <(get_queue_info)

Kafka 监控脚本

#!/bin/bash
# kafka_monitor.sh - Kafka消费者滞后监控
# 配置参数
KAFKA_HOME="/opt/kafka"
BOOTSTRAP_SERVERS="localhost:9092"
GROUP_ID="my-consumer-group"
THRESHOLD=5000
# 获取消费者组信息
get_consumer_lag() {
    $KAFKA_HOME/bin/kafka-consumer-groups.sh \
        --bootstrap-server $BOOTSTRAP_SERVERS \
        --group $GROUP_ID \
        --describe 2>/dev/null | tail -n +2
}
# 检查单个分区的滞后
check_partition_lag() {
    local topic=$1
    local partition=$2
    local lag=$3
    if [ "$lag" -gt "$THRESHOLD" ]; then
        echo "[WARNING] Topic: $topic Partition: $partition 滞后 $lag 条消息"
    fi
}
# 主逻辑
echo "=== Kafka 消费者滞后监控 ==="
echo "消费组: $GROUP_ID"
echo "时间: $(date '+%Y-%m-%d %H:%M:%S')"
echo "=============================="
while IFS= read -r line; do
    topic=$(echo "$line" | awk '{print $1}')
    partition=$(echo "$line" | awk '{print $2}')
    current_offset=$(echo "$line" | awk '{print $3}')
    log_end_offset=$(echo "$line" | awk '{print $4}')
    lag=$(echo "$line" | awk '{print $5}')  # 改为第5列
    if [ -n "$lag" ] && [ "$lag" != "LAG" ]; then
        check_partition_lag "$topic" "$partition" "$lag"
        echo "Topic: $topic Partition: $partition | 滞后: $lag"
    fi
done < <(get_consumer_lag)

Python版本（支持多种MQ）

#!/usr/bin/env python3
# mq_monitor.py - 通用消息队列监控
import subprocess
import json
import requests
import time
from datetime import datetime
import smtplib
from email.mime.text import MIMEText
class MQMonitor:
    def __init__(self, mq_type='rabbitmq', config=None):
        self.mq_type = mq_type
        self.config = config or {}
        self.threshold = config.get('threshold', 1000)
        self.alert_methods = config.get('alert_methods', ['stdout'])
    def check_rabbitmq(self):
        """监控RabbitMQ队列积压"""
        try:
            url = f"http://{self.config['host']}:{self.config['port']}/api/queues"
            response = requests.get(
                url,
                auth=(self.config['user'], self.config['password']),
                timeout=10
            )
            queues = response.json()
            alerts = []
            for queue in queues:
                name = queue['name']
                ready = queue.get('messages_ready', 0)
                unacked = queue.get('messages_unacknowledged', 0)
                total = ready + unacked
                queue_info = {
                    'name': name,
                    'ready': ready,
                    'unacked': unacked,
                    'total': total
                }
                if total > self.threshold:
                    alerts.append(queue_info)
            return queues, alerts
        except Exception as e:
            print(f"RabbitMQ监控失败: {e}")
            return [], []
    def check_kafka(self, group_id):
        """监控Kafka消费者滞后"""
        try:
            cmd = [
                f"{self.config['kafka_home']}/bin/kafka-consumer-groups.sh",
                "--bootstrap-server", self.config['bootstrap_servers'],
                "--group", group_id,
                "--describe"
            ]
            result = subprocess.run(cmd, capture_output=True, text=True)
            partitions = []
            for line in result.stdout.split('\n')[1:]:  # 跳过标题行
                if line.strip():
                    parts = line.split()
                    if len(parts) >= 5:
                        partition = {
                            'topic': parts[0],
                            'partition': parts[1],
                            'current_offset': int(parts[2]),
                            'log_end_offset': int(parts[3]),
                            'lag': int(parts[4]) if parts[4] != '-' else 0
                        }
                        partitions.append(partition)
            alerts = [p for p in partitions if p['lag'] > self.threshold]
            return partitions, alerts
        except Exception as e:
            print(f"Kafka监控失败: {e}")
            return [], []
    def check_redis(self):
        """监控Redis队列长度"""
        import redis
        try:
            r = redis.Redis(
                host=self.config.get('redis_host', 'localhost'),
                port=self.config.get('redis_port', 6379)
            )
            queues = []
            alerts = []
            # 监控指定队列
            queue_names = self.config.get('queue_names', [])
            for queue_name in queue_names:
                length = r.llen(queue_name)
                queue_info = {
                    'name': queue_name,
                    'length': length
                }
                queues.append(queue_info)
                if length > self.threshold:
                    alerts.append(queue_info)
            return queues, alerts
        except Exception as e:
            print(f"Redis监控失败: {e}")
            return [], []
    def send_alert(self, alert_info):
        """发送告警"""
        timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        message = f"[{timestamp}] 队列告警: {json.dumps(alert_info)}"
        if 'stdout' in self.alert_methods:
            print(f"[ALERT] {message}")
        if 'slack' in self.alert_methods and 'webhook_url' in self.config:
            try:
                requests.post(
                    self.config['webhook_url'],
                    json={'text': message}
                )
            except Exception as e:
                print(f"Slack告警失败: {e}")
        if 'email' in self.alert_methods:
            self._send_email_alert(message)
    def _send_email_alert(self, message):
        """发送邮件告警"""
        try:
            msg = MIMEText(message)
            msg['Subject'] = 'MQ队列积压告警'
            msg['From'] = self.config['email_from']
            msg['To'] = self.config['email_to']
            with smtplib.SMTP(self.config['smtp_server'], self.config['smtp_port']) as server:
                server.login(self.config['smtp_user'], self.config['smtp_password'])
                server.send_message(msg)
        except Exception as e:
            print(f"邮件告警失败: {e}")
    def run_monitor(self, interval=60):
        """运行监控循环"""
        while True:
            print(f"\n=== 消息队列监控 [{datetime.now()}] ===")
            if self.mq_type == 'rabbitmq':
                queues, alerts = self.check_rabbitmq()
                for q in queues:
                    print(f"队列: {q['name']} | 积压: {q['total']}")
            elif self.mq_type == 'kafka':
                partitions, alerts = self.check_kafka(
                    self.config.get('group_id', 'default-group')
                )
                for p in partitions:
                    print(f"Topic: {p['topic']} Partition: {p['partition']} | 滞后: {p['lag']}")
            elif self.mq_type == 'redis':
                queues, alerts = self.check_redis()
                for q in queues:
                    print(f"队列: {q['name']} | 长度: {q['length']}")
            # 处理告警
            for alert in alerts:
                self.send_alert(alert)
            time.sleep(interval)
# 使用示例
if __name__ == "__main__":
    # RabbitMQ监控
    config = {
        'host': 'localhost',
        'port': 15672,
        'user': 'guest',
        'password': 'guest',
        'threshold': 1000,
        'alert_methods': ['stdout', 'slack'],
        'webhook_url': 'https://hooks.slack.com/services/xxx'
    }
    monitor = MQMonitor(mq_type='rabbitmq', config=config)
    monitor.run_monitor(interval=60)  # 每60秒检查一次

定时任务配置

# crontab 配置 - 每5分钟执行一次
*/5 * * * * /path/to/rabbitmq_monitor.sh >> /var/log/mq_monitor.log 2>&1
# 或者作为systemd服务
# /etc/systemd/system/mq-monitor.service
[Unit]
Description=MQ Queue Monitor
After=network.target
[Service]
Type=simple
User=monitor
ExecStart=/usr/bin/python3 /path/to/mq_monitor.py
Restart=always
RestartSec=10
[Install]
WantedBy=multi-user.target

Prometheus + Grafana 集成

# prometheus.yml 配置
scrape_configs:
  - job_name: 'rabbitmq'
    static_configs:
      - targets: ['localhost:15672']
    metrics_path: /metrics
    params:
      username: ['guest']
      password: ['guest']

这个脚本方案提供了：

多MQ系统支持（RabbitMQ、Kafka、Redis）
灵活的告警通知（终端、Slack、邮件）
可配置的阈值和检查间隔
完整的错误处理
适合集成到现有监控系统

你可以根据实际需求选择对应的脚本,并调整配置参数。