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

在 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 参数,代表创建的线程数或者进程数。

这两个类可以在不同的线程或进程中执行 可调用对象ProcessPoolExecutormax_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),安排可调用对象 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 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_COMPLETEDFIRST_COMPLETEDFIRST_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 的用法基本一致,所以可以互通。

写在后面

以上内容就是本文的全部内容,希望对学习路上的你有所帮助~

今天是持续写作的第 235 / 365 天。
期待 关注点赞评论收藏

更多精彩

  • 滚雪球学 Python(完结)
  • 滚雪球学 Python 第二轮(完结)
  • 滚雪球学 Python 第三轮
  • 滚雪球学 Python 番外篇(完结)

你可能感兴趣的