在 python 編碼過程中，有時存在這樣的一個需求，同時下載 N 張圖片，并且要快。

一般這樣的需求，只需要編寫一個 for 循環(huán)即可實現(xiàn)，但是加上 快 這個要求，就不好實現(xiàn)了。

圖片下載屬于 I/O 操作，比較耗時，基于此，可以利用 python 中的多線程將其實現(xiàn)。

為了不讓大家學(xué)的太困倦，特意找來 6 張美麗的圖片，本次學(xué)習(xí)將圍繞這幾張圖片進(jìn)行。

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 循環(huán)，同步代碼如下所示：

import timeimport requestsurls = [    "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 模塊 實現(xiàn)對 6 張圖片的下載，這個模塊實現(xiàn)了 ThreadPoolExecutor 類和 ProcessPoolExecutor 類，都繼承自Executor，分別被用來創(chuàng)建 線程池 和 進(jìn)程池，接受 max_workers 參數(shù)，代表創(chuàng)建的線程數(shù)或者進(jìn)程數(shù)。

這兩個類可以在不同的線程或進(jìn)程中執(zhí)行 可調(diào)用對象，ProcessPoolExecutor 的 max_workers 參數(shù)可以為空，程序會自動創(chuàng)建與電腦 CPU數(shù)目相同的進(jìn)程數(shù)。

使用 ThreadPoolExecutor 實現(xiàn)多線程下載

import timeimport requestsfrom concurrent import futuresMAX_WORKERS = 20  # 最大線程數(shù)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() 方法會返回一個生成器，后續(xù)代碼可以迭代獲取每個線程的執(zhí)行結(jié)果    print("下載 6 張圖片消耗時間為：", time.perf_counter() - start_time)    # 下載 6 張圖片消耗時間為： 0.415939759

當(dāng)使用多線程代碼之后，時間從單線程的 1.9s 變?yōu)榱硕嗑€程的 0.4s，能看到效率的提升。

Future 類

在上述多線程代碼中，使用了 concurrent 庫中的 future 對象，該對象是 Future 類的對象，它的實力表示可能已經(jīng)完成或尚未完成的 延遲計算，該類具備 done() 方法，返回調(diào)用對象是否已經(jīng)執(zhí)行，有該方法的同時還具備一個 add_done_callback() 方法，表示調(diào)用對象執(zhí)行完畢的回調(diào)函數(shù)。

from concurrent.futures import ThreadPoolExecutordef print_name():    return "橡皮擦"def say_hello(obj):    """可調(diào)用對象執(zhí)行完畢，綁定的回調(diào)函數(shù)"""    w_name = obj.result()    s = w_name + "你好"    print(s)    return swith ThreadPoolExecutor(1) as executor:    executor.submit(print_name).add_done_callback(say_hello)

在上述代碼中用到了如下知識點：

• executor.map()：該方法類似 map 函數(shù)，原型為 map(func, *iterables, timeout=None, chunksize=1)，異步執(zhí)行 func，并支持多次并發(fā)調(diào)用；
• executor.submit()：方法原型為 submit(fn, *args, **kwargs)，安排可調(diào)用對象 fn 以 fn(*args, **kwargs) 的形式執(zhí)行，并返回 Future 對象來表示它的執(zhí)行結(jié)果，該方法只能進(jìn)行單個任務(wù)，如果需要并發(fā)多個任務(wù)，需要使用 map 或者 as_completed；
• future對象.result()：返回調(diào)用返回的值，有個等待時間參數(shù) timeout 可以設(shè)置；
• future對象add_done_callback()：該方法中綁定的回調(diào)函數(shù)在 future 取消或者完成后運行，表示 future 本身

補充說明

as_completed() 方法
該方法參數(shù)是一個 Future 列表，返回值是一個 Future 組成的生成器，在調(diào)用 as_completed() 方法時不會阻塞，只有當(dāng)對迭代器進(jìn)行循環(huán)時，每調(diào)用一次 next() 方法，如果當(dāng)前 Future 對象還未完成，則會阻塞。

修改下載 6 張圖片的代碼，使用 as_completed() 進(jìn)行實現(xiàn)，最后得到的時間優(yōu)于單線程，但如果對結(jié)果進(jìn)行迭代，調(diào)用 result() 方法，則時間會加長。

import timeimport requestsfrom concurrent.futures import ThreadPoolExecutor, as_completedMAX_WORKERS = 20  # 最大線程數(shù)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 方法可以讓主線程阻塞，直到滿足設(shè)定的要求，該要求為 return_when 參數(shù)，其值有 ALL_COMPLETED，FIRST_COMPLETED，FIRST_EXCEPTION。

import timeimport requestsfrom concurrent.futures import ThreadPoolExecutor, as_completed, wait, ALL_COMPLETED, FIRST_COMPLETEDMAX_WORKERS = 20  # 最大線程數(shù)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 的用法基本一致，所以可以互通。

寫在后面

以上內(nèi)容就是本文的全部內(nèi)容，希望對學(xué)習(xí)路上的你有所幫助~

今天是持續(xù)寫作的第 235 / 365 天。
期待 關(guān)注，點贊、評論、收藏。

更多精彩

• 滾雪球?qū)W Python（完結(jié)）
• 滾雪球?qū)W Python 第二輪（完結(jié)）
• 滾雪球?qū)W Python 第三輪
• 滾雪球?qū)W Python 番外篇（完結(jié)）