当前位置：首页 > news >正文

第六天 - os/subprocess模块 - 系统进程管理 - 练习：服务状态监控脚本

news 来源：原创 2025/5/19 12:14:13

Python系统管理全攻略：从subprocess到服务监控脚本

一、为什么要学习系统进程管理？

在日常开发工作中，我们经常会遇到需要与操作系统交互的场景：

执行Shell命令
管理后台服务进程
监控系统资源使用情况
自动化运维操作

掌握Python的os和subprocess模块，可以帮助我们以编程方式完成这些系统管理工作。本文将从基础用法到实战案例，带你系统掌握进程管理的核心技能。

二、os模块基础操作

2.1 执行简单命令

import os

# 执行系统命令（阻塞式）
return_code = os.system("ls -l")
print(f"命令执行返回码：{return_code}")

虽然os.system()使用简单，但它存在明显局限性：

无法获取命令输出
不能处理复杂交互
安全性较差（存在命令注入风险）

2.2 环境变量管理

# 获取环境变量
print(os.environ['PATH'])

# 设置临时环境变量
os.environ['MY_VAR'] = 'test_value'

三、subprocess模块深入解析

subprocess模块是更现代、更强大的命令执行解决方案，提供了丰富的控制功能。

3.1 基础用法

import subprocess

# 执行简单命令
result = subprocess.run(['ls', '-l'], capture_output=True, text=True)
print(result.stdout)

参数说明：

capture_output：捕获命令输出
text：以文本形式返回结果
check：自动检查返回码

3.2 高级功能示例

# 带超时控制的命令执行
try:
    result = subprocess.run(
        ['ping', 'google.com'],
        timeout=5,
        capture_output=True,
        text=True
    )
except subprocess.TimeoutExpired:
    print("命令执行超时！")

# 管道操作
p1 = subprocess.Popen(['cat', 'large_file.txt'], stdout=subprocess.PIPE)
p2 = subprocess.Popen(['grep', 'error'], stdin=p1.stdout, stdout=subprocess.PIPE)
output = p2.communicate()[0]

3.3 安全注意事项

# 危险用法（存在命令注入风险）
subprocess.run(f"rm -rf {user_input}", shell=True)

# 安全用法
subprocess.run(['rm', '-rf', user_input])

四、系统进程管理实战

4.1 进程生命周期管理

# 启动后台进程
server_process = subprocess.Popen(
    ['python3', 'my_server.py'],
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT
)

# 终止进程
server_process.terminate()  # 优雅终止
server_process.kill()       # 强制终止

# 等待进程结束
try:
    server_process.wait(timeout=60)
except subprocess.TimeoutExpired:
    print("进程未在指定时间内退出")

4.2 进程信息监控

def get_process_info(pid):
    """获取指定进程的详细信息"""
    try:
        output = subprocess.check_output(
            ['ps', '-p', str(pid), '-o', 'pid,ppid,time,%cpu,%mem,cmd'],
            text=True
        )
        return output.split('\n')[1]  # 跳过标题行
    except subprocess.CalledProcessError:
        return "进程不存在"

五、服务状态监控脚本开发

5.1 需求分析

开发一个具备以下功能的监控脚本：

定期检查指定服务状态
发现异常自动重启
记录监控日志
支持邮件报警

5.2 核心实现代码

import subprocess
import time
import logging
from datetime import datetime

# 配置日志
logging.basicConfig(
    filename='service_monitor.log',
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

def check_service(service_name):
    """检查服务是否在运行"""
    try:
        output = subprocess.check_output(
            ['pgrep', '-f', service_name],
            stderr=subprocess.STDOUT
        )
        return True
    except subprocess.CalledProcessError:
        return False

def restart_service(service_cmd):
    """重启指定服务"""
    try:
        subprocess.run(service_cmd, check=True, timeout=30)
        return True
    except Exception as e:
        logging.error(f"服务重启失败：{str(e)}")
        return False

def monitor(service_name, service_cmd, interval=60):
    """监控主循环"""
    while True:
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        if not check_service(service_name):
            logging.warning(f"[{timestamp}] 服务 {service_name} 已停止")
            if restart_service(service_cmd):
                logging.info(f"[{timestamp}] 服务重启成功")
                # 这里可以添加邮件通知功能
            else:
                logging.error(f"[{timestamp}] 服务重启失败")
        else:
            logging.info(f"[{timestamp}] 服务运行正常")
        time.sleep(interval)

if __name__ == "__main__":
    # 示例：监控Nginx服务
    monitor(
        service_name="nginx",
        service_cmd=["sudo", "systemctl", "restart", "nginx"],
        interval=300  # 5分钟检查一次
    )

5.3 功能扩展建议

通知功能增强

import smtplib
from email.mime.text import MIMEText

def send_alert(email, message):
    msg = MIMEText(message)
    msg['Subject'] = '服务异常报警'
    msg['From'] = 'monitor@example.com'
    msg['To'] = email
    
    with smtplib.SMTP('smtp.example.com') as server:
        server.send_message(msg)

WEB管理界面

from flask import Flask, jsonify

app = Flask(__name__)

@app.route('/service/status')
def get_status():
    return jsonify({
        'status': 'running' if check_service('nginx') else 'stopped'
    })

六、最佳实践与常见问题

6.1 安全注意事项

避免使用shell=True
正确处理用户输入
限制子进程权限
设置合理的超时时间

6.2 性能优化技巧

# 使用命令内置参数减少数据处理
subprocess.run(['ps', '-eo', 'pid,ppid,comm'])  # 指定需要输出的列

# 复用Process实例
with subprocess.Popen(['top'], stdout=subprocess.PIPE) as proc:
    while True:
        line = proc.stdout.readline()
        if not line:
            break
        # 处理实时输出

6.3 常见错误处理

try:
    output = subprocess.check_output(
        ['invalid_cmd'],
        stderr=subprocess.STDOUT,
        text=True
    )
except FileNotFoundError as e:
    print(f"命令不存在：{e}")
except subprocess.CalledProcessError as e:
    print(f"命令执行失败，返回码：{e.returncode}")
    print(f"错误输出：{e.output}")