在 python 编码过程中,有时存在这样的一个需求,同时下载 N 张图片,并且要快。
一般这样的需求,只需要编写一个 for 循环即可实现,但是加上 快 这个要求,就不好实现了。
图片下载属于 I/O 操作,比较耗时,基于此,可以利用 python 中的多线程将其实现。
为了不让大家学的太困倦,特意找来 6 张美丽的图片,本次学习将围绕这几张图片进行。
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
复制代码
单线程下载 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
参数,代表创建的线程数或者进程数。
这两个类可以在不同的线程或进程中执行 可调用对象,ProcessPoolExecutor
的 max_workers
参数可以为空,程序会自动创建与电脑 CPU
数目相同的进程数。
使用 ThreadPoolExecutor
实现多线程下载
import time
import requests
from 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)
,安排可调用对象 fn
以 fn(*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 requests
from 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_COMPLETED
,FIRST_COMPLETED
,FIRST_EXCEPTION
。
import time
import requests
from 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
的用法基本一致,所以可以互通。
评论