Python生成压缩文件性能优化：多线程可行吗？还有哪些方案？

一、优化文本生成逻辑 #

原代码循环逐个生成字符串再拼接的方式效率极低，频繁的format调用和列表append操作会产生大量Python层面的开销。可以通过批量生成文本减少循环次数，同时改用更快的f-string替代str.format：

import gzip
from time import time

def gen_data(chunk_lines=100000):
    total_size = 50 * (1024**3)
    current_size = 0
    n = 0
    while current_size < total_size:
        # 一次性生成上万行文本，减少循环开销
        lines = [f"int {n+i} = {n+i}\n" for i in range(chunk_lines)]
        chunk = "".join(lines).encode('utf-8')
        current_size += len(chunk)
        n += chunk_lines
        yield chunk

二、调整gzip压缩参数 #

gzip默认使用最高压缩级别（level=9），这会极大拖慢压缩速度。可以根据需求降低压缩级别（平衡速度与压缩比），同时增大写入缓冲区减少磁盘IO次数：

def stream_compress():
    start = time()
    # 设置压缩级别为6，缓冲区设为16MB
    with gzip.open("compressed.gz", 'wb', compresslevel=6, buffer_size=16*1024*1024) as f:
        for chunk in gen_data():
            f.write(chunk)
    print(f"Done. Time elapsed is {time() - start:.2f} seconds")

• compresslevel：取值1_{9，级别越低速度越快、压缩比越低，建议测试level=3}6的实际效果
• buffer_size：增大缓冲区可减少磁盘写入的系统调用次数，提升IO效率

三、多线程分离生成与压缩（生产者消费者模式） #

数据生成和压缩是独立操作，可以用多线程让两者并行执行，避免单线程下的串行等待开销：一个线程生成数据（生产者），另一个线程负责压缩写入（消费者），通过队列传递数据：

import gzip
from time import time
from threading import Thread
from queue import Queue

def gen_data(queue, chunk_lines=100000):
    total_size = 50 * (1024**3)
    current_size = 0
    n = 0
    while current_size < total_size:
        lines = [f"int {n+i} = {n+i}\n" for i in range(chunk_lines)]
        chunk = "".join(lines).encode('utf-8')
        current_size += len(chunk)
        n += chunk_lines
        queue.put(chunk)
    # 放入结束标记
    queue.put(None)

def compress_worker(queue):
    with gzip.open("compressed.gz", 'wb', compresslevel=6, buffer_size=16*1024*1024) as f:
        while True:
            chunk = queue.get()
            if chunk is None:
                break
            f.write(chunk)
            queue.task_done()

def stream_compress():
    start = time()
    # 队列大小设为5，避免内存占用过高
    q = Queue(maxsize=5)
    producer = Thread(target=gen_data, args=(q,))
    consumer = Thread(target=compress_worker, args=(q,))
    producer.start()
    consumer.start()
    producer.join()
    consumer.join()
    print(f"Done. Time elapsed is {time() - start:.2f} seconds")

额外提示 #

• 原代码中total_size的计算逻辑存在偏差：text.format(n=n)的长度会随n的位数变化（比如n=9和n=10的字符串长度不同），导致实际生成的总大小和预期50G不符，建议改为按行数控制或实时计算实际字节数
• 如果磁盘IO是瓶颈，可使用SSD存储介质，或临时写入内存文件系统再导出

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