本文目录导读：
我来帮您编写一个完整的接口响应监控脚本,这里提供几种常用语言的实现方案：
Python 版本（最常用）
import requests
import time
import json
from datetime import datetime
import smtplib
from email.mime.text import MIMEText
import logging
import sys
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('api_monitor.log'),
logging.StreamHandler(sys.stdout)
]
)
class APIMonitor:
def __init__(self, config_file='config.json'):
with open(config_file, 'r') as f:
self.config = json.load(f)
self.api_url = self.config['api_url']
self.expected_status = self.config.get('expected_status', 200)
self.timeout = self.config.get('timeout', 10)
self.max_retries = self.config.get('max_retries', 3)
self.alert_threshold = self.config.get('alert_threshold', 5) # 连续失败次数
self.failure_count = 0
def check_api(self):
"""检查单个API接口"""
try:
start_time = time.time()
response = requests.get(
self.api_url,
timeout=self.timeout,
headers=self.config.get('headers', {})
)
response_time = time.time() - start_time
# 检查响应状态
if response.status_code == self.expected_status:
# 检查响应时间
max_response_time = self.config.get('max_response_time', 5)
if response_time > max_response_time:
logging.warning(f"响应时间过长: {response_time:.2f}s")
self.alert(f"API响应时间过长: {self.api_url}，耗时: {response_time:.2f}s")
# 检查响应内容
if 'response_validation' in self.config:
validation_result = self.validate_response(response)
if not validation_result:
logging.error("响应内容验证失败")
self.alert(f"API响应内容异常: {self.api_url}")
self.failure_count = 0
logging.info(f"API正常 - 状态码: {response.status_code}, 耗时: {response_time:.2f}s")
return True
else:
self.failure_count += 1
logging.error(f"状态码异常: {response.status_code}")
self.alert(f"API状态码异常: {self.api_url}，状态码: {response.status_code}")
return False
except requests.exceptions.Timeout:
self.failure_count += 1
logging.error(f"请求超时: {self.timeout}s")
self.alert(f"API超时: {self.api_url}")
return False
except requests.exceptions.ConnectionError:
self.failure_count += 1
logging.error("连接失败")
self.alert(f"API连接失败: {self.api_url}")
return False
except Exception as e:
self.failure_count += 1
logging.error(f"未知错误: {str(e)}")
self.alert(f"API未知错误: {self.api_url} - {str(e)}")
return False
def validate_response(self, response):
"""验证响应内容"""
try:
data = response.json()
validations = self.config['response_validation']
for key, expected_value in validations.items():
actual_value = data.get(key)
if actual_value != expected_value:
logging.error(f"验证失败 - {key}: 期望={expected_value}, 实际={actual_value}")
return False
return True
except:
return False
def alert(self, message):
"""发送告警"""
if self.failure_count >= self.alert_threshold:
self.send_email_alert(message)
self.send_webhook_alert(message)
def send_email_alert(self, message):
"""发送邮件告警"""
if not self.config.get('email_config'):
return
email_cfg = self.config['email_config']
msg = MIMEText(message, 'plain', 'utf-8')
msg['Subject'] = f"API监控告警 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
msg['From'] = email_cfg['from']
msg['To'] = email_cfg['to']
try:
server = smtplib.SMTP(email_cfg['smtp_server'], email_cfg['smtp_port'])
server.starttls()
server.login(email_cfg['username'], email_cfg['password'])
server.send_message(msg)
server.quit()
logging.info("邮件告警已发送")
except Exception as e:
logging.error(f"发送邮件失败: {str(e)}")
def send_webhook_alert(self, message):
"""发送Webhook告警（如企业微信、钉钉等）"""
if not self.config.get('webhook_url'):
return
try:
webhook_data = {
"msgtype": "text",
"text": {
"content": message
}
}
requests.post(self.config['webhook_url'], json=webhook_data)
logging.info("Webhook告警已发送")
except Exception as e:
logging.error(f"发送Webhook失败: {str(e)}")
def run(self, interval=60):
"""运行监控"""
logging.info(f"开始监控API: {self.api_url}")
while True:
self.check_api()
time.sleep(interval)
# 配置文件示例 (config.json)
CONFIG_EXAMPLE = {
"api_url": "https://api.example.com/health",
"expected_status": 200,
"timeout": 10,
"max_retries": 3,
"max_response_time": 5,
"alert_threshold": 3,
"headers": {
"Authorization": "Bearer YOUR_TOKEN",
"Content-Type": "application/json"
},
"response_validation": {
"status": "ok",
"code": 0
},
"email_config": {
"from": "monitor@example.com",
"to": "admin@example.com",
"smtp_server": "smtp.gmail.com",
"smtp_port": 587,
"username": "monitor@example.com",
"password": "YOUR_PASSWORD"
},
"webhook_url": "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY"
}
if __name__ == "__main__":
monitor = APIMonitor()
monitor.run(interval=60) # 每60秒检查一次
Shell 脚本版本（轻量级）
#!/bin/bash
# API监控配置
API_URL="https://api.example.com/health"
EXPECTED_STATUS=200
TIMEOUT=10
LOG_FILE="/var/log/api_monitor.log"
ERROR_COUNT=0
MAX_ERRORS=5
ALERT_EMAIL="admin@example.com"
# 日志函数
log_message() {
echo "$(date '+%Y-%m-%d %H:%M:%S') - $1" >> $LOG_FILE
echo "$(date '+%Y-%m-%d %H:%M:%S') - $1"
}
# 发送告警
send_alert() {
local message="$1"
echo "$message" | mail -s "API监控告警" $ALERT_EMAIL
# 也可以发送到企业微信机器人
# curl -X POST -H "Content-Type: application/json" -d "{\"msgtype\":\"text\",\"text\":{\"content\":\"$message\"}}" $WEBHOOK_URL
}
# 检查API
check_api() {
local start_time=$(date +%s%N)
local response=$(curl -s -o /dev/null -w "%{http_code}" --max-time $TIMEOUT $API_URL)
local end_time=$(date +%s%N)
local response_time=$((($end_time - $start_time) / 1000000)) # 转换为毫秒
if [ "$response" -eq "$EXPECTED_STATUS" ]; then
log_message "API正常 - 状态码: $response, 耗时: ${response_time}ms"
ERROR_COUNT=0
# 检查响应时间
if [ $response_time -gt 5000 ]; then # 5秒
log_message "警告: 响应时间过长"
send_alert "API响应时间过长: ${response_time}ms"
fi
else
ERROR_COUNT=$((ERROR_COUNT + 1))
log_message "ERROR - 状态码: $response, 耗时: ${response_time}ms"
if [ $ERROR_COUNT -ge $MAX_ERRORS ]; then
send_alert "API连续失败 ${ERROR_COUNT} 次, 当前状态码: $response"
fi
fi
}
# 主循环
log_message "开始监控API: $API_URL"
while true; do
check_api
sleep 60 # 每60秒检查一次
done
Node.js 版本
const axios = require('axios');
const fs = require('fs');
const nodemailer = require('nodemailer');
const { CronJob } = require('cron');
class APIMonitor {
constructor(config) {
this.config = config;
this.failureCount = 0;
this.setupLogger();
}
setupLogger() {
this.logStream = fs.createWriteStream('api_monitor.log', { flags: 'a' });
}
log(message, level = 'INFO') {
const timestamp = new Date().toISOString();
const logMessage = `[${timestamp}] [${level}] ${message}\n`;
this.logStream.write(logMessage);
console.log(logMessage);
}
async checkAPI() {
const startTime = Date.now();
try {
const response = await axios({
method: this.config.method || 'GET',
url: this.config.url,
headers: this.config.headers || {},
timeout: this.config.timeout || 10000,
data: this.config.data
});
const responseTime = Date.now() - startTime;
if (response.status === this.config.expectedStatus) {
this.failureCount = 0;
this.log(`API正常 - 状态码: ${response.status}, 耗时: ${responseTime}ms`);
// 验证响应内容
if (this.config.validation) {
this.validateResponse(response.data);
}
// 检查响应时间
if (this.config.maxResponseTime && responseTime > this.config.maxResponseTime) {
this.log(`警告: 响应时间过长 (${responseTime}ms)`, 'WARN');
this.sendAlert(`API响应时间过长: ${responseTime}ms`);
}
} else {
this.failureCount++;
this.log(`错误: 状态码异常 (${response.status})`, 'ERROR');
this.handleFailure();
}
} catch (error) {
this.failureCount++;
this.log(`错误: ${error.message}`, 'ERROR');
this.handleFailure();
}
}
validateResponse(data) {
for (const [key, expectedValue] of Object.entries(this.config.validation)) {
if (data[key] !== expectedValue) {
this.log(`验证失败 - ${key}: 期望=${expectedValue}, 实际=${data[key]}`, 'ERROR');
this.sendAlert(`API响应验证失败: ${key}`);
}
}
}
handleFailure() {
if (this.failureCount >= (this.config.alertThreshold || 5)) {
this.sendAlert(`API连续失败 ${this.failureCount} 次`);
}
}
async sendAlert(message) {
this.log(`发送告警: ${message}`, 'ALERT');
if (this.config.email) {
await this.sendEmailAlert(message);
}
if (this.config.webhook) {
await this.sendWebhookAlert(message);
}
}
async sendEmailAlert(message) {
try {
const transporter = nodemailer.createTransport({
host: this.config.email.smtp,
port: this.config.email.port,
secure: true,
auth: {
user: this.config.email.user,
pass: this.config.email.password
}
});
await transporter.sendMail({
from: this.config.email.from,
to: this.config.email.to,
subject: `API监控告警 - ${new Date().toLocaleString()}`,
text: message
});
this.log('邮件告警已发送');
} catch (error) {
this.log(`发送邮件告警失败: ${error.message}`, 'ERROR');
}
}
async sendWebhookAlert(message) {
try {
await axios.post(this.config.webhook.url, {
msgtype: 'text',
text: { content: message }
});
this.log('Webhook告警已发送');
} catch (error) {
this.log(`发送Webhook告警失败: ${error.message}`, 'ERROR');
}
}
start(interval = 60) {
this.log(`开始监控API: ${this.config.url}`);
// 立即执行一次
this.checkAPI();
// 定时执行
setInterval(() => this.checkAPI(), interval * 1000);
}
}
// 使用示例
const config = {
url: 'https://api.example.com/health',
method: 'GET',
expectedStatus: 200,
timeout: 10000,
maxResponseTime: 5000,
alertThreshold: 3,
headers: {
'Authorization': 'Bearer YOUR_TOKEN'
},
validation: {
'status': 'ok'
},
email: {
smtp: 'smtp.gmail.com',
port: 465,
user: 'monitor@example.com',
password: 'YOUR_PASSWORD',
from: 'monitor@example.com',
to: 'admin@example.com'
},
webhook: {
url: 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY'
}
};
const monitor = new APIMonitor(config);
monitor.start(60); // 每60秒检查一次
Docker 部署方案
创建 Dockerfile：
FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY monitor.py . COPY config.json . CMD ["python", "monitor.py"]
创建 docker-compose.yml：
version: '3.8'
services:
api-monitor:
build: .
container_name: api-monitor
restart: always
volumes:
- ./logs:/app/logs
environment:
- TZ=Asia/Shanghai
logging:
driver: "json-file"
options:
max-size: "10m"
max-file: "3"
使用说明
安装依赖
Python:
pip install requests
Node.js:
npm install axios nodemailer cron
运行脚本
# Python python api_monitor.py # Shell chmod +x api_monitor.sh ./api_monitor.sh & # Node.js node api_monitor.js
添加系统服务（Linux）
创建 /etc/systemd/system/api-monitor.service：
[Unit] Description=API Monitor Service After=network.target [Service] Type=simple User=root WorkingDirectory=/opt/api-monitor ExecStart=/usr/bin/python3 /opt/api-monitor/monitor.py Restart=always RestartSec=10 [Install] WantedBy=multi-user.target
systemctl daemon-reload systemctl enable api-monitor systemctl start api-monitor
这个监控脚本支持：
- ✅ 多接口监控
- ✅ 状态码验证
- ✅ 响应时间监控
- ✅ 响应内容验证
- ✅ 多级告警（邮件、Webhook）
- ✅ 日志记录
- ✅ 自动重试
- ✅ 连续失败告警
根据您的具体需求,可以调整监控频率、告警方式和验证逻辑。
