怎样用脚本监控MySQL主从状态？

wen 实用脚本 2026-07-21 168

本文目录导读：

怎样用脚本监控MySQL主从状态？

基础Shell脚本监控
带报警功能的完整监控脚本
Python脚本监控(更灵活)
设置定时监控(crontab)
监控脚本使用建议

我可以提供多种脚本来监控MySQL主从状态,以下是几个实用的监控脚本方案：

基础Shell脚本监控

#!/bin/bash
# MySQL连接信息
MYSQL_USER="monitor"
MYSQL_PASS="your_password"
MYSQL_HOST="localhost"
MYSQL_PORT=3306
echo "=== MySQL主从状态监控 ==="
echo "时间: $(date '+%Y-%m-%d %H:%M:%S')"
echo "------------------------"
# 检查主从状态
mysql -u${MYSQL_USER} -p${MYSQL_PASS} -h${MYSQL_HOST} -P${MYSQL_PORT} -e "
SHOW SLAVE STATUS\G
" 2>/dev/null | grep -E "Slave_IO_Running|Slave_SQL_Running|Seconds_Behind_Master" | while read line; do
    echo "$line"
done
# 检查状态并输出结果
SLAVE_STATUS=$(mysql -u${MYSQL_USER} -p${MYSQL_PASS} -h${MYSQL_HOST} -P${MYSQL_PORT} -e "SHOW SLAVE STATUS\G" 2>/dev/null)
IO_STATUS=$(echo "$SLAVE_STATUS" | grep "Slave_IO_Running:" | awk '{print $2}')
SQL_STATUS=$(echo "$SLAVE_STATUS" | grep "Slave_SQL_Running:" | awk '{print $2}')
DELAY=$(echo "$SLAVE_STATUS" | grep "Seconds_Behind_Master:" | awk '{print $2}')
echo ""
echo "=== 检查结果 ==="
if [ "$IO_STATUS" = "Yes" ] && [ "$SQL_STATUS" = "Yes" ]; then
    echo "✅ 主从同步正常"
    echo "延迟: ${DELAY}秒"
    if [ "$DELAY" -gt 10 ]; then
        echo "⚠️ 警告: 主从延迟超过10秒"
    fi
else
    echo "❌ 主从同步异常!"
    echo "IO线程状态: $IO_STATUS"
    echo "SQL线程状态: $SQL_STATUS"
fi

带报警功能的完整监控脚本

#!/bin/bash
# ========== 配置参数 ==========
MYSQL_USER="monitor"
MYSQL_PASS="your_password"
MYSQL_HOST="localhost"
MYSQL_PORT=3306
DELAY_THRESHOLD=10  # 延迟阈值(秒)
LOG_FILE="/var/log/mysql_replication_monitor.log"
ALERT_EMAIL="admin@example.com"
# ========== 函数定义 ==========
log_message() {
    echo "[$(date '+%Y-%m-%d %H:%M:%S')] $1" | tee -a $LOG_FILE
}
send_alert() {
    local subject="MySQL主从同步报警"
    local message="$1"
    echo "$message" | mail -s "$subject" $ALERT_EMAIL
    log_message "已发送报警邮件: $message"
}
check_replication() {
    # 执行SHOW SLAVE STATUS
    SLAVE_STATUS=$(mysql -u${MYSQL_USER} -p${MYSQL_PASS} -h${MYSQL_HOST} -P${MYSQL_PORT} -e "SHOW SLAVE STATUS\G" 2>/dev/null)
    if [ $? -ne 0 ]; then
        log_message "错误: 无法连接到MySQL数据库"
        send_alert "无法连接到MySQL数据库，请检查服务状态"
        return 1
    fi
    if [ -z "$SLAVE_STATUS" ]; then
        log_message "错误: 该服务器不是从库"
        send_alert "该服务器不是MySQL从库"
        return 1
    fi
    # 提取关键状态信息
    IO_RUNNING=$(echo "$SLAVE_STATUS" | grep "Slave_IO_Running:" | awk '{print $2}')
    SQL_RUNNING=$(echo "$SLAVE_STATUS" | grep "Slave_SQL_Running:" | awk '{print $2}')
    LAST_ERRNO=$(echo "$SLAVE_STATUS" | grep "Last_Errno:" | awk '{print $2}')
    LAST_ERROR=$(echo "$SLAVE_STATUS" | grep "Last_Error:" | awk '{print $2}')
    DELAY=$(echo "$SLAVE_STATUS" | grep "Seconds_Behind_Master:" | awk '{print $2}')
    MASTER_HOST=$(echo "$SLAVE_STATUS" | grep "Master_Host:" | awk '{print $2}')
    # 主从状态摘要
    SUMMARY="
=== MySQL主从状态摘要 ===
主库地址: $MASTER_HOST
IO线程状态: $IO_RUNNING
SQL线程状态: $SQL_RUNNING
同步延迟: ${DELAY:-NULL}秒
最后错误: $LAST_ERROR
错误编号: $LAST_ERRNO
"
    log_message "$SUMMARY"
    # 判断状态
    if [ "$IO_RUNNING" == "Yes" ] && [ "$SQL_RUNNING" == "Yes" ]; then
        log_message "✅ 主从同步正常"
        # 检查延迟
        if [ "$DELAY" != "NULL" ] && [ "$DELAY" -gt "$DELAY_THRESHOLD" ]; then
            log_message "⚠️ 主从延迟过高: ${DELAY}秒"
            send_alert "主从延迟超过阈值(${DELAY_THRESHOLD}秒)，当前延迟: ${DELAY}秒"
        fi
        return 0
    else
        log_message "❌ 主从同步异常!"
        send_alert "主从同步异常! IO线程: $IO_RUNNING, SQL线程: $SQL_RUNNING, 错误: $LAST_ERROR"
        return 1
    fi
}
# ========== 主程序 ==========
log_message "开始MySQL主从状态检查"
# 检查MySQL是否运行
if ! pgrep -x mysqld > /dev/null; then
    log_message "错误: MySQL服务未运行"
    send_alert "MySQL服务未运行"
    exit 1
fi
# 执行检查
check_replication
# 输出JSON格式状态(方便其他程序调用)
echo ""
echo "JSON输出:"
echo "{\"status\": \"$?\", \"io_running\": \"$IO_RUNNING\", \"sql_running\": \"$SQL_RUNNING\", \"delay\": \"$DELAY\"}"

Python脚本监控(更灵活)

#!/usr/bin/env python3
import pymysql
import time
import json
import logging
from datetime import datetime
# 配置
config = {
    'host': 'localhost',
    'user': 'monitor',
    'password': 'your_password',
    'port': 3306,
    'database': 'mysql'
}
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('/var/log/mysql_replication_monitor.log'),
        logging.StreamHandler()
    ]
)
class MySQLReplicationMonitor:
    def __init__(self, config):
        self.config = config
        self.connection = None
    def connect(self):
        """连接到MySQL"""
        try:
            self.connection = pymysql.connect(**self.config)
            logging.info("MySQL连接成功")
            return True
        except Exception as e:
            logging.error(f"MySQL连接失败: {e}")
            return False
    def check_replication(self):
        """检查主从状态"""
        if not self.connection:
            if not self.connect():
                return False
        try:
            with self.connection.cursor() as cursor:
                cursor.execute("SHOW SLAVE STATUS")
                result = cursor.fetchone()
                if not result:
                    logging.warning("该服务器不是从库")
                    return False
                # 获取字段名
                cursor.execute("SHOW SLAVE STATUS")
                columns = [desc[0] for desc in cursor.description]
                # 创建字典
                status = dict(zip(columns, result))
                # 提取关键信息
                slave_info = {
                    'timestamp': datetime.now().isoformat(),
                    'master_host': status.get('Master_Host'),
                    'master_port': status.get('Master_Port'),
                    'io_running': status.get('Slave_IO_Running'),
                    'sql_running': status.get('Slave_SQL_Running'),
                    'seconds_behind_master': status.get('Seconds_Behind_Master'),
                    'last_error': status.get('Last_Error'),
                    'last_errno': status.get('Last_Errno'),
                    'read_master_log_pos': status.get('Read_Master_Log_Pos'),
                    'exec_master_log_pos': status.get('Exec_Master_Log_Pos'),
                    'relay_log_space': status.get('Relay_Log_Space'),
                    'last_io_error': status.get('Last_IO_Error'),
                    'last_sql_error': status.get('Last_SQL_Error')
                }
                # 检查状态
                io_status = slave_info['io_running'] == 'Yes'
                sql_status = slave_info['sql_running'] == 'Yes'
                delay = slave_info['seconds_behind_master']
                logging.info(f"主从状态检查完成")
                logging.info(f"IO线程: {'✅ 正常' if io_status else '❌ 异常'}")
                logging.info(f"SQL线程: {'✅ 正常' if sql_status else '❌ 异常'}")
                logging.info(f"延迟: {delay}秒")
                # 输出JSON
                print(json.dumps(slave_info, indent=2, default=str))
                return io_status and sql_status
        except Exception as e:
            logging.error(f"检查主从状态失败: {e}")
            return False
    def close(self):
        """关闭连接"""
        if self.connection:
            self.connection.close()
            logging.info("MySQL连接已关闭")
def main():
    monitor = MySQLReplicationMonitor(config)
    try:
        # 执行检查
        status = monitor.check_replication()
        if status:
            logging.info("主从同步正常")
        else:
            logging.warning("主从同步异常")
    finally:
        monitor.close()
if __name__ == "__main__":
    main()

设置定时监控(crontab)

# 编辑crontab
crontab -e
# 添加以下内容(每5分钟检查一次)
*/5 * * * * /path/to/mysql_replication_monitor.sh
# 或者使用系统service
# /etc/systemd/system/mysql-replication-monitor.service

监控脚本使用建议

创建监控用户

-- 在主库和从库创建监控用户
CREATE USER 'monitor'@'localhost' IDENTIFIED BY 'your_password';
GRANT REPLICATION CLIENT, PROCESS ON *.* TO 'monitor'@'localhost';
FLUSH PRIVILEGES;

安装依赖

# Python脚本需要安装pymysql
pip install pymysql
# 发送邮件需要配置mail或使用其他通知方式
yum install mailx -y

关键监控指标说明

Slave_IO_Running: IO线程状态(Yes/No)
Slave_SQL_Running: SQL线程状态(Yes/No)
Seconds_Behind_Master: 延迟时间(秒)
Last_Error: 最后错误信息
Last_IO_Error: IO线程错误
Last_SQL_Error: SQL线程错误
Master_Log_Pos: 主库日志位置
Read_Master_Log_Pos: 从库读取位置
Relay_Log_Space: 中继日志空间

报警阈值建议

延迟 > 10秒：警告
延迟 > 60秒：严重告警
IO/SQL线程为No：紧急告警

选择适合你环境的脚本,并根据需要调整参数即可，建议使用Python脚本配合监控系统(如Prometheus、Zabbix)使用。

本文地址： https://www.bmjup.com/post/26174.html