写点什么

这篇博客和你唠唠 python 并发,滚雪球学 python 第四季,第 16 篇

发布于: 刚刚

在 python 编码过程中,有时存在这样的一个需求,同时下载 N 张图片,并且要快。


一般这样的需求,只需要编写一个 for 循环即可实现,但是加上 这个要求,就不好实现了。


图片下载属于 I/O 操作,比较耗时,基于此,可以利用 python 中的多线程将其实现。


为了不让大家学的太困倦,特意找来 6 张美丽的图片,本次学习将围绕这几张图片进行。


https://img-pre.ivsky.com/img/tupian/pre/202102/21/oumei_meinv.jpghttps://img-pre.ivsky.com/img/tupian/pre/202102/21/oumei_meinv-004.jpghttps://img-pre.ivsky.com/img/tupian/pre/202102/21/oumei_meinv-012.jpghttps://img-pre.ivsky.com/img/tupian/pre/202102/21/oumei_meinv-013.jpghttps://img-pre.ivsky.com/img/tupian/pre/202102/21/oumei_meinv-016.jpghttps://img-pre.ivsky.com/img/tupian/pre/202102/21/oumei_meinv-010.jpg
复制代码

单线程下载 6 张图片

使用 for 循环,同步代码如下所示:


import time
import requests
urls = [ "https://img-pre.ivsky.com/img/tupian/pre/202102/21/oumei_meinv.jpg", "https://img-pre.ivsky.com/img/tupian/pre/202102/21/oumei_meinv-004.jpg", "https://img-pre.ivsky.com/img/tupian/pre/202102/21/oumei_meinv-012.jpg", "https://img-pre.ivsky.com/img/tupian/pre/202102/21/oumei_meinv-013.jpg", "https://img-pre.ivsky.com/img/tupian/pre/202102/21/oumei_meinv-016.jpg", "https://img-pre.ivsky.com/img/tupian/pre/202102/21/oumei_meinv-010.jpg"]# 文件保存路径SAVE_DIR = './928/'

def save_img(url): res = requests.get(url) with open(F'{SAVE_DIR}{time.time()}.jpg', 'wb+') as f: f.write(res.content)

if __name__ == '__main__': # 下载开始时间 start_time = time.perf_counter() for url in urls: save_img(url)
print("下载 6 张图片消耗时间为:", time.perf_counter() - start_time) # 下载 6 张图片消耗时间为: 1.911142665
复制代码

concurrent.futures 模块下载 6 张图片

接下来使用 concurrent.futures 模块 实现对 6 张图片的下载,这个模块实现了 ThreadPoolExecutor 类和 ProcessPoolExecutor 类,都继承自Executor,分别被用来创建 线程池进程池,接受 max_workers 参数,代表创建的线程数或者进程数。


这两个类可以在不同的线程或进程中执行 可调用对象ProcessPoolExecutormax_workers 参数可以为空,程序会自动创建与电脑 CPU 数目相同的进程数。


使用 ThreadPoolExecutor 实现多线程下载


import time
import requestsfrom concurrent import futures
MAX_WORKERS = 20 # 最大线程数SAVE_DIR = './928/' # 文件保存路径urls = [ "https://img-pre.ivsky.com/img/tupian/pre/202102/21/oumei_meinv.jpg", "https://img-pre.ivsky.com/img/tupian/pre/202102/21/oumei_meinv-004.jpg", "https://img-pre.ivsky.com/img/tupian/pre/202102/21/oumei_meinv-012.jpg", "https://img-pre.ivsky.com/img/tupian/pre/202102/21/oumei_meinv-013.jpg", "https://img-pre.ivsky.com/img/tupian/pre/202102/21/oumei_meinv-016.jpg", "https://img-pre.ivsky.com/img/tupian/pre/202102/21/oumei_meinv-010.jpg"]

def save_img(url): res = requests.get(url) with open(F'{SAVE_DIR}{time.time()}.jpg', 'wb+') as f: f.write(res.content)

if __name__ == '__main__': start_time = time.perf_counter() # 下载开始时间 with futures.ThreadPoolExecutor(MAX_WORKERS) as executor: res = executor.map(save_img, urls) # executor.map() 方法会返回一个生成器,后续代码可以迭代获取每个线程的执行结果
print("下载 6 张图片消耗时间为:", time.perf_counter() - start_time) # 下载 6 张图片消耗时间为: 0.415939759
复制代码


当使用多线程代码之后,时间从单线程的 1.9s 变为了多线程的 0.4s,能看到效率的提升。

Future 类

在上述多线程代码中,使用了 concurrent 库中的 future 对象,该对象是 Future 类的对象,它的实力表示可能已经完成尚未完成的 延迟计算,该类具备 done() 方法,返回调用对象是否已经执行,有该方法的同时还具备一个 add_done_callback() 方法,表示调用对象执行完毕的回调函数。


from concurrent.futures import ThreadPoolExecutor

def print_name(): return "橡皮擦"

def say_hello(obj): """可调用对象执行完毕,绑定的回调函数""" w_name = obj.result() s = w_name + "你好" print(s) return s

with ThreadPoolExecutor(1) as executor: executor.submit(print_name).add_done_callback(say_hello)
复制代码


在上述代码中用到了如下知识点:


  • executor.map():该方法类似 map 函数,原型为 map(func, *iterables, timeout=None, chunksize=1),异步执行 func,并支持多次并发调用;

  • executor.submit():方法原型为 submit(fn, *args, **kwargs),安排可调用对象 fnfn(*args, **kwargs) 的形式执行,并返回 Future 对象来表示它的执行结果,该方法只能进行单个任务,如果需要并发多个任务,需要使用 map 或者 as_completed

  • future对象.result():返回调用返回的值,有个等待时间参数 timeout 可以设置;

  • future对象add_done_callback():该方法中绑定的回调函数在 future 取消或者完成后运行,表示 future 本身

补充说明

as_completed() 方法该方法参数是一个 Future 列表,返回值是一个 Future 组成的生成器,在调用 as_completed() 方法时不会阻塞,只有当对迭代器进行循环时,每调用一次 next() 方法,如果当前 Future 对象还未完成,则会阻塞。


修改下载 6 张图片的代码,使用 as_completed() 进行实现,最后得到的时间优于单线程,但如果对结果进行迭代,调用 result() 方法,则时间会加长。


import time
import requestsfrom concurrent.futures import ThreadPoolExecutor, as_completed
MAX_WORKERS = 20 # 最大线程数SAVE_DIR = './928/' # 文件保存路径urls = [ "https://img-pre.ivsky.com/img/tupian/pre/202102/21/oumei_meinv.jpg", "https://img-pre.ivsky.com/img/tupian/pre/202102/21/oumei_meinv-004.jpg", "https://img-pre.ivsky.com/img/tupian/pre/202102/21/oumei_meinv-012.jpg", "https://img-pre.ivsky.com/img/tupian/pre/202102/21/oumei_meinv-013.jpg", "https://img-pre.ivsky.com/img/tupian/pre/202102/21/oumei_meinv-016.jpg", "https://img-pre.ivsky.com/img/tupian/pre/202102/21/oumei_meinv-010.jpg"]

def save_img(url): res = requests.get(url) with open(F'{SAVE_DIR}{time.time()}.jpg', 'wb+') as f: f.write(res.content)

if __name__ == '__main__': start_time = time.perf_counter() # 下载开始时间 with ThreadPoolExecutor(MAX_WORKERS) as executor: tasks = [executor.submit(save_img, url) for url in urls] # 去除下部分代码,时间基本与 map 一致。 for future in as_completed(tasks): print(future.result())
print("下载 6 张图片消耗时间为:", time.perf_counter() - start_time) # 下载 6 张图片消耗时间为: 0.840261401
复制代码


wait 方法wait 方法可以让主线程阻塞,直到满足设定的要求,该要求为 return_when 参数,其值有 ALL_COMPLETEDFIRST_COMPLETEDFIRST_EXCEPTION


import time
import requestsfrom concurrent.futures import ThreadPoolExecutor, as_completed, wait, ALL_COMPLETED, FIRST_COMPLETED
MAX_WORKERS = 20 # 最大线程数SAVE_DIR = './928/' # 文件保存路径urls = [ "https://img-pre.ivsky.com/img/tupian/pre/202102/21/oumei_meinv.jpg", "https://img-pre.ivsky.com/img/tupian/pre/202102/21/oumei_meinv-004.jpg", "https://img-pre.ivsky.com/img/tupian/pre/202102/21/oumei_meinv-012.jpg", "https://img-pre.ivsky.com/img/tupian/pre/202102/21/oumei_meinv-013.jpg", "https://img-pre.ivsky.com/img/tupian/pre/202102/21/oumei_meinv-016.jpg", "https://img-pre.ivsky.com/img/tupian/pre/202102/21/oumei_meinv-010.jpg"]

def save_img(url): res = requests.get(url) with open(F'{SAVE_DIR}{time.time()}.jpg', 'wb+') as f: f.write(res.content)

if __name__ == '__main__': start_time = time.perf_counter() # 下载开始时间 with ThreadPoolExecutor(MAX_WORKERS) as executor: tasks = [executor.submit(save_img, url) for url in urls] wait(tasks, return_when=ALL_COMPLETED) print("程序运行完毕")
print("下载 6 张图片消耗时间为:", time.perf_counter() - start_time) # 下载 6 张图片消耗时间为: 0.48876672
复制代码


最后一句:ProcessPoolExecutor 的用法与 ThreadPoolExecutor 的用法基本一致,所以可以互通。


发布于: 刚刚阅读数: 2
用户头像

爬虫 100 例作者,蓝桥签约作者,博客专家 2021.02.06 加入

6 年产品经理+教学经验,3 年互联网项目管理经验; 互联网资深爱好者; 沉迷各种技术无法自拔,导致年龄被困在 25 岁; CSDN 爬虫 100 例作者。 个人公众号“梦想橡皮擦”。

评论

发布
暂无评论
这篇博客和你唠唠 python 并发,滚雪球学python第四季,第16篇