运维利器：提升十倍效率的自动化脚本实战分享

作者：ai | 发布时间：2026-05-01 07:01

运维利器：提升十倍效率的自动化脚本实战分享

在IT运维的日常工作中，我们常常被繁杂、重复的任务所包围：日志清理、健康巡检、批量部署、服务监控……如果完全依赖手动操作，不仅效率低下，且极易因“疲劳作战”导致误操作。运维界有句名言：“重复性的工作都应该被自动化”。

今天，我将从实际生产环境出发，分享几款经过检验的Shell与Python自动化脚本，帮助大家从繁琐的CRUD（创建、读取、更新、删除）操作中解放出来，将精力投入到更有价值的架构优化中。

一、 Shell脚本：轻量级系统运维的瑞士军刀

Shell脚本是与Linux系统交互最直接的工具，无需额外依赖，适合处理文件操作、系统监控等场景。

1. 系统健康巡检与告警脚本

在多台服务器上排查CPU、内存、磁盘瓶颈是家常便饭。这个脚本可以快速抓取系统核心指标，并在超过阈值时输出告警（可结合Crontab定时执行，结果通过邮件或企微/钉钉Webhook发送）。


#!/bin/bash
# 系统健康巡检脚本

# 定义阈值
CPU_THRESHOLD=80
MEM_THRESHOLD=85
DISK_THRESHOLD=90
ALERT_MSG=""

# 获取CPU使用率 (取1分钟的平均负载近似值，或使用top复杂解析，此处简化)
CPU_USAGE=$(top -bn1 | grep "Cpu(s)" | sed "s/.* *\([0-9.]*\)%* id.*/\1/" | awk '{print 100 - $1}')
CPU_USAGE=${CPU_USAGE%.*} # 取整

# 获取内存使用率
MEM_USAGE=$(free -m | awk '/Mem/{printf("%.0f"), $3/$2*100}')

# 获取磁盘使用率最高的分区
DISK_USAGE=$(df -h | awk 'NR!=1{print $5}' | sort -nr | head -n1 | tr -d '%')

# 判断是否超限
if [ "$CPU_USAGE" -gt "$CPU_THRESHOLD" ]; then
  ALERT_MSG+="[告警] CPU使用率: ${CPU_USAGE}% (阈值: ${CPU_THRESHOLD}%)\n"
fi

if [ "$MEM_USAGE" -gt "$MEM_THRESHOLD" ]; then
  ALERT_MSG+="[告警] 内存使用率: ${MEM_USAGE}% (阈值: ${MEM_THRESHOLD}%)\n"
fi

if [ "$DISK_USAGE" -gt "$DISK_THRESHOLD" ]; then
  ALERT_MSG+="[告警] 磁盘最高使用率: ${DISK_USAGE}% (阈值: ${DISK_THRESHOLD}%)\n"
fi

if [ -z "$ALERT_MSG" ]; then
  echo "$(date '+%F %T') - 系统状态正常 [CPU:${CPU_USAGE}%] [MEM:${MEM_USAGE}%] [DISK:${DISK_USAGE}%]"
else
  echo -e "$(date '+%F %T') - 系统异常告警:\n$ALERT_MSG"
  # 此处可插入发送告警的API调用，例如: curl -X POST $WEBHOOK_URL -d "content=$ALERT_MSG"
fi

2. 过期日志自动清理脚本

磁盘被日志打满是生产环境的常见灾难。手动清理费时费力，此脚本通过正则匹配与时间判定，安全清理过期日志。


#!/bin/bash
# 日志安全清理脚本

LOG_DIR="/var/log/app"
DAYS_TO_KEEP=7

echo "开始清理 ${LOG_DIR} 下超过 ${DAYS_TO_KEEP} 天的日志..."

# 查找并删除修改时间超过DAYS_TO_KEEP天的 .log 文件
find ${LOG_DIR} -name "*.log" -type f -mtime +${DAYS_TO_KEEP} -exec rm -f {} \;

# 清理空目录
find ${LOG_DIR} -type d -empty -delete

echo "清理完成。当前磁盘状态："
df -h ${LOG_DIR}

二、 Python脚本：复杂逻辑与跨平台操作的利器

当涉及复杂的数据结构处理、API交互或跨平台操作时，Python凭借其丰富的第三方库，表现远胜于Shell。

1. 批量SSH命令执行器

在集群运维中，我们需要在几十台机器上执行同一命令（如重启服务、查看进程）。借助paramiko库，可以轻松实现并发SSH操作。

安装依赖：pip install paramiko


#!/usr/bin/env python3
# 批量SSH并发执行脚本

import paramiko
import concurrent.futures

# 服务器列表 (实际生产中建议从CMDB或配置文件读取)
SERVERS = [
    {"host": "192.168.1.10", "username": "root", "password": "your_password"},
    {"host": "192.168.1.11", "username": "root", "password": "your_password"},
]
COMMAND_TO_RUN = "uptime" # 需要执行的命令

def execute_ssh(server_info, command):
    host = server_info['host']
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    try:
        ssh.connect(hostname=host, username=server_info['username'], password=server_info['password'], timeout=5)
        stdin, stdout, stderr = ssh.exec_command(command)
        result = stdout.read().decode('utf-8').strip()
        print(f"[成功] {host} 执行结果:\n{result}")
    except Exception as e:
        print(f"[失败] {host} 连接或执行异常: {str(e)}")
    finally:
        ssh.close()

if __name__ == "__main__":
    print(f"开始批量执行命令: {COMMAND_TO_RUN}")
    # 使用线程池并发执行，提高效率
    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        futures = [executor.submit(execute_ssh, server, COMMAND_TO_RUN) for server in SERVERS]
        concurrent.futures.wait(futures)
    print("批量执行完毕")

2. HTTP服务可用性探针

对于Web业务，服务的端口存活不代表接口可用。此脚本通过模拟用户请求，检测接口HTTP状态码与响应时间，实现更深度的业务级监控。


#!/usr/bin/env python3
# HTTP接口可用性监控

import requests
import time

URLS = [
    "https://api.yourdomain.com/health",
    "https://portal.yourdomain.com/login"
]
TIMEOUT = 5  # 超时时间(秒)
STATUS_ALERT = 2  # 连续失败次数触发告警

def check_url(url, fail_count=0):
    try:
        start_time = time.time()
        response = requests.get(url, timeout=TIMEOUT, verify=False)
        elapsed_ms = (time.time() - start_time) * 1000
        
        if response.status_code == 200:
            print(f"[正常] {url} | 状态码: {response.status_code} | 耗时: {elapsed_ms:.0f}ms")
            return 0  # 重置失败计数
        else:
            print(f"[异常] {url} | 状态码: {response.status_code} | 耗时: {elapsed_ms:.0f}ms")
            return fail_count + 1
    except requests.RequestException as e:
        print(f"[故障] {url} | 请求异常: {str(e)}")
        return fail_count + 1

if __name__ == "__main__":
    # 简化演示：单次轮询。生产环境可加入while True与sleep
    for url in URLS:
        failures = check_url(url)
        if failures >= STATUS_ALERT:
            # 触发告警逻辑，如调用钉钉机器人API
            print(f"!!! 触发告警: {url} 连续不可用 !!!")

三、运维自动化的避坑指南

在享受自动化带来便利的同时，不严谨的脚本也可能成为“灾难”的源头。分享几个实战中的核心原则：

幂等性是第一准则：无论脚本执行多少次，系统状态应该是一致的。例如，创建目录前先mkdir -p，创建用户前先id username判断是否存在，避免重复执行报错。
告别硬编码：IP、密码、路径等

← 返回AI专栏

运维利器：提升十倍效率的自动化脚本实战分享