本文目录导读：
我来分享几个Python内存监控的实用案例：
基础内存监控器
import psutil
import os
import time
from datetime import datetime
class MemoryMonitor:
def __init__(self):
self.process = psutil.Process(os.getpid())
def get_memory_usage(self):
"""获取当前进程内存使用"""
memory_info = self.process.memory_info()
return {
'rss': memory_info.rss / 1024 / 1024, # MB
'vms': memory_info.vms / 1024 / 1024, # MB
'percent': self.process.memory_percent(),
'time': datetime.now().strftime('%H:%M:%S')
}
def monitor_task(self, task_func, *args, **kwargs):
"""监控任务执行期间的内存变化"""
start_mem = self.get_memory_usage()
print(f"开始内存: RSS={start_mem['rss']:.2f}MB, VMS={start_mem['vms']:.2f}MB")
result = task_func(*args, **kwargs)
end_mem = self.get_memory_usage()
print(f"结束内存: RSS={end_mem['rss']:.2f}MB, VMS={end_mem['vms']:.2f}MB")
print(f"内存变化: +{end_mem['rss']-start_mem['rss']:.2f}MB")
return result
# 使用示例
monitor = MemoryMonitor()
def memory_intensive_task(size_mb):
"""模拟内存密集型任务"""
data = [0] * (size_mb * 1024 * 1024 // 8) # 每个int大约28字节
time.sleep(2)
return len(data)
# 监控任务
result = monitor.monitor_task(memory_intensive_task, 100)
持续内存监控器
import threading
import matplotlib.pyplot as plt
from collections import deque
class ContinuousMonitor:
def __init__(self, interval=1):
self.interval = interval
self.monitor = MemoryMonitor()
self.memory_history = deque(maxlen=100)
self.time_history = deque(maxlen=100)
self.is_running = False
def start(self):
"""启动持续监控"""
self.is_running = True
self.thread = threading.Thread(target=self._monitor_loop)
self.thread.daemon = True
self.thread.start()
def stop(self):
"""停止监控"""
self.is_running = False
def _monitor_loop(self):
"""监控循环"""
while self.is_running:
mem_info = self.monitor.get_memory_usage()
self.memory_history.append(mem_info['rss'])
self.time_history.append(mem_info['time'])
# 实时打印
self._print_status(mem_info)
time.sleep(self.interval)
def _print_status(self, mem_info):
"""打印状态"""
mem_bar = '█' * int(mem_info['percent'] / 2) + '░' * (50 - int(mem_info['percent'] / 2))
print(f"\r[{mem_info['time']}] 内存: {mem_info['rss']:.1f}MB | {mem_bar} {mem_info['percent']:.1f}%", end='')
def plot_memory_usage(self):
"""绘制内存使用曲线"""
plt.figure(figsize=(10, 6))
plt.plot(list(self.time_history), list(self.memory_history), 'b-', linewidth=2)
plt.xlabel('时间')
plt.ylabel('内存使用 (MB)')
plt.title('内存使用监控')
plt.xticks(rotation=45)
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
# 使用示例
monitor = ContinuousMonitor(interval=0.5)
monitor.start()
time.sleep(10) # 模拟程序运行
monitor.stop()
monitor.plot_memory_usage()
函数级内存监控装饰器
import functools
import tracemalloc
def memory_profile(func):
"""内存监控装饰器"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
# 启动内存追踪
tracemalloc.start()
# 记录初始内存
snapshot1 = tracemalloc.take_snapshot()
# 执行函数
result = func(*args, **kwargs)
# 记录结束内存
snapshot2 = tracemalloc.take_snapshot()
# 计算差异
top_stats = snapshot2.compare_to(snapshot1, 'lineno')
print(f"\n函数 '{func.__name__}' 内存分析:")
print(f"{'='*60}")
total_diff = sum(stat.size_diff for stat in top_stats)
print(f"总内存变化: {total_diff / 1024:.2f} KB")
# 显示内存变化最大的10个位置
print(f"\n{'内存变化最大的位置':^50}")
print(f"{'大小(KB)':<12} {'计数':<10} {'位置':<30}")
print("-"*52)
for stat in top_stats[:10]:
size_kb = stat.size_diff / 1024
if size_kb != 0:
print(f"{size_kb:<12.1f} {stat.count_diff:<10} {str(stat.traceback)[:30]}")
tracemalloc.stop()
return result
return wrapper
# 使用示例
@memory_profile
def create_large_list(size):
"""创建大列表"""
return [i for i in range(size)]
@memory_profile
def process_data(size):
"""处理数据"""
data = [x * 2 for x in range(size)]
data_dict = {i: x for i, x in enumerate(data)}
return data_dict
create_large_list(1000000)
process_data(500000)
完整的内存监控系统
import json
import logging
from dataclasses import dataclass
from typing import Callable, Any
@dataclass
class MemoryAlert:
"""内存警报"""
threshold: float
current: float
message: str
timestamp: datetime
class MemoryMonitoringSystem:
def __init__(self,
warning_threshold: float = 80.0, # 警告阈值 (%)
critical_threshold: float = 90.0, # 严重阈值 (%)
log_file: str = 'memory_monitor.log'):
self.warning_threshold = warning_threshold
self.critical_threshold = critical_threshold
self.alerts = []
# 配置日志
logging.basicConfig(
filename=log_file,
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
self.logger = logging.getLogger(__name__)
self.monitor = MemoryMonitor()
def check_memory(self) -> dict:
"""检查内存状态"""
mem_info = self.monitor.get_memory_usage()
# 检查是否需要发出警报
if mem_info['percent'] >= self.critical_threshold:
alert = MemoryAlert(
threshold=self.critical_threshold,
current=mem_info['percent'],
message=f"严重内存使用率: {mem_info['percent']:.1f}%",
timestamp=datetime.now()
)
self.alerts.append(alert)
self.logger.critical(alert.message)
elif mem_info['percent'] >= self.warning_threshold:
alert = MemoryAlert(
threshold=self.warning_threshold,
current=mem_info['percent'],
message=f"内存使用率警告: {mem_info['percent']:.1f}%",
timestamp=datetime.now()
)
self.alerts.append(alert)
self.logger.warning(alert.message)
return mem_info
def save_snapshot(self, filename: str = 'memory_snapshot.json'):
"""保存内存快照"""
mem_info = self.check_memory()
snapshot = {
'timestamp': datetime.now().isoformat(),
'memory': mem_info,
'alerts': [
{
'threshold': a.threshold,
'current': a.current,
'message': a.message,
'time': a.timestamp.isoformat()
}
for a in self.alerts[-10:] # 只保存最近10条警报
]
}
with open(filename, 'w') as f:
json.dump(snapshot, f, indent=2)
self.logger.info(f"内存快照已保存: {filename}")
def run_monitoring(self,
duration: int = 60,
callback: Callable = None):
"""运行监控"""
start_time = time.time()
snapshot_count = 0
while time.time() - start_time < duration:
mem_info = self.check_memory()
if callback:
callback(mem_info)
# 每10秒保存一次快照
if snapshot_count % 10 == 0:
self.save_snapshot(f'memory_snapshot_{snapshot_count}.json')
snapshot_count += 1
time.sleep(1)
# 生成报告
self.generate_report()
def generate_report(self):
"""生成监控报告"""
print(f"\n{'='*50}")
print(f"内存监控报告")
print(f"{'='*50}")
print(f"监控时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print(f"警报总数: {len(self.alerts)}")
warning_alerts = [a for a in self.alerts if a.current < self.critical_threshold]
critical_alerts = [a for a in self.alerts if a.current >= self.critical_threshold]
print(f"警告警报: {len(warning_alerts)}")
print(f"严重警报: {len(critical_alerts)}")
if self.alerts:
print(f"\n最近警报:")
for alert in self.alerts[-3:]:
print(f" - [{alert.timestamp}] {alert.message}")
# 使用示例
def memory_callback(mem_info):
"""内存监控回调函数"""
if mem_info['percent'] > 75:
print(f"⚠ 注意: 内存使用较高 ({mem_info['percent']:.1f}%)")
def main():
# 创建监控系统
ms = MemoryMonitoringSystem(
warning_threshold=70.0,
critical_threshold=85.0
)
# 模拟内存使用
def memory_workload():
data = []
for i in range(10):
print(f"\n创建大对象 #{i+1}")
data.append([j for j in range(1000000)])
time.sleep(2)
# 清理一部分
print("\n清理部分内存...")
data = data[:3]
time.sleep(2)
return data
# 运行监控（运行30秒）
print("开始内存监控...")
# 在单独的线程中运行监控
monitor_thread = threading.Thread(
target=ms.run_monitoring,
args=(30, memory_callback)
)
monitor_thread.start()
# 执行内存密集型任务
memory_workload()
monitor_thread.join()
print("\n监控完成!")
if __name__ == '__main__':
main()
使用memory_profiler（第三方库）
# 安装: pip install memory_profiler
from memory_profiler import profile
import time
@profile
def memory_hungry_function():
"""使用memory_profiler装饰器进行内存分析"""
# 创建大列表
large_list = [i for i in range(1000000)]
print(f"大列表大小: {len(large_list)}")
# 创建字典
large_dict = {i: str(i) for i in range(500000)}
print(f"大字典大小: {len(large_dict)}")
# 字符串操作
large_string = " ".join([str(i) for i in range(100000)])
# 清理
del large_list
del large_dict
time.sleep(1)
return "完成"
# 运行函数
result = memory_hungry_function()
安装依赖
# 基础依赖 pip install psutil matplotlib # 可选：更高级的内存分析 pip install memory_profiler objgraph
使用建议
- 生产环境监控：使用
MemoryMonitoringSystem结合日志系统 - 开发调试：使用
@memory_profile装饰器分析函数内存 - 实时监控：使用
ContinuousMonitor可视化内存变化 - 性能优化：使用
memory_profiler的@profile装饰器
这些案例涵盖了从简单到复杂的各种内存监控需求,可以根据实际场景选择合适的方案。
