使用numpy memmap时如何避免不必要的数据缓存？

我用numpy memmap分块读取超大（100GB到TB级）二进制文件，程序只做单次遍历，不需要缓存任何数据，但np.memmap默认会缓存数据，导致内存很快占满。能不能关闭这个缓存？如果不行，有没有更好的手动清理缓存的方法？目前我试过的方法是对memmap执行flush、删除所有引用、运行垃圾回收器（gc）再重新创建memmap，虽然有效但不够理想。

下面是最小可复现示例（MWE）：运行后会生成2GB随机数据，能看到哪怕chunk_size很小，内存占用还是会随着加载的数据量累积增加，理想状态下内存应该只占用单个数据块的大小。

import numpy as np
import os
import psutil
import gc
import time

# 参数设置
filename = 'test_memmap.bin'
file_size_gb = 2  # 可按需修改
dtype = np.float32
element_size = np.dtype(dtype).itemsize
num_elements = (file_size_gb * 1024**3) // element_size
chunk_size = 1_000_000  # 单次读取的元素数量

# 步骤1：生成大二进制文件
if not os.path.exists(filename):
    print("正在创建文件...")
    with open(filename, 'wb') as f:
        f.write(np.random.rand(num_elements).astype(dtype).tobytes())

# 步骤2：用memmap分块处理文件
print("使用memmap处理文件...")
mm = np.memmap(filename, dtype=dtype, mode='r')
process = psutil.Process(os.getpid())

for i in range(0, len(mm), chunk_size):
    chunk = mm[i:i+chunk_size]
    # 模拟处理操作
    chunk.sum()
    
    # 监控内存使用情况
    mem = process.memory_info().rss / (1024 ** 2)  # 单位：MB
    print(f"第 {i // chunk_size + 1} 步，内存占用：{mem:.2f} MB")
del mm
gc.collect()
time.sleep(5) # 系统需要一点时间更新内存状态
mem = process.memory_info().rss / (1024 ** 2)  # 单位：MB
print(f"删除memmap后的最终内存占用：{mem:.2f} MB")

1. 关闭缓存的可行性 #

np.memmap依赖操作系统的虚拟内存机制，默认的缓存是系统层面的页缓存，numpy没有直接提供关闭缓存的API。但可以通过调用系统API告知系统不要缓存数据，或者换用不依赖系统缓存的读取方式。

2. 高效的缓存控制方法 #

方法一：用posix_fadvise告知系统释放已处理数据（Linux/macOS） #

在处理完每个数据块后，调用系统函数告诉操作系统不需要缓存这块数据，系统会及时释放对应的页缓存，内存占用能稳定在单个chunk大小附近。

修改循环部分的代码：

import ctypes
from ctypes import cdll

# 加载系统库（Linux用下面这句，macOS替换为libSystem.dylib）
libc = cdll.LoadLibrary('libc.so.6')
# libc = cdll.LoadLibrary('libSystem.dylib')  # macOS环境启用

POSIX_FADV_DONTNEED = 4  # 告知系统不需要缓存该区域

for i in range(0, len(mm), chunk_size):
    chunk = mm[i:i+chunk_size]
    chunk.sum()
    
    # 通知系统释放当前chunk的缓存
    libc.posix_fadvise(mm.fileno(), i*element_size, chunk_size*element_size, POSIX_FADV_DONTNEED)
    
    mem = process.memory_info().rss / (1024 ** 2)
    print(f"第 {i // chunk_size + 1} 步，内存占用：{mem:.2f} MB")

方法二：分块创建memmap，用完即销毁 #

不用维持全局memmap对象，每次只映射当前需要的chunk区域，处理完立即删除引用并触发垃圾回收，避免累积缓存：

# 替换原步骤2的代码
print("使用分块memmap处理文件...")
process = psutil.Process(os.getpid())

for i in range(0, num_elements, chunk_size):
    # 仅映射当前需要的chunk
    mm_chunk = np.memmap(
        filename,
        dtype=dtype,
        mode='r',
        offset=i*element_size,
        shape=(min(chunk_size, num_elements - i),)
    )
    # 模拟处理
    mm_chunk.sum()
    
    # 立即清理当前chunk的映射
    del mm_chunk
    gc.collect()
    
    mem = process.memory_info().rss / (1024 ** 2)
    print(f"第 {i // chunk_size + 1} 步，内存占用：{mem:.2f} MB")

方法三：直接用文件对象读取（绕过系统缓存） #

如果可以接受手动处理数据对齐，用os.open配合O_DIRECT标志打开文件，直接读取二进制数据并转换为numpy数组，完全绕过系统缓存：

# 替换原步骤2的代码
print("使用直接文件读取处理...")
process = psutil.Process(os.getpid())
block_size = os.stat(filename).st_blksize  # 获取文件系统块大小

with os.open(filename, os.O_RDONLY | os.O_DIRECT) as fd:
    for i in range(0, num_elements, chunk_size):
        # 计算对齐后的偏移和读取大小（O_DIRECT要求偏移、读取大小都是块大小的倍数）
        raw_offset = i * element_size
        aligned_offset = (raw_offset // block_size) * block_size
        read_length = min(
            ((chunk_size * element_size + block_size - 1) // block_size) * block_size,
            num_elements * element_size - aligned_offset
        )
        
        # 读取对齐后的数据
        buffer = os.read(fd, read_length)
        # 截取实际需要的chunk数据并转换为numpy数组
        chunk = np.frombuffer(
            buffer[raw_offset - aligned_offset : raw_offset - aligned_offset + chunk_size*element_size],
            dtype=dtype
        )
        # 模拟处理
        chunk.sum()
        
        mem = process.memory_info().rss / (1024 ** 2)
        print(f"第 {i // chunk_size + 1} 步，内存占用：{mem:.2f} MB")

注意：O_DIRECT仅在Linux系统稳定支持，且要求偏移、读取大小必须是文件系统块大小的倍数，否则会报错。

总结 #

• 优先选方法一：改动最小，在原有memmap代码基础上增加一行系统调用，就能有效控制内存，适合Linux/macOS环境；
• 跨平台场景选方法二：逻辑简单，兼容性强，无需依赖系统API；
• 极端内存控制需求选方法三：完全绕过系统缓存，但需要处理数据对齐问题。

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