如何实现多进程并行下载并直接写入同一文件，省略分片合并步骤

问题根因 #

你之前的代码写入失败的核心原因是：多进程间共享同一个Python文件对象时，底层的文件偏移量是内核级共享的，多个进程同时执行seek、write操作会互相干扰写入位置，最终导致文件内容混乱。

实现思路 #

• 主进程仅负责预创建和目标文件大小一致的空白文件，不向子进程传递打开的文件句柄
• 每个下载子进程独立以r+b模式打开目标文件，持有独立的文件描述符，文件偏移量互不干扰
• 各子进程负责写入的字节区间完全不重叠，无需额外加锁，不会出现写入冲突

完整实现代码 #

import requests
import multiprocessing
import os

class my_download(object): 
    def __init__(self,url):
        self.url = url
        self.process_num = multiprocessing.cpu_count()
        self.fn = url.split('/')[-1]
        url_headers = requests.head(self.url)
        self.size = int(url_headers.headers['Content-Length'])
        self.ranges = self.get_file_range()
        # 预创建指定大小的空白文件
        with open(self.fn,'wb') as f:
            f.truncate(self.size)

    def get_file_range(self):
        ranges = []
        download_num = int(self.size/self.process_num)
        for i in range(self.process_num):
            if i == self.process_num-1:
                ranges.append((download_num*i,''))
            else:
                ranges.append((download_num*i,download_num*(i+1)))
        return ranges

    def run_task(self,i):
        print(f'process {i} start')
        start_offset = self.ranges[i][0]
        headers={'Range': f'Bytes={self.ranges[i][0]}-{self.ranges[i][1]}','Accept-Encoding':'*'}
        r = requests.get(self.url, headers=headers, stream=True)
        # 子进程独立打开文件，持有独立文件描述符
        with open(self.fn, 'r+b') as f:
            f.seek(start_offset)
            for chunk in r.iter_content(chunk_size=1024*1024): # 调大chunk size提升写入效率
                if chunk:
                    f.write(chunk)
        print(f'process {i} end')

    def run(self):
        pool = multiprocessing.Pool(processes = self.process_num)
        for i in range(self.process_num):
            pool.apply_async(self.run_task,args = (i,))
        pool.close()
        pool.join()


if __name__ == "__main__":
    url = "https://chuangtzu.ftp.acc.umu.se/debian-cd/current/amd64/iso-dvd/debian-11.0.0-amd64-DVD-1.iso"
    downloader = my_download(url)
    downloader.run()

优化建议 #

• 可以将chunk size调大到1MB~4MB，减少IO次数提升写入效率
• 可增加下载进度统计、断点续传、异常重试逻辑提升鲁棒性
• 如果需要支持跨平台运行，注意文件路径的兼容性处理

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