并发体验：Python抓图的8种方式

Posted 2020-11-18 pythonedu

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了并发体验：Python抓图的8种方式相关的知识，希望对你有一定的参考价值。

本文是通过用爬虫示例来说明并发相关的多线程、多进程、协程之间的执行效率对比。

技术分享图片

假设我们现在要在网上下载图片，一个简单的方法是用 requests+BeautifulSoup。注：本文所有例子都使用python3.5）

单线程

示例 1：get_photos.py

import os
import time
import uuid

import requests
from bs4 import BeautifulSoup

def out_wrapper(func):  # 记录程序执行时间的简单装饰器
    def inner_wrapper():
        start_time = time.time()
        func()
        stop_time = time.time()
        print(‘Used time {}‘.format(stop_time-start_time))
    return inner_wrapper

def save_flag(img, filename):  # 保存图片
    path = os.path.join(‘down_photos‘, filename)
    with open(path, ‘wb‘) as fp:
        fp.write(img)

def download_one(url):  # 下载一个图片
    image = requests.get(url)
    save_flag(image.content, str(uuid.uuid4()))

def user_conf():  # 返回30个图片的url
    url = ‘https://unsplash.com/‘
    ret = requests.get(url)
    soup = BeautifulSoup(ret.text, "lxml")
    zzr = soup.find_all(‘img‘)
    ret = []
    num = 0
    for item in zzr:
        if item.get("src").endswith(‘80‘) and num < 30:
            num += 1
            ret.append(item.get("src"))
    return ret

@out_wrapper
def download_many():
    zzr = user_conf()
    for item in zzr:
        download_one(item)

if __name__ == ‘__main__‘:
    download_many()

示例1进行的是顺序下载，下载30张图片的平均时间在60s左右（结果因实验环境不同而不同）。

这个代码能用但并不高效，怎么才能提高效率呢？

参考开篇的示意图，有三种方式：多进程、多线程和协程。下面我们一一说明：

我们都知道 Python 中存在 GIL（主要是Cpython），但 GIL 并不影响 IO 密集型任务，因此对于 IO 密集型任务而言，多线程更加适合（线程可以开100个，1000个而进程同时运行的数量受 CPU 核数的限制，开多了也没用）

不过，这并不妨碍我们通过实验来了解多进程。

多进程

示例2

from multiprocessing import Process
from get_photos import out_wrapper, download_one, user_conf

@out_wrapper
def download_many():
    zzr = user_conf()
    task_list = []
    for item in zzr:
        t = Process(target=download_one, args=(item,))
        t.start()
        task_list.append(t)
    [t.join() for t in task_list]  # 等待进程全部执行完毕（为了记录时间）

if __name__ == ‘__main__‘:
    download_many()

本示例重用了示例1的部分代码，我们只需关注使用多进程的这部分。

笔者测试了3次（使用的机器是双核超线程，即同时只能有4个下载任务在进行），输出分别是：19.5s、17.4s和18.6s。速度提升并不是很多，也证明了多进程不适合io密集型任务。

还有一种使用多进程的方法，那就是内置模块futures中的ProcessPoolExecutor。

示例3

from concurrent import futures
from get_photos import out_wrapper, download_one, user_conf

@out_wrapper
def download_many():
    zzr = user_conf()
    with futures.ProcessPoolExecutor(len(zzr)) as executor:
        res = executor.map(download_one, zzr)
    return len(list(res))

if __name__ == ‘__main__‘:
    download_many()

使用 ProcessPoolExecutor 代码简洁了不少，executor.map 和标准库中的 map用法类似。耗时和示例2相差无几。多进程就到这里，下面来体验一下多线程。

多线程

示例4

import threading
from get_photos import out_wrapper, download_one, user_conf

@out_wrapper
def download_many():
    zzr = user_conf()
    task_list = []
    for item in zzr:
        t = threading.Thread(target=download_one, args=(item,))
        t.start()
        task_list.append(t)
    [t.join() for t in task_list]

if __name__ == ‘__main__‘:
    download_many()

threading 和 multiprocessing 的语法基本一样，但是速度在9s左右，相较多进程提升了1倍。

下面的示例5和示例6中分别使用内置模块 futures.ThreadPoolExecutor 中的 map 和submit、as_completed

示例5

from concurrent import futures
from get_photos import out_wrapper, download_one, user_conf

@out_wrapper
def download_many():
    zzr = user_conf()
    with futures.ThreadPoolExecutor(len(zzr)) as executor:
        res = executor.map(download_one, zzr)
    return len(list(res))

if __name__ == ‘__main__‘:
    download_many()

示例6：

from concurrent import futures
from get_photos import out_wrapper, download_one, user_conf

@out_wrapper
def download_many():
    zzr = user_conf()
    with futures.ThreadPoolExecutor(len(zzr)) as executor:
        to_do = [executor.submit(download_one, item) for item in zzr]
        ret = [future.result() for future in futures.as_completed(to_do)]
    return ret

if __name__ == ‘__main__‘:
    download_many()

Executor.map 由于和内置的map用法相似所以更易于使用，它有个特性：返回结果的顺序与调用开始的顺序一致。不过，通常更可取的方式是，不管提交的顺序，只要有结果就获取。

为此，要把 Executor.submit 和 futures.as_completed结合起来使用。

最后到了协程，这里分别介绍 gevent 和 asyncio。

gevent

示例7

from gevent import monkey
monkey.patch_all()

import gevent
from get_photos import out_wrapper, download_one, user_conf

@out_wrapper
def download_many():
    zzr = user_conf()
    jobs = [gevent.spawn(download_one, item) for item in zzr]
    gevent.joinall(jobs)

if __name__ == ‘__main__‘:
    download_many()

asyncio

示例8

import uuid
import asyncio

import aiohttp
from get_photos import out_wrapper, user_conf, save_flag

async def download_one(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            save_flag(await resp.read(), str(uuid.uuid4()))

@out_wrapper
def download_many():
    urls = user_conf()
    loop = asyncio.get_event_loop()
    to_do = [download_one(url) for url in urls]
    wait_coro = asyncio.wait(to_do)
    res, _ = loop.run_until_complete(wait_coro)
    loop.close()
    return len(res)

if __name__ == ‘__main__‘:
    download_many()