定时检查数据库连接的脚本

本文目录导读：

1. PostgreSQL 检查脚本
2. MySQL 检查脚本
3. 通用数据库检查脚本（带告警功能）
4. Python 版本（更灵活）
5. crontab 定时配置
6. systemd timer 方式
7. 使用建议

PostgreSQL 检查脚本

#!/bin/bash
# pg_check.sh - 定时检查PostgreSQL数据库连接
# 数据库连接参数
DB_HOST="localhost"
DB_PORT="5432"
DB_NAME="testdb"
DB_USER="postgres"
DB_PASSWORD="password"
LOG_FILE="/var/log/db_check.log"
# 检查函数
check_pg_connection() {
    PGPASSWORD=$DB_PASSWORD psql -h $DB_HOST -p $DB_PORT -U $DB_USER -d $DB_NAME -c "SELECT 1" > /dev/null 2>&1
    if [ $? -eq 0 ]; then
        echo "$(date '+%Y-%m-%d %H:%M:%S') - PostgreSQL连接正常" >> $LOG_FILE
        return 0
    else
        echo "$(date '+%Y-%m-%d %H:%M:%S') - PostgreSQL连接失败" >> $LOG_FILE
        return 1
    fi
}
# 执行检查
check_pg_connection

MySQL 检查脚本

#!/bin/bash
# mysql_check.sh - 定时检查MySQL数据库连接
# 数据库连接参数
DB_HOST="localhost"
DB_PORT="3306"
DB_USER="root"
DB_PASSWORD="password"
DB_NAME="testdb"
LOG_FILE="/var/log/db_check.log"
# 检查函数
check_mysql_connection() {
    mysql -h $DB_HOST -P $DB_PORT -u $DB_USER -p$DB_PASSWORD -e "SELECT 1" $DB_NAME > /dev/null 2>&1
    if [ $? -eq 0 ]; then
        echo "$(date '+%Y-%m-%d %H:%M:%S') - MySQL连接正常" >> $LOG_FILE
        return 0
    else
        echo "$(date '+%Y-%m-%d %H:%M:%S') - MySQL连接失败" >> $LOG_FILE
        return 1
    fi
}
# 执行检查
check_mysql_connection

通用数据库检查脚本（带告警功能）

#!/bin/bash
# db_check_and_alert.sh - 带告警功能的数据库连接检查
# 配置
DB_TYPE="postgresql"  # mysql, postgresql, oracle
DB_HOST="localhost"
DB_PORT="5432"
DB_NAME="testdb"
DB_USER="admin"
DB_PASSWORD="password"
LOG_FILE="/var/log/db_health.log"
ALERT_EMAIL="admin@example.com"
MAX_RETRIES=3
RETRY_INTERVAL=5
# 日志函数
log_message() {
    local level=$1
    local message=$2
    echo "$(date '+%Y-%m-%d %H:%M:%S') [$level] $message" >> $LOG_FILE
}
# 发送告警邮件
send_alert() {
    local subject="数据库连接异常告警"
    local body="时间: $(date '+%Y-%m-%d %H:%M:%S')\n数据库: $DB_TYPE://$DB_HOST:$DB_PORT/$DB_NAME\n状态: 连接失败"
    echo -e "$body" | mail -s "$subject" $ALERT_EMAIL
}
# 检查数据库连接
check_connection() {
    local retry_count=0
    while [ $retry_count -lt $MAX_RETRIES ]; do
        case $DB_TYPE in
            "postgresql")
                PGPASSWORD=$DB_PASSWORD psql -h $DB_HOST -p $DB_PORT -U $DB_USER -d $DB_NAME -c "SELECT 1" > /dev/null 2>&1
                ;;
            "mysql")
                mysql -h $DB_HOST -P $DB_PORT -u $DB_USER -p$DB_PASSWORD -e "SELECT 1" $DB_NAME > /dev/null 2>&1
                ;;
            "oracle")
                sqlplus -S $DB_USER/$DB_PASSWORD@//$DB_HOST:$DB_PORT/$DB_NAME <<< "SELECT 1 FROM DUAL;" > /dev/null 2>&1
                ;;
            *)
                log_message "ERROR" "不支持的数据库类型: $DB_TYPE"
                return 2
                ;;
        esac
        if [ $? -eq 0 ]; then
            log_message "INFO" "连接成功"
            return 0
        fi
        ((retry_count++))
        log_message "WARN" "第 $retry_count 次连接失败，等待重试..."
        sleep $RETRY_INTERVAL
    done
    log_message "ERROR" "连接失败（重试$MAX_RETRIES次后）"
    return 1
}
# 主逻辑
check_connection
if [ $? -eq 1 ]; then
    send_alert
    log_message "ALERT" "已发送告警通知"
fi

Python 版本（更灵活）

#!/usr/bin/env python3
# db_check.py - Python版本的数据库连接检查
import psycopg2  # PostgreSQL
# import mysql.connector  # MySQL
import time
import logging
import smtplib
from email.mime.text import MIMEText
from datetime import datetime
# 配置
DB_CONFIG = {
    'host': 'localhost',
    'port': 5432,
    'database': 'testdb',
    'user': 'admin',
    'password': 'password'
}
ALERT_EMAIL = {
    'smtp_server': 'smtp.example.com',
    'smtp_port': 587,
    'sender': 'monitor@example.com',
    'password': 'email_password',
    'recipients': ['admin@example.com']
}
# 日志配置
logging.basicConfig(
    filename='/var/log/db_check.log',
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s'
)
def check_postgresql():
    """检查PostgreSQL连接"""
    try:
        conn = psycopg2.connect(**DB_CONFIG)
        cursor = conn.cursor()
        cursor.execute("SELECT 1")
        conn.close()
        return True
    except Exception as e:
        logging.error(f"PostgreSQL连接失败: {str(e)}")
        return False
def check_mysql():
    """检查MySQL连接"""
    # 需要安装mysql-connector-python
    # import mysql.connector
    # try:
    #     conn = mysql.connector.connect(**DB_CONFIG)
    #     cursor = conn.cursor()
    #     cursor.execute("SELECT 1")
    #     conn.close()
    #     return True
    # except Exception as e:
    #     logging.error(f"MySQL连接失败: {str(e)}")
    #     return False
    pass
def send_alert_email():
    """发送告警邮件"""
    try:
        msg = MIMEText(f"数据库连接异常\n时间: {datetime.now()}\n数据库: {DB_CONFIG}")
        msg['Subject'] = '数据库连接告警'
        msg['From'] = ALERT_EMAIL['sender']
        msg['To'] = ', '.join(ALERT_EMAIL['recipients'])
        with smtplib.SMTP(ALERT_EMAIL['smtp_server'], ALERT_EMAIL['smtp_port']) as server:
            server.starttls()
            server.login(ALERT_EMAIL['sender'], ALERT_EMAIL['password'])
            server.send_message(msg)
        logging.info("告警邮件已发送")
    except Exception as e:
        logging.error(f"发送邮件失败: {str(e)}")
def main():
    """主函数"""
    max_retries = 3
    retry_interval = 5
    for attempt in range(1, max_retries + 1):
        if check_postgresql():
            logging.info("数据库连接正常")
            return True
        else:
            logging.warning(f"第{attempt}次连接失败")
            if attempt < max_retries:
                time.sleep(retry_interval)
    logging.error("数据库连接失败（重试3次后）")
    send_alert_email()
    return False
if __name__ == "__main__":
    main()

crontab 定时配置

# 编辑crontab
crontab -e
# 添加定时任务示例：
# 每5分钟检查一次
*/5 * * * * /path/to/db_check.sh
# 每小时检查一次
0 * * * * /path/to/db_check_and_alert.sh
# 每天8点和18点检查
0 8,18 * * * /path/to/python_script.py
# 每30分钟检查一次并记录日志
*/30 * * * * /path/to/db_check.sh >> /var/log/cron_db.log 2>&1

systemd timer 方式

创建服务文件 /etc/systemd/system/db-check.service:

[Unit]
Description=Database Connection Check Service
[Service]
Type=oneshot
ExecStart=/path/to/db_check.sh
User=root
Group=root

创建timer文件 /etc/systemd/system/db-check.timer:

[Unit]
Description=Run db-check every 5 minutes
[Timer]
OnCalendar=*:0/5
Persistent=true
[Install]
WantedBy=timers.target

启用并启动：

systemctl daemon-reload
systemctl enable db-check.timer
systemctl start db-check.timer

使用建议

1. 安全性：将密码存储在环境变量或密钥管理服务中
2. 日志轮转：配置logrotate避免日志文件过大
3. 告警级别：根据连接失败次数设置不同告警级别
4. 性能影响：设置合理的检查间隔（建议>30秒）
5. 错误处理：添加超时机制防止脚本卡住
6. 监控集成：可集成到Prometheus/Grafana等监控系统

请根据实际环境选择合适的脚本并调整配置参数。