Python内存监控案例如何监控内存占用

wen python案例 2026-06-29 199

本文目录导读：

基础内存监控器
持续内存监控器
函数级内存监控装饰器
完整的内存监控系统
使用memory_profiler（第三方库）
安装依赖
使用建议

我来分享几个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装饰器

这些案例涵盖了从简单到复杂的各种内存监控需求,可以根据实际场景选择合适的方案。

本文地址： https://www.bmjup.com/post/11536.html