Python使用requests库实现视频流式下载与多线程优化

温以宁

发布时间：2026-10-13 | 6人浏览过

原创

本文详细解析如何使用Python的requests库进行视频文件的流式下载，解决内存占用问题。通过引入contextlib.closing处理上下文管理器缺失，并优化文件写入策略以实时保存数据。最后结合多线程与队列机制，实现高效的多线程并发下载，包含断点判断与进度显示功能。

窗外细雨蒙蒙，作为一名单身程序员，我在浏览知乎时偶然发现了关于“你都用Python来做什么？”的高赞回答，其中分享了一些实用的技巧。

视频流式下载基础实现

查看该回答中的链接，发现视频地址均为明文，于是决定立即动手尝试下载。

要下载流式文件，只需在requests库的请求中将stream参数设置为True即可，相关文档可查阅官方说明。

首先选取一个视频地址进行测试：


# -*- coding: utf-8 -*-
import requests 
def download_file(url, path):
  with requests.get(url, stream=True) as r:
    chunk_size = 1024
    content_size = int(r.headers['content-length'])
    print '下载开始'
    with open(path, "wb") as f:
      for chunk in r.iter_content(chunk_size=chunk_size):
        f.write(chunk) 
if __name__ == '__main__':
  url = '就在原帖...'
  path = '想存哪都行'
  download_file(url, path)

然而，程序运行后遇到了报错：


AttributeError: __exit__

这是否意味着文档存在误导？

处理上下文管理器缺失问题

经分析，发现是未实现上下文管理器所需的__exit__方法。既然目的是确保请求对象r在结束后能正确关闭以释放连接池，可以使用contextlib库的closing特性来替代：


# -*- coding: utf-8 -*-
import requests
from contextlib import closing
 
def download_file(url, path):
  with closing(requests.get(url, stream=True)) as r:
    chunk_size = 1024
    content_size = int(r.headers['content-length'])
    print '下载开始'
    with open(path, "wb") as f:
      for chunk in r.iter_content(chunk_size=chunk_size):
        f.write(chunk)

优化文件写入与内存管理

程序现已正常运行，但观察文件发现其大小并未变化，无法得知下载进度。为了节省内存并让内容及时写入硬盘，需要优化写入策略：


# -*- coding: utf-8 -*-
import requests
from contextlib import closing
import os
 
def download_file(url, path):
  with closing(requests.get(url, stream=True)) as r:
    chunk_size = 1024
    content_size = int(r.headers['content-length'])
    print '下载开始'
    with open(path, "wb") as f:
      for chunk in r.iter_content(chunk_size=chunk_size):
        f.write(chunk)
        f.flush()
        os.fsync(f.fileno())

此时文件大小以肉眼可见的速度增加。虽然心疼硬盘空间，但为了减少磁盘IO次数，可以选择仅在最后写入，或在程序中记录下载数量即可：


def download_file(url, path):
  with closing(requests.get(url, stream=True)) as r:
    chunk_size = 1024
    content_size = int(r.headers['content-length'])
    print '下载开始'
    with open(path, "wb") as f:
      n = 1
      for chunk in r.iter_content(chunk_size=chunk_size):
        loaded = n*1024.0/content_size
        f.write(chunk)
        print '已下载{0:%}'.format(loaded)
        n += 1

最终结果直观展示了下载进度：


已下载2.579129%
已下载2.581255%
已下载2.583382%
已下载2.585508%

封装下载类以提升复用性

不满足于单一功能的实现，我将逻辑封装为一个类以便复用：


# -*- coding: utf-8 -*-
import requests
from contextlib import closing
import time 
def download_file(url, path):
  with closing(requests.get(url, stream=True)) as r:
    chunk_size = 1024*10
    content_size = int(r.headers['content-length'])
    print '下载开始'
    with open(path, "wb") as f:
      p = ProgressData(size = content_size, unit='Kb', block=chunk_size)
      for chunk in r.iter_content(chunk_size=chunk_size):
        f.write(chunk)
        p.output()
 
 
class ProgressData(object):
 
  def __init__(self, block,size, unit, file_name='', ):
    self.file_name = file_name
    self.block = block/1000.0
    self.size = size/1000.0
    self.unit = unit
    self.count = 0
    self.start = time.time()
  def output(self):
    self.end = time.time()
    self.count += 1
    speed = self.block/(self.end-self.start) if (self.end-self.start)>0 else 0
    self.start = time.time()
    loaded = self.count*self.block
    progress = round(loaded/self.size, 4)
    if loaded >= self.size:
      print u'%s下载完成\r\n'%self.file_name
    else:
      print u'{0}下载进度{1:.2f}{2}/{3:.2f}{4} 下载速度{5:.2%} {6:.2f}{7}/s'.\
         format(self.file_name, loaded, self.unit,\
         self.size, self.unit, progress, speed, self.unit)
      print '%50s'%('/'*int((1-progress)*50))

运行该封装类：


下载开始
下载进度10.24Kb/120174.05Kb 0.01% 下载速度4.75Kb/s
/////////////////////////////////////////////////
下载进度20.48Kb/120174.05Kb 0.02% 下载速度32.93Kb/s
/////////////////////////////////////////////////

代码结构看起来更加整洁美观。

多线程并发下载实现

接下来实现多线程同时下载，主线程负责生产URL并放入队列，下载线程从队列获取URL进行下载：


# -*- coding: utf-8 -*-
import requests
from contextlib import closing
import time
import Queue
import hashlib
import threading
import os 
def download_file(url, path):
  with closing(requests.get(url, stream=True)) as r:
    chunk_size = 1024*10
    content_size = int(r.headers['content-length'])
    if os.path.exists(path) and os.path.getsize(path)>=content_size:
      print '已下载'
      return
    print '下载开始'
    with open(path, "wb") as f:
      p = ProgressData(size = content_size, unit='Kb', block=chunk_size, file_name=path)
      for chunk in r.iter_content(chunk_size=chunk_size):
        f.write(chunk)
        p.output()
 
class ProgressData(object):
 
  def __init__(self, block,size, unit, file_name='', ):
    self.file_name = file_name
    self.block = block/1000.0
    self.size = size/1000.0
    self.unit = unit
    self.count = 0
    self.start = time.time()
  def output(self):
    self.end = time.time()
    self.count += 1
    speed = self.block/(self.end-self.start) if (self.end-self.start)>0 else 0
    self.start = time.time()
    loaded = self.count*self.block
    progress = round(loaded/self.size, 4)
    if loaded >= self.size:
      print u'%s下载完成\r\n'%self.file_name
    else:
      print u'{0}下载进度{1:.2f}{2}/{3:.2f}{4} {5:.2%} 下载速度{6:.2f}{7}/s'.\
         format(self.file_name, loaded, self.unit,\
         self.size, self.unit, progress, speed, self.unit)
      print '%50s'%('/'*int((1-progress)*50))
 queue = Queue.Queue() 
def run():
  while True:
    url = queue.get(timeout=100)
    if url is None:
      print u'全下完啦'
      break
    h = hashlib.md5()
    h.update(url)
    name = h.hexdigest()
    path = 'e:/download/' + name + '.mp4'
    download_file(url, path) 
def get_url():
  queue.put(None)
if __name__ == '__main__':
  get_url()
  for i in xrange(4):
    t = threading.Thread(target=run)
    t.daemon = True
    t.start()

增加了重复下载的判断逻辑。至于如何持续生成URL，大家可以自行探索，注意保重身体！

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持本站。