Python脚本如何监控数据同步延迟状态

wen python案例 2026-06-26 236

本文目录导读：

MySQL主从复制延迟监控
Redis主从延迟监控
通用数据同步延迟监控
带告警功能的监控器
使用方法

我来为你提供几个监控数据同步延迟的Python脚本方案：

MySQL主从复制延迟监控

import pymysql
import time
from datetime import datetime
import logging
# 配置日志
logging.basicConfig(level=logging.INFO, 
                   format='%(asctime)s - %(levelname)s - %(message)s')
class MySQLReplicationMonitor:
    def __init__(self, host, user, password, port=3306):
        self.config = {
            'host': host,
            'user': user,
            'password': password,
            'port': port
        }
    def get_replication_delay(self):
        """获取主从复制延迟"""
        try:
            conn = pymysql.connect(**self.config)
            cursor = conn.cursor()
            # 检查从库状态
            cursor.execute("SHOW SLAVE STATUS")
            slave_status = cursor.fetchone()
            if slave_status:
                # Seconds_Behind_Master 是延迟秒数
                seconds_behind = slave_status[32]  # 第33个字段
                return {
                    'delay_seconds': seconds_behind or 0,
                    'slave_io_running': slave_status[10],  # Slave_IO_Running
                    'slave_sql_running': slave_status[11], # Slave_SQL_Running
                    'last_error': slave_status[19]         # Last_Error
                }
            else:
                return None
        except Exception as e:
            logging.error(f"获取复制延迟失败: {e}")
            return None
        finally:
            if 'conn' in locals():
                conn.close()
    def monitor(self, interval=5, threshold=60):
        """持续监控复制延迟"""
        logging.info("开始监控MySQL主从复制延迟...")
        while True:
            status = self.get_replication_delay()
            if status:
                delay = status['delay_seconds']
                if delay > threshold:
                    logging.warning(f"复制延迟过高: {delay}秒")
                    if status['slave_io_running'] != 'Yes':
                        logging.error("Slave IO线程停止运行")
                    if status['slave_sql_running'] != 'Yes':
                        logging.error("Slave SQL线程停止运行")
                else:
                    logging.info(f"复制延迟正常: {delay}秒")
            else:
                logging.warning("无法获取复制状态")
            time.sleep(interval)
# 使用示例
monitor = MySQLReplicationMonitor(
    host='10.0.0.2',  # 从库地址
    user='repl_user',
    password='password'
)
monitor.monitor(interval=10, threshold=30)

Redis主从延迟监控

import redis
import time
import logging
logging.basicConfig(level=logging.INFO, 
                   format='%(asctime)s - %(levelname)s - %(message)s')
class RedisReplicationMonitor:
    def __init__(self, host='localhost', port=6379, password=None):
        self.redis_client = redis.Redis(
            host=host,
            port=port,
            password=password,
            decode_responses=True
        )
    def get_replication_delay(self):
        """获取Redis主从复制延迟"""
        try:
            # 从库使用 info replication 命令
            info = self.redis_client.info('replication')
            if info['role'] == 'slave':
                # master_last_io_seconds_ago 表示上次与主库通信的秒数
                delay = info.get('master_last_io_seconds_ago', 0)
                # 计算复制积压量
                master_repl_offset = info.get('master_repl_offset', 0)
                slave_repl_offset = info.get('slave_repl_offset', 0)
                replication_gap = master_repl_offset - slave_repl_offset
                return {
                    'role': 'slave',
                    'delay_seconds': delay,
                    'master_link_status': info.get('master_link_status', 'down'),
                    'replication_gap': replication_gap,
                    'master_host': info.get('master_host', 'unknown')
                }
            else:
                return {'role': 'master', 'connected_slaves': info.get('connected_slaves', 0)}
        except Exception as e:
            logging.error(f"获取Redis复制状态失败: {e}")
            return None
    def monitor(self, interval=5, threshold=10):
        """持续监控复制延迟"""
        logging.info("开始监控Redis主从复制延迟...")
        while True:
            status = self.get_replication_delay()
            if status and status.get('role') == 'slave':
                delay = status['delay_seconds']
                if delay > threshold:
                    logging.warning(f"Redis复制延迟: {delay}秒, "
                                  f"复制gap: {status['replication_gap']}字节")
                    if status['master_link_status'] != 'up':
                        logging.error("Redis主从连接断开")
                else:
                    logging.info(f"Redis复制正常: {delay}秒")
            time.sleep(interval)
# 使用示例
monitor = RedisReplicationMonitor(
    host='10.0.0.3',  # 从库地址
    port=6379
)
monitor.monitor(interval=10, threshold=5)

通用数据同步延迟监控

import requests
import time
import json
from datetime import datetime
import hashlib
class DataSyncMonitor:
    def __init__(self, source_api, target_api):
        self.source_api = source_api
        self.target_api = target_api
    def get_latest_record(self, api_url, table_name):
        """获取指定表的最新记录"""
        try:
            response = requests.get(
                f"{api_url}/latest/{table_name}",
                timeout=5
            )
            if response.status_code == 200:
                return response.json()
            return None
        except Exception as e:
            print(f"获取记录失败: {e}")
            return None
    def calculate_delay(self, source_time, target_time):
        """计算同步延迟时间"""
        if source_time and target_time:
            s_time = datetime.fromisoformat(source_time.replace('Z', '+00:00'))
            t_time = datetime.fromisoformat(target_time.replace('Z', '+00:00'))
            delay = (s_time - t_time).total_seconds()
            return max(0, delay)
        return -1
    def check_data_integrity(self, source_data, target_data):
        """检查数据完整性"""
        if not source_data or not target_data:
            return False, "数据不存在"
        source_hash = hashlib.md5(
            json.dumps(source_data, sort_keys=True).encode()
        ).hexdigest()
        target_hash = hashlib.md5(
            json.dumps(target_data, sort_keys=True).encode()
        ).hexdigest()
        if source_hash != target_hash:
            return False, "数据不一致"
        return True, "数据一致"
    def monitor_table(self, table_name, interval=30):
        """监控指定表的同步状态"""
        while True:
            source_record = self.get_latest_record(self.source_api, table_name)
            target_record = self.get_latest_record(self.target_api, table_name)
            if source_record and target_record:
                # 计算时间延迟
                delay = self.calculate_delay(
                    source_record.get('update_time'),
                    target_record.get('update_time')
                )
                # 检查数据完整性
                is_integrity, integrity_msg = self.check_data_integrity(
                    source_record, target_record
                )
                print(f"[{datetime.now()}] 表: {table_name}")
                print(f"  数据延迟: {delay}秒")
                print(f"  数据完整性: {integrity_msg}")
                if delay > 60:
                    print(f"  ⚠️ 警告: 同步延迟超过1分钟!")
            else:
                print(f"[{datetime.now()}] 无法获取源或目标数据")
            time.sleep(interval)
# 使用示例
monitor = DataSyncMonitor(
    source_api='http://source-system:8000/api',
    target_api='http://target-system:8000/api'
)
monitor.monitor_table('orders', interval=60)

带告警功能的监控器

import smtplib
from email.mime.text import MIMEText
import requests
import time
class AlertSyncMonitor:
    def __init__(self, alert_config):
        self.alert_config = alert_config
        self.alert_thresholds = {
            'critical': 300,  # 5分钟
            'warning': 60     # 1分钟
        }
    def send_email_alert(self, subject, message):
        """发送邮件告警"""
        msg = MIMEText(message)
        msg['Subject'] = subject
        msg['From'] = self.alert_config['email_from']
        msg['To'] = self.alert_config['email_to']
        try:
            with smtplib.SMTP(self.alert_config['smtp_server'], 
                             self.alert_config['smtp_port']) as server:
                server.starttls()
                server.login(
                    self.alert_config['email_user'],
                    self.alert_config['email_password']
                )
                server.send_message(msg)
        except Exception as e:
            print(f"发送邮件告警失败: {e}")
    def send_dingtalk_alert(self, message):
        """发送钉钉告警"""
        webhook_url = self.alert_config.get('dingtalk_webhook')
        if webhook_url:
            data = {
                "msgtype": "text",
                "text": {"content": message}
            }
            requests.post(webhook_url, json=data)
    def check_and_alert(self, delay, component_name):
        """检查延迟并发送告警"""
        if delay >= self.alert_thresholds['critical']:
            message = f"严重告警: {component_name}同步延迟 {delay}秒"
            self.send_email_alert(f"[CRITICAL] {component_name}延迟告警", message)
            self.send_dingtalk_alert(message)
            print(message)
        elif delay >= self.alert_thresholds['warning']:
            message = f"警告: {component_name}同步延迟 {delay}秒"
            print(message)
# 使用示例
alert_config = {
    'email_from': 'monitor@example.com',
    'email_to': 'admin@example.com',
    'smtp_server': 'smtp.example.com',
    'smtp_port': 587,
    'email_user': 'monitor@example.com',
    'email_password': 'password',
    'dingtalk_webhook': 'https://oapi.dingtalk.com/robot/send?access_token=xxxx'
}
monitor = AlertSyncMonitor(alert_config)
monitor.check_and_alert(120, "MySQL复制")

使用方法

安装依赖：
```
pip install pymysql redis requests
```
选择合适的监控方案：
- MySQL从库：使用方案1
- Redis主从：使用方案2
- 自定义数据同步：使用方案3
- 需要告警功能：整合方案4

作为后台服务运行：

nohup python mysql_monitor.py > monitor.log 2>&1 &

这些脚本可以根据你的具体需求进行修改和扩展,比如添加更多的监控指标、集成不同的告警渠道等。

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