如何编写接口响应监控脚本

wen 实用脚本 2026-06-27 248

本文目录导读：

如何编写接口响应监控脚本

Python 版本（最常用）
Shell 脚本版本（轻量级）
Node.js 版本
Docker 部署方案
使用说明

我来帮您编写一个完整的接口响应监控脚本,这里提供几种常用语言的实现方案：

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）
✅ 日志记录
✅ 自动重试
✅ 连续失败告警

根据您的具体需求,可以调整监控频率、告警方式和验证逻辑。

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