如何在Python中实现1秒内快速写入1GB文件至磁盘？

当前现状 #

你当前通过一次性生成1GB字符串并写入的方式，耗时4.8秒，吞吐量约209MB/s。结合你的硬件配置（Intel Xeon E3-1505M v5 + 64GB内存 + 本地磁盘），可以通过以下方案优化到1秒内完成写入：

1. 替换字符串为二进制字节串，跳过编码开销 #

用"a"*1GB生成超大字符串会产生额外的内存拷贝和文本编码开销，换成二进制字节串+二进制写入模式，能直接减少CPU消耗：

import time
# 生成1GB二进制字节串，避免字符串编码步骤
chunk = b"a" * 1024 * 1024 * 1024
t0 = time.time()
with open("test.txt", "wb") as f:
    f.write(chunk)
d = time.time() - t0
print(f"duration: {d:.2f} s.")

2. 手动增大缓冲区+分块写入 #

Python默认文件缓冲区较小，手动设置大缓冲区（如64MB/128MB），同时分块写入能减少用户态到内核态的切换次数，提升写入效率：

import time
block_size = 64 * 1024 * 1024  # 64MB块大小
total_size = 1024 * 1024 * 1024
t0 = time.time()
with open("test.txt", "wb", buffering=block_size) as f:
    remaining = total_size
    block = b"a" * block_size
    while remaining > 0:
        write_size = min(block_size, remaining)
        f.write(block[:write_size])
        remaining -= write_size
d = time.time() - t0
print(f"duration: {d:.2f} s.")

3. 用内存映射（mmap）实现零拷贝写入 #

内存映射可以让文件操作直接在内存层面完成，跳过用户态缓存的多次数据拷贝，大幅提升速度：

import time
import mmap
import os

total_size = 1024 * 1024 * 1024
t0 = time.time()
# 先预分配文件空间
with open("test.txt", "wb") as f:
    f.truncate(total_size)
# 内存映射文件，直接写入内存地址
with open("test.txt", "r+b") as f:
    with mmap.mmap(f.fileno(), length=total_size, access=mmap.ACCESS_WRITE) as mm:
        mm[:] = b"a" * total_size
d = time.time() - t0
print(f"duration: {d:.2f} s.")

4. 硬件与系统层面优化 #

• 确认磁盘类型：如果是HDD，连续写入速度通常在100-200MB/s，很难突破1秒；换成NVMe SSD，连续写入速度可达3000MB/s以上，完全满足需求。
• 优化文件系统：Linux用ext4/xfs并开启noatime挂载选项；Windows用NTFS，确保预分配功能正常。
• 调整系统缓存参数：Linux可临时调高vm.dirty_ratio（如设为40），让系统缓存更多写入数据再批量刷盘，注意这会增加断电数据丢失风险。

5. 多进程分块写入（仅HDD适用） #

机械硬盘可通过多进程并行写入利用多磁头特性，SSD单线程即可跑满速度，多进程反而会增加开销：

import time
import multiprocessing as mp

block_size = 64 * 1024 * 1024
total_size = 1024 * 1024 * 1024
num_blocks = total_size // block_size

def write_block(start):
    with open("test.txt", "r+b") as f:
        f.seek(start)
        f.write(b"a" * block_size)

if __name__ == "__main__":
    # 预分配文件空间
    with open("test.txt", "wb") as f:
        f.truncate(total_size)
    t0 = time.time()
    pool = mp.Pool(processes=4)
    pool.map(write_block, [i*block_size for i in range(num_blocks)])
    pool.close()
    pool.join()
    d = time.time() - t0
    print(f"duration: {d:.2f} s.")

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