读取IO流为大字符串时出现Python Memory Error问题求助

嘿，这个内存错误的根源其实很明显——你现在的写法是把整个IO流的内容一次性全部加载到内存里：先是iostream.getvalue()把所有字节读出来，解码成字符串后又塞进io.StringIO里再读一遍，等于内存里存了至少两份完整的大数据，流稍微大一点内存直接就爆了。

咱们换个思路，不要一次性加载全部数据，用逐行处理的方式来搞定，内存占用会低很多：

方案一：逐行读取原流，避免全量加载 #

如果你的iostream是字节流（比如io.BytesIO），先把它包装成文本流，然后逐行读取处理，这样内存里只会保留当前行的内容：

import io

# 把字节流包装成UTF-8编码的文本流（原流已是文本流可跳过）
text_stream = io.TextIOWrapper(iostream, encoding='utf-8')
split_lines = []

# 逐行读取，避免一次性加载整个流
for line in text_stream:
    # 按需处理换行符，这里去掉末尾的换行符
    split_lines.append(line.rstrip('\n'))

# 插入对应的表头
if name == 'a':
    split_lines.insert(0, 'a_ID,a_VALUE')
elif name == 'b':
    split_lines.insert(0, 'b_ID,b_VALUE')

# 最后合并成字符串（如果必须要的话），或者直接写入新IO流
data_new_headers_str = '\n'.join(split_lines)
data_new_headers_io_str = io.StringIO(data_new_headers_str)

方案二：用生成器进一步降低内存占用 #

如果数据量特别大，连把所有行存在列表里都嫌占内存，可以用生成器来逐行输出，全程只占一行数据的内存：

import io

def process_stream(stream, name):
    # 先输出表头
    if name == 'a':
        yield 'a_ID,a_VALUE'
    elif name == 'b':
        yield 'b_ID,b_VALUE'
    
    # 逐行读取原流并输出
    text_stream = io.TextIOWrapper(stream, encoding='utf-8')
    for line in text_stream:
        yield line.rstrip('\n')

# 把生成器的内容写入新的IO流
data_new_headers_io_str = io.StringIO()
for line in process_stream(iostream, name):
    data_new_headers_io_str.write(f"{line}\n")

# 记得把IO流指针移到开头，方便后续读取
data_new_headers_io_str.seek(0)

额外注意事项 #

• 如果原IO流是不可重复读取的（比如网络流、管道流），逐行处理是最安全的方式；
• 如果原流是可重复读取的（比如BytesIO），处理完后可以调用iostream.seek(0)重置指针，方便后续复用。

核心逻辑就是把“一次性加载全量数据”改成“按需逐行处理”，这样不管你的IO流多大，内存消耗都能控制在极小的范围内，自然就不会出现内存错误了。

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