這篇博客和你嘮嘮 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 引數，代表創建的執行緒數或者行程數，

這兩個類可以在不同的執行緒或行程中執行 可呼叫物件，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 的用法基本一致，所以可以互通，

寫在后面

以上內容就是本文的全部內容，希望對學習路上的你有所幫助~

今天是持續寫作的第 235 / 365 天，
期待關注，點贊、評論、收藏，

更多精彩

滾雪球學 Python（完結）
滾雪球學 Python 第二輪（完結）
滾雪球學 Python 第三輪
滾雪球學 Python 番外篇（完結）

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/307241.html

標籤：python

上一篇：華為二面！！！面試官直接問我Java中到底什么是NIO？這不是直接送分題？？？

下一篇：100 個基本 Python 面試問題第二部分(21-40)