异步协程开发实战：优化大文件上传与下载的速度

import aioHttp
import asyncio

async def upload_chunk(session, url, file, offset, size):
    headers = {'Content-Length': str(size), 'Content-Range': f'bytes {offset}-{offset+size-1}/{file_size}'}
    data = file.read(size)
    async with session.put(url, headers=headers, data=data) as resp:
        return await resp.JSON()

async def upload_file_with_chunks(session, url, file):
    file_size = os.path.getsize(file.name)
    chunk_size = 1024 * 1024 * 5 #每块数据的大小为5MB
    offset = 0
    tasks = []
    while offset < file_size:
        size = chunk_size if offset+chunk_size < file_size else file_size-offset
        tasks.append(upload_chunk(session, url, file, offset, size))
        offset += size
    return await asyncio.gather(*tasks)

async def main():
    async with aiohttp.ClientSession() as session:
        url = 'http://example.com/upload'
        file = open('large_file.mp4', 'rb')
        result = await upload_file_with_chunks(session, url, file)
        print(result)

asyncio.run(main())

import threading
import requests

class MultiPartUpload(object):
    def __init__(self, url, file_path, num_thread=4):
        self.url = url
        self.file_path = file_path
        self.num_thread = num_thread
        self.file_size = os.path.getsize(self.file_path)
        self.chunk_size = self.file_size//num_thread
        self.threads = []
        self.lock = threading.Lock()

    def upload(self, i):
        start = i * self.chunk_size
        end = start + self.chunk_size - 1
        headers = {"Content-Range": "bytes %s-%s/%s" % (start, end, self.file_size),
                   "Content-Length": str(self.chunk_size)}
        data = open(self.file_path, 'rb')
        data.seek(start)
        resp = requests.put(self.url, headers=headers, data=data.read(self.chunk_size))
        self.lock.acquire()
        print("Part %d status: %s" % (i, resp.status_code))
        self.lock.release()

    def run(self):
        for i in range(self.num_thread):
            t = threading.Thread(target=self.upload, args=(i,))
            self.threads.append(t)
        for t in self.threads:
            t.start()

        for t in self.threads:
            t.join()

if __name__ == '__main__':
    url = 'http://example.com/upload'
    file = 'large_file.mp4'
    uploader = MultiPartUpload(url, file)
    uploader.run()

import aiohttp
import asyncio
import os

async def download_chunk(session, url, file, offset, size):
    headers = {'Range': f'bytes={offset}-{offset+size-1}'}
    async with session.get(url, headers=headers) as resp:
        data = await resp.read()
        file.seek(offset)
        file.write(data)
        return len(data)

async def download_file_with_chunks(session, url, file):
    async with session.head(url) as resp:
        file_size = int(resp.headers.get('Content-Length'))
        chunk_size = 1024 * 1024 * 5 #每块数据的大小为5MB
        offset = 0
        tasks = []
        while offset < file_size:
            size = chunk_size if offset+chunk_size < file_size else file_size-offset
            tasks.append(download_chunk(session, url, file, offset, size))
            offset += size
        return await asyncio.gather(*tasks)

async def main():
    async with aiohttp.ClientSession() as session:
        url = 'http://example.com/download/large_file.mp4'
        file = open('large_file.mp4', 'wb+')
        await download_file_with_chunks(session, url, file)

asyncio.run(main())

import threading
import requests

class MultiPartDownload(object):
    def __init__(self, url, file_path, num_thread=4):
        self.url = url
        self.file_path = file_path
        self.num_thread = num_thread
        self.file_size = requests.get(self.url, stream=True).headers.get('Content-Length')
        self.chunk_size = int(self.file_size) // self.num_thread
        self.threads = []
        self.lock = threading.Lock()

    def download(self, i):
        start = i * self.chunk_size
        end = start + self.chunk_size - 1 if i != self.num_thread - 1 else ''
        headers = {"Range": "bytes=%s-%s" % (start, end)}
        data = requests.get(self.url, headers=headers, stream=True)
        with open(self.file_path, 'rb+') as f:
            f.seek(start)
            f.write(data.content)
        self.lock.acquire()
        print("Part %d Downloaded." % i)
        self.lock.release()

    def run(self):
        for i in range(self.num_thread):
            t = threading.Thread(target=self.download, args=(i,))
            self.threads.append(t)
        for t in self.threads:
            t.start()

        for t in self.threads:
            t.join()

if __name__ == '__main__':
    url = 'http://example.com/download/large_file.mp4'
    file = 'large_file.mp4'
    downloader = MultiPartDownload(url, file)
    downloader.run()