python - 使用多个并行线程分部分下载大文件

Question

我有一个用例，需要使用多个线程分部分下载大型远程文件。每个线程必须同时(并行)运行，获取文件的特定部分。期望是在成功下载所有部分后将这些部分组合成一个(原始)文件。

也许使用请求库可以完成这项工作，但我不确定如何将其多线程化为将 block 组合在一起的解决方案。

url = 'https://url.com/file.iso'
headers = {"Range": "bytes=0-1000000"}  # first megabyte
r = get(url, headers=headers)

我还考虑使用curl，让Python 来编排下载，但我不确定这是正确的方法。它似乎太复杂了，并且偏离了普通的 Python 解决方案。像这样的事情:

curl --range 200000000-399999999 -o file.iso.part2

有人可以解释一下你会如何处理这样的事情吗？或者发布一个在 Python 3 中工作的代码示例？我通常很容易找到与 Python 相关的答案，但这个问题的解决方案似乎让我难以理解。

Answer 1

这是一个使用 Python 3 和 Asyncio 的版本，它只是一个示例，它可以改进，但您应该能够获得您需要的一切。

• get_size:发送 HEAD 请求获取文件大小
• download_range:下载单个 block
• download:下载所有 block 并合并它们

import asyncio
import concurrent.futures
import functools
import requests
import os


# WARNING:
# Here I'm pointing to a publicly available sample video.
# If you are planning on running this code, make sure the
# video is still available as it might change location or get deleted.
# If necessary, replace it with a URL you know is working.
URL = 'https://download.samplelib.com/mp4/sample-30s.mp4'
OUTPUT = 'video.mp4'


async def get_size(url):
    response = requests.head(url)
    size = int(response.headers['Content-Length'])
    return size


def download_range(url, start, end, output):
    headers = {'Range': f'bytes={start}-{end}'}
    response = requests.get(url, headers=headers)

    with open(output, 'wb') as f:
        for part in response.iter_content(1024):
            f.write(part)


async def download(run, loop, url, output, chunk_size=1000000):
    file_size = await get_size(url)
    chunks = range(0, file_size, chunk_size)

    tasks = [
        run(
            download_range,
            url,
            start,
            start + chunk_size - 1,
            f'{output}.part{i}',
        )
        for i, start in enumerate(chunks)
    ]

    await asyncio.wait(tasks)

    with open(output, 'wb') as o:
        for i in range(len(chunks)):
            chunk_path = f'{output}.part{i}'

            with open(chunk_path, 'rb') as s:
                o.write(s.read())

            os.remove(chunk_path)


if __name__ == '__main__':
    executor = concurrent.futures.ThreadPoolExecutor(max_workers=3)
    loop = asyncio.new_event_loop()
    run = functools.partial(loop.run_in_executor, executor)

    asyncio.set_event_loop(loop)

    try:
        loop.run_until_complete(
            download(run, loop, URL, OUTPUT)
        )
    finally:
        loop.close()