如何修复numpy.genfromtxt()引发的内存泄漏问题？

问题背景 #

程序通过numpy.genfromtxt()持续读取CSV文件生成浮点数组，运行数天后内存占用飙升至数GB。垃圾回收检测到LineSplitter对象及其绑定方法无法被回收，且随程序运行不断累积。

根因分析 #

numpy.genfromtxt内部依赖LineSplitter类处理CSV行分割，该类实例存在引用循环或内部缓存机制，导致垃圾回收无法清理。加上进程池worker长期存活，未回收的对象持续占用内存，最终引发内存泄漏。

解决方案 #

方案1：替换genfromtxt为loadtxt #

numpy.loadtxt是轻量型CSV读取函数，功能满足你的场景（读取浮点数组、跳过首尾行、指定列），且内部实现不会产生LineSplitter相关泄漏问题。

修改Data_Collection方法：

def Data_Collection(self):
    # 先获取文件总行数，实现skip_footer效果
    with open(self.file_to_read, 'r') as f:
        total_lines = sum(1 for _ in f)
    read_rows = total_lines - 2 - 2  # 减去skip_header和skip_footer的行数

    raw_ar = np.loadtxt(
        self.file_to_read,
        delimiter=",",
        skiprows=2,
        max_rows=read_rows,
        usecols=self.arguments,
        dtype=np.float64,
        fill_values=1
    ).T

    # 直接取值避免不必要拷贝
    ar1 = (float(raw_ar[0][0]), float(raw_ar[0][-1]))
    
    d_df = np.delete(arr=raw_ar, obj=0, axis=0)
    
    # 用numpy切片高效拆分数组
    ddf = d_df[::2, :]
    dtf = d_df[1::2, :]
    
    tpl = (ddf, dtf)
    
    return (ar1, tpl)

方案2：优化genfromtxt调用并强制垃圾回收 #

若必须保留genfromtxt，可在每次调用后显式触发垃圾回收，同时减少不必要的内存拷贝：

import gc

def Data_Collection(self):
    raw_ar = np.genfromtxt(
        self.file_to_read,
        delimiter=",",
        skip_header=2,
        skip_footer=2,
        usecols=self.arguments,
        loose=True,
        dtype=np.float64,
        filling_values=1
    ).T

    # 直接访问原数组元素，避免copy
    ar1 = (float(raw_ar[0][0]), float(raw_ar[0][-1]))
    
    d_df = np.delete(arr=raw_ar, obj=0, axis=0)
    
    # 切片替代列表推导式+append，减少内存开销
    ddf = d_df[::2, :]
    dtf = d_df[1::2, :]
    
    tpl = (ddf, dtf)
    
    # 显式删除引用并触发垃圾回收
    del raw_ar, d_df
    gc.collect()
    
    return (ar1, tpl)

方案3：定期重启进程池 #

进程池worker长期存活会累积内存垃圾，可定期重启进程池清理：

@classmethod
async def StandingBy(cls):
    global pooller
    task_count = 0
    MAX_TASKS_PER_WORKER = 100  # 每执行100次任务重启一次进程池
    
    while True:
        try:
            heart = AAA.Spine(cls.arguments)
            childp = await asyncio.wrap_future(pooller.submit(heart.Rendering, True))
            del childp
            del heart
            
            task_count += 1
            if task_count >= MAX_TASKS_PER_WORKER:
                pooller.shutdown()
                pooller = confu.ProcessPoolExecutor(2)
                task_count = 0

        except OSError as err:
            # 错误处理逻辑
        await asyncio.sleep(some)

额外优化建议 #

• 避免使用列表推导式+append拆分数组，改用numpy切片操作更高效。
• 移除不必要的raw_ar.copy()调用，直接访问原数组元素减少内存拷贝。

内容的提问来源于stack exchange，提问作者ericclaptoso