从Google Cloud Storage Blob下载文件时的Python内存泄漏问题

我开发了一个从GCS存储桶下载文件的函数，供线程用于下载小于150KB的压缩HTML文件并解析。问题在于，经过数千次下载后内存呈线性增长，这对于需要处理数百万文件的场景来说会导致内存不足。

函数代码 #

def download_html(self, md5_hash):
    filename = f'{md5_hash}.html.gz'
    gcs_object_key = f'{self.gcs_prefix}/pages/{filename}'

    try:
        blob = self.gcs_bucket.blob(gcs_object_key)

        # same result for gz_content = blob.download_as_bytes()
        with blob.open('rb') as f:
            gz_content = f.read()

        data = gzip.decompress(gz_content)
        del gz_content
        del blob
        return data
    except:
        return None

该函数被另一个函数调用，后者仅解码获取到的HTML字节数据并重复处理下一个文件。

内存分析结果 #

我使用tracemalloc对程序进行了性能分析，发现GCS的blob文件中存在持续增长的内存分配：

lib/python3.8/site-packages/google/cloud/storage/blob.py:830: size=398 B (+199 B), count=2 (+1), average=199 B
lib/python3.8/site-packages/google/cloud/storage/blob.py:830: size=597 B (+398 B), count=3 (+2), average=199 B
lib/python3.8/site-packages/google/cloud/storage/blob.py:830: size=796 B (+597 B), count=4 (+3), average=199 B
...

内存分配随每次迭代持续增长，且似乎从未释放。

我尝试在每次使用后删除blob并调用gc.collect()，但毫无效果。我还尝试始终下载同一个文件，此时内存会在某个点稳定下来不再增长，这让我怀疑存在文件缓存导致内存增加。仅使用本地文件时无此问题，显然问题来自文件下载过程。

基准测试更新 #

我运行了基准测试，对比下载5000个文件时有无线程的内存使用情况，结果如下：

• 无线程：
  • 初始内存：112.0078125
  • 最终内存：116.48046875
  • 内存差值：4.47265625
• 10个线程：
  • 初始内存：130.22265625
  • 最终内存：162.1328125
  • 内存差值：31.91015625

需要说明的是，我未使用线程池，而是创建了Thread的子类，实现了一个无限循环的run函数，从Redis数据库获取待下载的文件名。目前我并未对下载的文件进行任何处理，每次下载后都会手动调用gc.collect()。

求解决此内存问题的建议。

1. 禁用GCS客户端缓存 #

Google Cloud Storage客户端默认会缓存请求相关数据，这是线程场景下内存增长的核心原因之一。可以通过两种方式禁用：

# 方式1：创建客户端时禁用HTTP缓存
from google.cloud import storage
client = storage.Client(_http=storage.requests.Session())

# 方式2：针对单个blob禁用属性缓存
blob = self.gcs_bucket.blob(gcs_object_key)
blob._properties_cache = {}

2. 改用线程池管理并发 #

自定义无限循环的Thread子类容易导致线程本地资源无法有效回收，改用concurrent.futures.ThreadPoolExecutor可以更好地管控线程生命周期：

from concurrent.futures import ThreadPoolExecutor

def process_single_file(md5_hash):
    data = download_html(md5_hash)
    # 后续解析逻辑...

# 初始化线程池，控制并发数
with ThreadPoolExecutor(max_workers=10) as executor:
    while True:
        md5_hash = redis_client.get_task()
        if not md5_hash:
            break
        executor.submit(process_single_file, md5_hash)

3. 流式处理压缩文件 #

避免一次性将整个gzip文件读入内存，改用流式解压减少内存占用：

import gzip

def download_html(self, md5_hash):
    filename = f'{md5_hash}.html.gz'
    gcs_object_key = f'{self.gcs_prefix}/pages/{filename}'

    try:
        blob = self.gcs_bucket.blob(gcs_object_key)
        with blob.open('rb') as f_in:
            with gzip.GzipFile(fileobj=f_in, mode='rb') as gz_in:
                data = gz_in.read()
        del blob
        return data
    except:
        return None

4. 升级GCS客户端库 #

旧版本的google-cloud-storage可能存在内存泄漏问题，升级到最新稳定版：

pip install --upgrade google-cloud-storage

5. 清理线程本地存储（若坚持自定义线程） #

如果必须使用自定义Thread子类，在每次循环结束后手动清理线程本地资源：

import threading
import gc

class MyThread(threading.Thread):
    def run(self):
        while True:
            md5_hash = redis_client.get_task()
            if not md5_hash:
                break
            try:
                data = self.download_html(md5_hash)
                # 处理数据...
            finally:
                threading.local().__dict__.clear()
                gc.collect()

内容的提问来源于stack exchange，提问作者Franco Lopez Paviolo