Python:使用future，并发下载图片

开局一张图&＃xff08;下载的图片&＃xff09;&＃xff1a;

1. 网络普通下载图片

为了高效处理网络I/O&＃xff0c;需要使用并发&＃xff0c;因为网络有很高的延迟&＃xff0c;所以为了不浪费CPU周期去等待&＃xff0c;最好在收到网络响应之前做些其他的事。

两个示例程序&＃xff0c;从网上下载图片。第一个示例程序是依序下载的&＃xff1a;下载完一个图&＃xff0c;并将其保存在硬盘中之后&＃xff0c;才请求下一个图像。另一个脚本是并发下载的&＃xff1a;几乎同时请求所有图像&＃xff0c;每下载完一个文件就保存一个文件&＃xff0c;脚本使用concurrent.futures模块。

在I/O密集型应用中&＃xff0c;如果代码写得正确&＃xff0c;那么不管使用哪种并发策略&＃xff08;使用线程或asyncio包&＃xff09;&＃xff0c;吞吐量都比依序执行的代码高很多。

这边我改了《流畅的Python》中的下载地址和对象&＃xff1a;

# a5_4_downloadimage.py
import os
import sys
import time
import requestsDOWNNLOAD_DIR &＃61; r&＃39;D:\downloadimage&＃39;
BASE_URL &＃61; &＃39;http://pic2.sc.chinaz.com/Files/pic/pic9/202002/&＃39;
image_list &＃61; [&＃39;zzpic231&＃39; &＃43; str(i) &＃43; &＃39;_s.jpg&＃39; for i in range(10, 90)]def save_image(img, filename):path &＃61; os.path.join(DOWNNLOAD_DIR, filename)with open(path, &＃39;wb&＃39;) as fp:fp.write(img)def get_image(suffix):url &＃61; os.path.join(BASE_URL, suffix)response &＃61; requests.get(url)return response.contentdef show(text):print(text,end&＃61;&＃39;\n&＃39;)sys.stdout.flush()def download_all(image_name_list): # download_all是与并发实现比较的关键函数。for image_name in image_name_list:image &＃61; get_image(image_name)save_image(image, image_name)show(image)return len(image_name_list)def main(download_task):t0 &＃61; time.time()count &＃61; download_task(image_list)elapsed &＃61; time.time() - t0msg &＃61; f&＃39;\n download {count} images in {elapsed}s&＃39;print(msg)if __name__ &＃61;&＃61; &＃39;__main__&＃39;:main(download_all)# download 80 images in 4.6661295890808105s
# download 80 images in 5.478628873825073s
# download 80 images in 4.028514862060547s


2. 使用concurrent.futures模块实现并发下载

concurrent.futures模块的主要特色是 ThreadPoolExecutor 和 ProcessPoolExecutor 类&＃xff0c;这两个类实现的接口能分别在不同的线程或进程中执行可调用的对象。这两个类在内部维护着一个工作线程或进程池&＃xff0c;以及要执行的任务队列。不过&＃xff0c;这个接口抽象的层级很高&＃xff0c;像下载图片这种简单的案例&＃xff0c;无需关心任何实现细节。

使用ThreadPoolExecutor.map方法&＃xff0c;以最简单的方式实现并发下载&＃xff1a;

# a5_4_downloadimage2.py
from concurrent import futures
from a5_4_downloadimage import save_image, get_image, show, mainMAX_WORDERS &＃61; 20 # 设定ThreadPoolExecutor类最多使用几个线程&＃xff1a;并发20个def download_single(image_name):image &＃61; get_image(image_name)save_image(image, image_name)show(image)return image_namedef download_multiple(image_name_list):tasks &＃61; min(MAX_WORDERS, len(image_name_list))with futures.ThreadPoolExecutor(tasks) as executor:res &＃61; executor.map(download_single, sorted(image_name_list))return len(list(res))if __name__ &＃61;&＃61; &＃39;__main__&＃39;:main(download_multiple)# download 80 images in 1.4081335067749023s
# download 80 images in 1.561039924621582s
# download 80 images in 1.393141746520996s


download_multiple 函数中设定工作的线程数量&＃xff1a;使用允许的最大&＃xff08;MAX_WORKERS&＃xff09;与要处理的数量之间较小的那个值&＃xff0c;以免创建多余的线程&＃xff1b;使用工作的线程数实例化ThreadPoolExecutor类&＃xff1b;executor.__exit__ 方法会调用 executor.shutdown(wait&＃61;True) 方法&＃xff0c;它会在所有线程都执行完毕前阻塞线程&＃xff1b;map方法的作用与内置的map函数类似&＃xff0c;不过 download_single 函数会在多个线程中并发调用&＃xff1b;map方法返回一个生成器&＃xff0c;因此可以迭代&＃xff0c;获取各个函数返回的值。最后返回获取的结果数量&＃xff0c;如果有线程抛出异常&＃xff0c;异常会在return语句处抛出&＃xff0c;这与隐式调用 next() 函数从迭代器中获取相应的返回值一样。

download_single 函数其实是前面例子中的 download_all 函数的 for 循环体。编写并发代码时经常这样重构&＃xff1a;把依序执行的for循环体改成函数&＃xff0c;以便并发调用。