Python运行指标采集是指通过代码获取程序运行时的系统资源占用、业务运行状态等信息，为性能优化、故障排查、服务监控提供数据支撑，常见的采集指标包括CPU使用率、内存占用、磁盘IO、网络流量、接口响应耗时等。
基于psutil库的通用系统指标采集
psutil是Python中常用的系统指标采集第三方库，支持跨平台获取CPU、内存、磁盘、网络等系统层面的运行指标，安装方式为pip install psutil。以下是采集基础系统指标的示例代码：
import psutil
import time
def collect_system_metrics():
# 采集CPU使用率，间隔1秒统计
cpu_percent = psutil.cpu_percent(interval=1)
# 采集内存信息
memory_info = psutil.virtual_memory()
# 采集磁盘根分区使用情况
disk_info = psutil.disk_usage('/')
# 采集网络收发字节数
net_info = psutil.net_io_counters()
metrics = {
"cpu_percent": cpu_percent,
"memory_total": memory_info.total,
"memory_used": memory_info.used,
"memory_percent": memory_info.percent,
"disk_total": disk_info.total,
"disk_used": disk_info.used,
"disk_percent": disk_info.percent,
"net_bytes_sent": net_info.bytes_sent,
"net_bytes_recv": net_info.bytes_recv,
"collect_time": time.time()
}
return metrics
if __name__ == "__main__":
result = collect_system_metrics()
for key, value in result.items():
print(f"{key}: {value}")
自定义业务运行指标采集
除了系统层面的通用指标，业务场景还需要采集自定义的运行指标，比如接口调用次数、接口响应耗时、异常发生次数等。这类指标通常通过埋点的方式实现，以下是接口耗时统计的示例：
import time
from functools import wraps
# 存储业务指标的全局字典
business_metrics = {
"api_call_count": 0,
"api_total_cost": 0,
"api_max_cost": 0,
"exception_count": 0
}
def api_cost_statistics(func):
"""接口耗时统计装饰器"""
@wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
try:
result = func(*args, **kwargs)
return result
except Exception as e:
business_metrics["exception_count"] += 1
raise e
finally:
cost = time.time() - start_time
business_metrics["api_call_count"] += 1
business_metrics["api_total_cost"] += cost
if cost > business_metrics["api_max_cost"]:
business_metrics["api_max_cost"] = cost
return wrapper
@api_cost_statistics
def test_api():
"""模拟接口逻辑"""
time.sleep(0.5)
return "success"
if __name__ == "__main__":
# 模拟调用3次接口
for _ in range(3):
test_api()
print("业务指标采集结果：")
for key, value in business_metrics.items():
print(f"{key}: {value}")
采集数据的上报与存储
采集到的运行指标需要上报到存储或监控系统中才能发挥价值，常见的上报方式包括写入本地日志文件、发送到时序数据库、推送到监控平台等。以下是写入本地JSON格式日志的示例：
import json
import time
import psutil
from datetime import datetime
def write_metrics_to_log(metrics, log_path="metrics.log"):
"""将采集到的指标写入本地日志文件"""
metrics["log_time"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
with open(log_path, "a", encoding="utf-8") as f:
f.write(json.dumps(metrics, ensure_ascii=False) + "n")
def collect_and_report():
"""采集系统指标并上报"""
cpu_percent = psutil.cpu_percent(interval=1)
memory_info = psutil.virtual_memory()
metrics = {
"cpu_percent": cpu_percent,
"memory_percent": memory_info.percent
}
write_metrics_to_log(metrics)
if __name__ == "__main__":
# 每10秒采集上报一次
while True:
collect_and_report()
time.sleep(10)
不同采集场景的适配思路
根据项目规模和需求的不同，运行指标采集的实现思路可以做针对性调整：
- 小型单机项目：直接使用psutil采集系统指标，结合自定义埋点采集业务指标，写入本地日志即可满足基本需求。
- 分布式服务场景：需要统一采集各个节点的指标，将指标上报到Prometheus等时序数据库，配合Grafana做可视化展示。
- 高并发场景：采集逻辑要避免影响主业务性能，可采用异步上报、采样采集等方式降低采集带来的额外开销。
注意事项
在实现Python运行指标采集时，需要注意以下几点：
采集频率要合理，过高的采集频率会增加系统额外负载，一般系统指标采集间隔设置为1-5秒即可。
敏感指标要做脱敏处理，比如避免采集和上报用户隐私相关的业务数据。
采集代码要做好异常处理，避免采集逻辑报错影响主业务的正常运行。