本文目录导读：
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
使用建议
- 安全性：将密码存储在环境变量或密钥管理服务中
- 日志轮转：配置logrotate避免日志文件过大
- 告警级别：根据连接失败次数设置不同告警级别
- 性能影响：设置合理的检查间隔（建议>30秒）
- 错误处理：添加超时机制防止脚本卡住
- 监控集成：可集成到Prometheus/Grafana等监控系统
请根据实际环境选择合适的脚本并调整配置参数。
