Python脚本如何防止进程内存持续增长

wen python案例 2026-06-27 28

本文目录导读：

使用弱引用打破循环引用
显式删除大对象
使用生成器替代列表
手动管理垃圾回收
使用 __slots__ 减少内存开销
使用内存分析工具监控
使用 array 或 numpy 替代列表
使用 mmap 处理大文件
完整的内存管理示例
监控和日志

防止Python进程内存持续增长,通常需要从对象生命周期管理、数据结构选择、资源释放以及内存分析工具几个方面入手，以下是系统性的解决方案：

使用弱引用打破循环引用

import weakref
import gc
class Node:
    def __init__(self, value):
        self.value = value
        self.parent = None  # 使用弱引用
        self.children = []
    def set_parent(self, parent_node):
        # 使用弱引用避免循环引用
        self.parent = weakref.ref(parent_node)
# 手动触发垃圾回收
gc.collect()

显式删除大对象

def process_large_data():
    # 处理完立即释放
    large_list = [i for i in range(10**7)]
    result = sum(large_list)
    # 显式删除
    del large_list
    return result
# 或者使用 with 语句管理上下文
class LargeDataProcessor:
    def __enter__(self):
        self.data = self.load_large_data()
        return self
    def __exit__(self, *args):
        # 退出时自动清理
        del self.data
        gc.collect()

使用生成器替代列表

# 错误示例：一次性加载所有数据
def load_all_data():
    return [expensive_operation(i) for i in range(10**6)]
# 正确示例：使用生成器
def stream_data():
    for i in range(10**6):
        yield expensive_operation(i)
# 使用
for item in stream_data():
    process(item)

手动管理垃圾回收

import gc
class MemoryManager:
    def __init__(self):
        # 关闭自动垃圾回收
        gc.disable()
        # 设置阈值
        gc.set_threshold(700, 10, 5)
    def process_batch(self, batch_size=1000):
        for i in range(batch_size):
            # 处理数据
            result = expensive_operation(i)
            # 每处理100个对象，手动触发一次回收
            if i % 100 == 0:
                gc.collect(generation=0)  # 只回收第0代
        # 批次结束后完全回收
        gc.collect()

使用 `slots` 减少内存开销

# 普通类（每个实例有 __dict__）
class RegularPoint:
    def __init__(self, x, y):
        self.x = x
        self.y = y
# 使用 __slots__（没有 __dict__，内存更小）
class SlottedPoint:
    __slots__ = ['x', 'y']
    def __init__(self, x, y):
        self.x = x
        self.y = y
# 内存对比
import sys
regular = RegularPoint(1, 2)
slotted = SlottedPoint(1, 2)
print(f"Regular: {sys.getsizeof(regular)} bytes")  # 56
print(f"Slotted: {sys.getsizeof(slotted)} bytes")  # 40

使用内存分析工具监控

import tracemalloc
import objgraph
class MemoryMonitor:
    def __init__(self):
        tracemalloc.start()
    def get_memory_snapshot(self):
        snapshot = tracemalloc.take_snapshot()
        top_stats = snapshot.statistics('lineno')
        print("[ Top 10 memory consumers ]")
        for stat in top_stats[:10]:
            print(stat)
    def find_leaks(self):
        # 找出对象增长
        growth = objgraph.growth(limit=10)
        print("Object growth analysis:")
        for obj_type, count, delta in growth:
            print(f"{obj_type}: {count} (+{delta})")
# 使用
monitor = MemoryMonitor()
# 运行你的代码
monitor.get_memory_snapshot()

使用 `array` 或 `numpy` 替代列表

import array
import numpy as np
# 使用 array 模块
def process_with_array(size=10**6):
    arr = array.array('d', [0.0]) * size  # 双精度浮点数数组
    for i in range(size):
        arr[i] = i * 1.5
    return arr
# 使用 numpy（更高效）
def process_with_numpy(size=10**6):
    arr = np.zeros(size, dtype=np.float64)
    arr[:] = np.arange(size) * 1.5
    return arr

使用 `mmap` 处理大文件

import mmap
import os
def process_large_file(filename):
    with open(filename, 'r') as f:
        with mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) as mm:
            # 直接处理内存映射，不加载到Python内存
            for line in iter(mm.readline, b''):
                process(line)
            # mmap 自动释放

完整的内存管理示例

import gc
import weakref
from functools import lru_cache
class MemorySafeProcessor:
    def __init__(self, max_cache_size=100):
        gc.enable()
        self._cache = {}
        self._max_cache_size = max_cache_size
        self._memory_usage = 0
    @lru_cache(maxsize=50)
    def cached_operation(self, key):
        """带缓存的耗时操作"""
        return expensive_computation(key)
    def process_batch(self, items):
        results = []
        for i, item in enumerate(items):
            result = self.cached_operation(item)
            results.append(result)
            # 定期清理
            if i % 100 == 0:
                self._cleanup()
        return results
    def _cleanup(self):
        # 清除过大的缓存
        if len(self._cache) > self._max_cache_size:
            self._cache.clear()
        # 检查内存使用
        import psutil
        process = psutil.Process()
        memory_percent = process.memory_percent()
        if memory_percent > 80:  # 如果内存使用超过80%
            gc.collect()
            self._cache.clear()
            print(f"Memory cleanup triggered: {memory_percent}%")
    def __del__(self):
        # 清理资源
        self._cache.clear()
        gc.collect()
# 使用示例
processor = MemorySafeProcessor()
results = processor.process_batch(range(10000))

监控和日志

import psutil
import logging
from memory_profiler import profile
logging.basicConfig(level=logging.INFO)
@profile  # 装饰器监控函数内存
def monitored_function():
    import numpy as np
    # 模拟内存增长
    data = []
    for i in range(1000):
        data.append(np.random.rand(1000, 1000))
        # 内存监控
        process = psutil.Process()
        memory_mb = process.memory_info().rss / 1024 / 1024
        if memory_mb > 500:  # 超过500MB报警
            logging.warning(f"Memory usage: {memory_mb:.2f} MB")
            # 清理
            del data[:]
            break
    return "Done"
# 运行监控
monitored_function()

策略	适用场景	效果
弱引用	循环引用问题	防止内存泄漏
生成器	大数据流处理	即时内存使用
`__slots__`	大量小对象	减少30-50%内存
array/numpy	数值计算	高效存储
mmap	大文件处理	几乎不占内存
gc.collect()	手动控制	及时回收
分析工具	定位问题	预防性维护