本文目录导读：
我来为你提供几个监控数据同步延迟的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 &
这些脚本可以根据你的具体需求进行修改和扩展,比如添加更多的监控指标、集成不同的告警渠道等。
