如何编写任务超时告警脚本

wen 实用脚本 2026-06-26 24

本文目录导读：

如何编写任务超时告警脚本

基础Shell脚本方案
带重试机制的脚本
多任务并行监控脚本
Python版本（更灵活）
告警通知函数示例
使用建议

我来为你提供几种不同场景下的任务超时告警脚本方案。

基础Shell脚本方案

通用任务超时监控

#!/bin/bash
# 任务超时告警脚本
# usage: ./timeout_monitor.sh "command" timeout_seconds
COMMAND="$1"
TIMEOUT=${2:-300}  # 默认5分钟
ALERT_EMAIL="admin@example.com"
LOG_FILE="/var/log/timeout_alerts.log"
# 执行命令并获取PID
eval "$COMMAND" &
PID=$!
START_TIME=$(date +%s)
# 监控循环
while kill -0 $PID 2>/dev/null; do
    CURRENT_TIME=$(date +%s)
    ELAPSED=$((CURRENT_TIME - START_TIME))
    if [ $ELAPSED -gt $TIMEOUT ]; then
        # 超时告警
        ALERT_MSG="任务超时告警
        时间: $(date)
        命令: $COMMAND
        PID: $PID
        超时时间: ${TIMEOUT}秒
        实际运行: ${ELAPSED}秒
        主机: $(hostname)"
        # 记录日志
        echo "$ALERT_MSG" >> "$LOG_FILE"
        # 发送邮件告警
        echo "$ALERT_MSG" | mail -s "任务超时告警 - $(hostname)" "$ALERT_EMAIL"
        # 可选：杀死超时任务
        # kill -9 $PID
        exit 1
    fi
    sleep 5
done
# 任务正常完成
echo "任务在超时时间内完成" >> "$LOG_FILE"

带重试机制的脚本

#!/bin/bash
# 带重试的超时监控脚本
RETRY_COUNT=3
RETRY_DELAY=60  # 重试间隔（秒）
TIMEOUT=600     # 单个任务超时（秒）
execute_with_timeout() {
    local command="$1"
    local timeout=$2
    local attempt=1
    while [ $attempt -le $RETRY_COUNT ]; do
        echo "开始第 $attempt 次执行: $command"
        # 在后台执行命令
        eval "$command" &
        local cmd_pid=$!
        # 监控超时
        local elapsed=0
        while kill -0 $cmd_pid 2>/dev/null; do
            if [ $elapsed -ge $timeout ]; then
                # 超时处理
                send_alert "FAILED" "任务超时" "$command" $attempt
                kill -9 $cmd_pid 2>/dev/null
                if [ $attempt -lt $RETRY_COUNT ]; then
                    echo "等待 $RETRY_DELAY 秒后重试..."
                    sleep $RETRY_DELAY
                fi
                break
            fi
            sleep 1
            ((elapsed++))
        done
        # 检查是否成功完成
        wait $cmd_pid
        if [ $? -eq 0 ]; then
            echo "任务执行成功"
            return 0
        fi
        ((attempt++))
    done
    send_alert "FAILED" "所有重试失败" "$command" $RETRY_COUNT
    return 1
}
send_alert() {
    local status=$1
    local reason=$2
    local command=$3
    local attempt=$4
    echo "发送告警: $status - $reason"
    # 这里实现实际的告警通知（邮件、短信、webhook等）
}

多任务并行监控脚本

#!/bin/bash
# 多任务并行超时监控
declare -A TASKS
TASKS["备份任务"]="backup_script.sh"
TASKS["数据处理"]="data_process.py"
TASKS["报表生成"]="report_gen.sh"
TIMEOUT=300  # 5分钟
ALERT_WEBHOOK="https://hooks.slack.com/services/xxx"
monitor_task() {
    local task_name=$1
    local command=$2
    local timeout=$3
    # 执行任务
    eval "$command" &
    local pid=$!
    local start_time=$(date +%s)
    while kill -0 $pid 2>/dev/null; do
        local current_time=$(date +%s)
        local elapsed=$((current_time - start_time))
        if [ $elapsed -gt $timeout ]; then
            # 超时告警
            local message="{\"text\":\"任务超时告警\n任务: $task_name\n超时: ${timeout}s\n实际: ${elapsed}s\"}"
            curl -X POST -H 'Content-type: application/json' --data "$message" $ALERT_WEBHOOK
            kill -9 $pid 2>/dev/null
            return 1
        fi
        sleep 2
    done
    wait $pid
    local exit_code=$?
    if [ $exit_code -eq 0 ]; then
        echo "任务 $task_name 完成"
    else
        echo "任务 $task_name 失败 (退出码: $exit_code)"
    fi
    return $exit_code
}
# 并行监控所有任务
for task in "${!TASKS[@]}"; do
    monitor_task "$task" "${TASKS[$task]}" $TIMEOUT &
done
# 等待所有任务完成
wait
echo "所有任务监控完成"

Python版本（更灵活）

#!/usr/bin/env python3
import subprocess
import time
import threading
import logging
from datetime import datetime
class TaskTimeoutMonitor:
    def __init__(self, command, timeout=300, alert_callback=None):
        self.command = command
        self.timeout = timeout
        self.alert_callback = alert_callback
        self.process = None
        self.start_time = None
        # 配置日志
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('task_monitor.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
    def run(self):
        """执行任务并监控超时"""
        self.start_time = time.time()
        try:
            # 启动进程
            self.process = subprocess.Popen(
                self.command,
                shell=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE
            )
            # 创建超时监控线程
            monitor_thread = threading.Thread(target=self._monitor_timeout)
            monitor_thread.daemon = True
            monitor_thread.start()
            # 等待进程完成
            stdout, stderr = self.process.communicate()
            if self.process.returncode == 0:
                self.logger.info(f"任务完成: {self.command}")
                return True
            else:
                self.logger.error(f"任务失败: {stderr.decode()}")
                return False
        except Exception as e:
            self.logger.error(f"执行异常: {str(e)}")
            return False
    def _monitor_timeout(self):
        """监控超时"""
        while self.process and self.process.poll() is None:
            elapsed = time.time() - self.start_time
            if elapsed > self.timeout:
                self._handle_timeout(elapsed)
                break
            time.sleep(1)
    def _handle_timeout(self, elapsed):
        """处理超时"""
        alert_info = {
            'command': self.command,
            'timeout': self.timeout,
            'elapsed': elapsed,
            'pid': self.process.pid,
            'timestamp': datetime.now().isoformat(),
            'host': subprocess.getoutput('hostname')
        }
        self.logger.warning(f"任务超时: {alert_info}")
        # 终止进程
        self.process.kill()
        # 调用告警回调
        if self.alert_callback:
            self.alert_callback(alert_info)
# 使用示例
def send_alert(alert_info):
    """发送告警（示例）"""
    print(f"发送告警: {alert_info}")
    # 这里可以集成邮件、短信、企业微信等
if __name__ == "__main__":
    monitor = TaskTimeoutMonitor(
        command="sleep 400",  # 测试命令
        timeout=300,          # 5分钟超时
        alert_callback=send_alert
    )
    monitor.run()

告警通知函数示例

#!/bin/bash
# 多种告警方式
# 邮件告警
send_email_alert() {
    local subject="$1"
    local body="$2"
    local recipient="admin@company.com"
    echo "$body" | mail -s "$subject" "$recipient"
}
# 企业微信/钉钉机器人告警
send_webhook_alert() {
    local message="$1"
    local webhook_url="https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx"
    curl -s -X POST "$webhook_url" \
        -H "Content-Type: application/json" \
        -d "{\"msgtype\":\"text\",\"text\":{\"content\":\"$message\"}}"
}
# 短信告警（使用第三方API）
send_sms_alert() {
    local message="$1"
    local phone="13800138000"
    # 使用阿里云、腾讯云等短信服务
    curl -s "https://sms.xxx.com/send" \
        -d "phone=$phone&message=$message"
}
# 综合告警
send_alert() {
    local level=$1       # INFO, WARNING, CRITICAL
    local task_name=$2
    local detail=$3
    local alert_message="[${level}] 任务超时告警
    任务名称: $task_name
    详情: $detail
    时间: $(date '+%Y-%m-%d %H:%M:%S')
    主机: $(hostname)"
    # 根据级别选择通知方式
    case $level in
        "CRITICAL")
            send_email_alert "严重告警: $task_name" "$alert_message"
            send_sms_alert "$alert_message"
            ;;
        "WARNING")
            send_email_alert "警告: $task_name" "$alert_message"
            send_webhook_alert "$alert_message"
            ;;
        "INFO")
            send_webhook_alert "$alert_message"
            ;;
    esac
}

使用建议

设置合理的超时阈值：根据任务历史执行时间分析
分级告警：不同的超时级别使用不同的通知方式
日志记录：保存详细的执行记录便于排查
重试机制：对于临时性问题，加入自动重试
资源清理：超时后确保正确清理临时文件和进程

选择适合你场景的方案,建议先从简单的Shell脚本开始，逐步增加功能。