系统与进程管理
巡检脚本的原料是系统指标:CPU、内存、磁盘、网络、进程。shell 里这些靠 top、df、ss 拼输出,到 Python 里一个 psutil 全包了——跨平台、返回结构化数据、不用解析文本。这篇笔记记录 psutil 的高频用法,外加进程管理、定时任务和常驻脚本的信号处理。示例环境:macOS / Linux,Python 3.12+。
python3 -m pip install psutil schedule apscheduler
platform、os、signal 是标准库,不用装。
psutil:运维巡检的瑞士军刀
python3 -m pip install psutil
逐项过一遍巡检常用接口,每项配真实输出。
CPU:cpu_percent
import psutil
psutil.cpu_percent(interval=1) # 阻塞采样 1 秒,返回这段时间的平均使用率
psutil.cpu_percent(interval=1, percpu=True) # 每核一个值
psutil.cpu_count(logical=False) # 物理核数
13.2
[11.0, 15.2, 9.8, 17.1, 12.4, 14.9, 11.5, 13.0]
8
第一次调用 cpu_percent(interval=0) 或非阻塞调用会返回没意义的 0.0(它算的是「距上次调用」的增量),巡检脚本里统一用 interval=1 采样一秒最稳。
内存:virtual_memory 与 swap_memory
mem = psutil.virtual_memory()
print(f"总量 {mem.total/1024**3:.1f}G 已用 {mem.percent}% 可用 {mem.available/1024**3:.1f}G")
swap = psutil.swap_memory()
print(f"swap 总量 {swap.total/1024**3:.1f}G 已用 {swap.percent}%")
总量 16.0G 已用 72.4% 可用 3.2G
swap 总量 2.0G 已用 41.3%
Linux 会把空闲内存拿去做页缓存,free 很小不代表内存紧张。available 才是「现在还能拿出多少给新进程」,告警阈值按它定。
磁盘:disk_usage 与 disk_io_counters
u = psutil.disk_usage("/")
print(f"/ 总量 {u.total/1024**3:.0f}G 已用 {u.percent}%")
io = psutil.disk_io_counters()
print(f"读 {io.read_bytes/1024**3:.1f}G 写 {io.write_bytes/1024**3:.1f}G(开机累计)")
/ 总量 465G 已用 61.2%
读 812.4G 写 1534.7G(开机累计)
disk_io_counters 返回的是开机以来的累计值,算 IOPS 或速率要自己间隔采样两次取差值:
import time
a = psutil.disk_io_counters()
time.sleep(1)
b = psutil.disk_io_counters()
print(f"读速率 {(b.read_bytes - a.read_bytes)/1024**2:.1f} MB/s")
网络:net_io_counters
net = psutil.net_io_counters()
print(f"收 {net.bytes_recv/1024**3:.1f}G 发 {net.bytes_sent/1024**3:.1f}G 丢包 {net.dropin + net.dropout}")
收 234.5G 发 891.2G 丢包 0
同样是累计值,算带宽就间隔采样。按网卡分开看用 psutil.net_io_counters(pernic=True)。
启动时间:boot_time
from datetime import datetime
bt = datetime.fromtimestamp(psutil.boot_time())
print(f"系统启动于 {bt:%Y-%m-%d %H:%M:%S}")
系统启动于 2026-09-08 21:14:32
拼一个巡检小报告
把上面的揉成一段,就是巡检脚本的骨架:
#!/usr/bin/env python3
"""inspect.py:打印机器基础巡检信息"""
from datetime import datetime
import psutil
mem = psutil.virtual_memory()
disk = psutil.disk_usage("/")
print(f"主机启动: {datetime.fromtimestamp(psutil.boot_time()):%Y-%m-%d %H:%M:%S}")
print(f"CPU: {psutil.cpu_percent(interval=1)}%")
print(f"内存: {mem.percent}% (可用 {mem.available/1024**3:.1f}G)")
print(f"磁盘/: {disk.percent}%")
swap = psutil.swap_memory()
if swap.percent > 50:
print(f"!! swap 使用 {swap.percent}%,内存可能不足")
进程管理
按名字找进程
process_iter 遍历全机进程,attrs 指定要取哪些属性(只取需要的,快很多):
import psutil
for p in psutil.process_iter(attrs=["pid", "name", "status"]):
if p.info["name"] == "nginx":
print(p.info)
{'pid': 812, 'name': 'nginx', 'status': 'sleeping'}
{'pid': 813, 'name': 'nginx', 'status': 'sleeping'}
资源占用 Top5
「这台机器谁最吃内存」的脚本版:
#!/usr/bin/env python3
"""top_mem.py:内存占用 Top5 进程"""
import psutil
procs = []
for p in psutil.process_iter(attrs=["pid", "name", "memory_percent"]):
try:
procs.append(p.info)
except (psutil.NoSuchProcess, psutil.AccessDenied):
continue
procs.sort(key=lambda x: x["memory_percent"] or 0, reverse=True)
for info in procs[:5]:
print(f"{info['pid']:<8} {info['memory_percent']:5.1f}% {info['name']}")
1024 12.3% java
812 5.8% nginx
...
遍历进程列表的瞬间,有的进程退出了(NoSuchProcess),有的进程你没权限看(AccessDenied)。不捕获这两个异常,TopN 脚本跑十次总有一次崩在半截。上面例子里的 try/except 是标配,不是可选。
kill 进程的姿势
import psutil
p = psutil.Process(2310)
p.terminate() # 发 SIGTERM,给进程清理的机会
try:
p.wait(timeout=5) # 等它真的退
except psutil.TimeoutExpired:
p.kill() # 5 秒还不退再 SIGKILL
原则:先 terminate 后 kill,直接 kill 可能留下写了一半的文件和没释放的锁。按名字批量杀(比如重启前清残留进程):
targets = [p for p in psutil.process_iter(attrs=["name"]) if p.info["name"] == "old_app"]
for p in targets:
p.terminate()
gone, alive = psutil.wait_procs(targets, timeout=5)
for p in alive: # 超期不退的补刀
p.kill()
网络连接
「谁占了 8080」——不用再 lsof -i :8080 然后解析输出:
#!/usr/bin/env python3
"""who_listen.py:找出占用指定端口的进程"""
import sys
import psutil
port = int(sys.argv[1]) if len(sys.argv) > 1 else 8080
for conn in psutil.net_connections(kind="inet"):
if conn.laddr.port == port and conn.status == "LISTEN":
p = psutil.Process(conn.pid)
print(f"端口 {port} 被 PID {conn.pid} 占用: {p.name()} (cmdline: {' '.join(p.cmdline())})")
python3 who_listen.py 8080
端口 8080 被 PID 2310 占用: java (cmdline: java -jar /opt/app/app.jar --server.port=8080)
net_connections 不加 kind 会连 UNIX socket 一起列出来,看 TCP/UDP 监听就固定 kind="inet",再筛 status == "LISTEN"。macOS 上看别的用户的进程需要 sudo 跑。
定位到进程之后,如果流量层面还是说不清(连接建了但没数据、对端是谁),就该抓包了——参考 tcpdump 实战笔记;目标进程在容器里的话,用 nsenter 进容器命名空间排查,宿主机上不用装任何东西。
系统信息
巡检报告的抬头——发行版、内核、主机名——用标准库就够了:
import os
import platform
import socket
print(platform.system()) # Linux / Darwin
print(platform.release()) # 内核版本 5.15.0-139-generic
print(socket.gethostname()) # 主机名 web1
u = os.uname() # Linux/macOS 专有,一把拿全
print(u.sysname, u.release, u.machine)
Linux
5.15.0-139-generic
web1
Linux 5.15.0-139-generic x86_64
运维脚本基本只跑 Linux/macOS,问题不大;真要考虑跨平台就用 platform.uname(),字段等价。
要拿发行版名字(Ubuntu 22.04 这种),标准库里没有现成接口,读 /etc/os-release 最实在:
def distro() -> str:
try:
with open("/etc/os-release") as f:
for line in f:
if line.startswith("PRETTY_NAME="):
return line.split("=", 1)[1].strip().strip('"')
except FileNotFoundError:
return platform.system() # macOS 没有 os-release
return "unknown"
print(distro()) # Ubuntu 22.04.5 LTS
定时任务
schedule:轻量定时
脚本级别的「每 N 分钟跑一次」,schedule 库最省事:
python3 -m pip install schedule
import time
import schedule
def job():
print("跑一轮巡检...")
schedule.every(5).minutes.do(job) # 每 5 分钟
schedule.every().day.at("03:30").do(job) # 每天 03:30
while True:
schedule.run_pending()
time.sleep(1)
就是一个 while 循环加时间表,没有依赖、没有后台线程,日志脚本、拨测脚本用这个刚好。
APScheduler:正经的调度器
要 cron 表达式、要后台跑、要任务持久化,上 APScheduler:
python3 -m pip install apscheduler
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
def backup():
print("执行备份...")
scheduler = BackgroundScheduler()
# 每天凌晨 3 点 30 跑(cron 五字段:分 时 日 月 周)
scheduler.add_job(backup, CronTrigger.from_crontab("30 3 * * *"))
# 每 10 分钟一次也可以直接写
scheduler.add_job(backup, "interval", minutes=10)
scheduler.start()
# 主线程得活着,BackgroundScheduler 是后台线程
import time
while True:
time.sleep(60)
BackgroundScheduler 适合嵌在常驻进程里(比如自己的 agent 程序);要在前台跑用 BlockingScheduler,它会自己占住主线程。
什么时候让位给系统 cron
「每天凌晨清一次日志」这种任务,写进 crontab 一行解决,系统帮你管启动、管日志重定向、机器重启自动恢复。Python 常驻进程意味着你要自己管:进程挂了谁拉起(systemd/supervisor)、内存泄漏怎么办、日志轮转怎么办。只有调度逻辑复杂(依赖上次执行结果、任务间有编排、要动态增删任务)时才值得常驻。固定时间跑固定脚本,交给 cron。
信号与优雅退出
常驻脚本(上面那种 while 循环)总有关掉的一天:systemctl stop 发 SIGTERM,Ctrl+C 发 SIGINT。默认行为是立刻死掉——写了一半的文件、没 flush 的日志、没关的连接全撂在半路上。接住信号、做完清理再退出,是常驻脚本的基本功:
#!/usr/bin/env python3
"""daemon.py:优雅退出的常驻脚本骨架"""
import signal
import time
running = True
def handle_exit(signum, frame):
global running
print(f"收到信号 {signum},做完手头的事就退出...")
running = False
signal.signal(signal.SIGTERM, handle_exit) # systemctl stop / kill
signal.signal(signal.SIGINT, handle_exit) # Ctrl+C
while running:
print("干活中...")
time.sleep(2)
# 循环退出后再做清理
print("清理资源(关连接、flush 日志、删 pid 文件)...")
跑起来后从另一个终端 kill <pid>,会看到它打印「收到信号」、跑完当前这一轮循环、执行清理、然后才退出,而不是瞬间消失。
两个要点:
- 信号处理函数里别干重活,只设标志位(像上面的
running = False),真正的清理由主循环退出后执行。处理函数里做复杂操作有重入风险。 - SIGKILL 接不住(这是它的设计目的),所以清理逻辑要容忍「没被执行」,重要数据别只放在内存里等退出时落盘。
小结
- 系统指标找 psutil:CPU 用
cpu_percent(interval=1),内存看available,磁盘/网络计数器是累计值要采样取差。 - 进程遍历必捕
NoSuchProcess和AccessDenied;杀进程先terminate再kill。 net_connections替代lsof/ss找端口占用;流量层面还说不清就联动 tcpdump、nsenter。- 定时任务:轻量用 schedule,要 cron 用 APScheduler,固定简单任务交给系统 cron。
- 常驻脚本必须接 SIGTERM/SIGINT,处理函数只设标志位,清理由主循环做——这样脚本才经得起
systemctl stop。