并发和并行抓取数据总结

ゞ浴缸里的玫瑰 2021-09-13 23:50 330阅读 0赞

首先介绍并发、并行、异步和同步的概念，然后介绍 Python 中实现这些技术的主要方式，包括多线程、多线程、协程 gevent 、asyncio、及 futures 等实现方式。

# 并发与并行 #

**并发**是指在一个时间段内发生若干时间的情况。  
**并行**是指同一时刻发生若干事件的情况。

**下面用单核和多核 CPU 的工作方式来说明这两个概念。**

**并发：**在单核 CPU 的情况下，多任务操作系统的各任务是以并发的方式运行的，因为只有一个 CPU，所以各任务会以分时的方式在一段时间内分别占用 CPU 依次执行，如果在自己分得的时间段（又称时间片）内没有运行完成，则需等待下一次得到 CPU 使用权时会继续执行，直至完成。在这种方式下，各任务的时间片很小，切换速度很快，所以给我们的感觉是多个任务在同时进行，这就是**并发**。

**并行：**在多核 CPU 的情况下，因为有两个或两个以上可以同时工作的内核，所以就有可能各核上运行的任务能够同时进行，这就是**并行**。

# 同步与异步 #

提到同步和异步的概念，一般会涉及多个任务或事件的参与。可以在并发或并行的背景下理解同步和异步。

**同步**指的是并发或并行发生的各任务之间不是孤立地独自运行的，一个任务的进行可能需要在获得另一个任务给出的结果之后，或者只有一个任务完成或给出一个结果之后，另一个任务在获得这个结果之后才能继续运行。总之，各任务的运行会彼此相互制约，合拍前进，节奏和步调要协调好，否则就会出现错误。

**异步**指的是并发或并行发生的各任务之间彼此是独立运行的，不受各自的影响。这是与同步的主要区别。

**适用于同步的情况：**  
当需要多个任务互相配合在并发或并行环境中合作完成时，就需要以同步的方式运行。

**适用于异步的情况：**  
当需要将一个大的任务分解为若干个小的子任务时，各子任务可以分别独自完成，彼此之间不必互相协作，这时就可以考虑使它们异步地并行或异步地并发执行。这在数据抓取时是一种常见的模式，将若干要抓取的链接分成几组，然后对每组分别使用子任务进行抓取，待各子任务抓取结束后，可以将结果进行汇总，完成任务。

下面介绍各种 Python 支持的并发或并行方式时，使用的抓取的实例就是一个异步任务。

--------------------

# 多线程抓取 #

多线程是以并发的方式执行的，由于 Python 语言自身的设计限制，因此即使是多核的机器，Python 的多线程程序也只能运行在一个单核上以并发的方式运行。代码示例如下：

import threading
    from bs4 import BeautifuSoup
    import requests
    
    def get_urls_in_pages(from_page_num,to_page_num):
        url = r'http://college.koolearn.com/kaoyan/s/fs-0-0-0-0-0-0/?p='
        urls = []
        for i in range(from_page_num,to_page_num+1):
            urls.append(url + str(i))
        #下面的操作，用requets模块、BeautifulSoup模块来抓取数据
    def main_test():
        page_ranges_lst = [
            (1,10),
            (11,20),
            (21,30),
            (31,40),
        ]
    
        th_lst = []
        for page_ranges in page_ranges_lst:
            th = threading.Thread(target=get_urls_in_pages,args=(page_ranges[0],page_ranges[1]))
            th_lst.append(th) #将生成的县城都放入 th_list 列表中，此时 th 只是线程初始化后的实例，还未开始执行。
    
        for th in th_lst:
            th.start()  #该语句这些线程在系统中接受系统的调度，异步地并发地执行了
        for th in th_lst:
            th.join()   #等待各线程执行完毕后，再退出外层函数 main_test()
    if __name__ == '__main__':
        main_test()

--------------------

# 多进程抓取 #

多进程方式依赖于所在机器的处理器个数。在多核机器上进行多进程编程时，各核上运行的进程之间是并行执行的。可以利用进程池，使每一个内核上运行一个进程，当池中的进程数量大于内核总数时，待运行的进程会等待，直至其他进程运行完毕让出内核。，当系统内只有一个单核 CPU 时，多进程并行不会发生，此时各进程会依次占用 CPU 运行至完成。

下面的代码给出在 8 核处理器上运行多进程的程序

**获得 CPU 可用的核数，命令如下：**

from multiprocessing import cpu_count
    print(cpu_count())

**多进程抓取代码如下：**

import multiprocessing
    from bs4 import BeautifuSoup
    import requests
    
    def get_urls_in_pages(from_page_num,to_page_num):
        url = r'http://college.koolearn.com/kaoyan/s/fs-0-0-0-0-0-0/?p='
        urls = []
        for i in range(from_page_num,to_page_num+1):
            urls.append(url + str(i))
        #下面的操作，用requets模块、BeautifulSoup模块来抓取数据
    def main_test():
        page_ranges_lst = [
            (1,10),
            (11,20),
            (21,30),
            (31,40),
        ]
    
        pool = multiprocessing.Pool(process=4)  # 4核CPU,该语句的作用可同时并行 4 个进程的进程池
        for page_ranges in page_ranges_lst:
            #apply_async 方法可以使进入进程池的进程以异步的方式并行运行
            pool.apply_async(get_urls_in_pages,(page_ranges[0],page_ranges[1]))
            pool.close()
            pool.join()
    if __name__ == '__main__':
        mian_test()

--------------------

# 利用协程抓取 #

一般在 Python 讨论协程时，都会与生成器（generator）联系在一起。生成器是一个函数，主要特点是生成器在返回值时，不是使用 return，而是使用 yield 关键字。在定义函数时，如果函数体中包含 yield 关键字，则该函数就被认为是一个生成器。

## 利用 gevent 库抓取 ##

使用 gevent 异步库可以方便地实现基于协程的并发设计，可以利用 gevent 的 “monkey patch” 能力，将标准的 I/O 函数转为异步执行的函数，而代码基本不用变化。

在 gevent 中使用 greenlet 对象实现并发，greeenlet 就是协程，可以认为是一种轻量线程。

**gevent 库的安装：**

sudo pip3 install gevent

**利用 gevent 库抓取的代码示例如下：**

from multiprocessing import cpu_count
    print(cpu_count())

**多进程抓取代码如下：**

import gevent
    from gevent import monkey
    monkey.patch_all()
    from bs4 import BeautifuSoup
    import requests
    
    def get_urls_in_pages(from_page_num,to_page_num):
        url = r'http://college.koolearn.com/kaoyan/s/fs-0-0-0-0-0-0/?p='
        urls = []
        for i in range(from_page_num,to_page_num+1):
            urls.append(url + str(i))
        #下面的操作，用requets模块、BeautifulSoup模块来抓取数据
    def main_test():
        page_ranges_lst = [
            (1,10),
            (11,20),
            (21,30),
            (31,40),
        ]
    
        jobs = []
        for page_range in page_ranges_lst:
            jobs.append(gevent.spawn(get_urls_in_pages,page_range[0],page_range[1]))
        gevent.joinall(jobs)
    
    if __name__ == '__main__':
        mian_test()

**注意：**

若没有如下两句，则会变为依次顺序抓取，失去了并发的能力。

from gevent import monkey
    monkey.patch_all()

## 利用 asyncio 库抓取 ##

从 Python3.4 开始，Python 自身提供了 asyncio 库用于实现异步 I/O 程序设计，利用 asyncio 可以使用协程进行并发编程。asyncio 提供了对套接字 API 的全面支持，但在使用时涉及的细节问题比较多，可以引入另一个 aiohttp 库。

asyncio 库主要提供了异步并发编程的框架  
aiohttp 库用于在抓取时发出 http 请求

**asyncio 的安装**

pip3 install asynci

**aiohttp 的安装：**

sudo pip3 install aiohttp

**使用 asyncio 和 aiohttp 实现异步并发抓取的代码如下：**

import asyncio
    import aiohttp
    import requests
    from bs4 import BeautifulSoup
    
    sem = asyncio.Semaphore(4)   #设置Semaphore为4，说明在抓取时最多并发发出4个请求 
    @asyncio.coroutine
    def get_urls_in_pages(from_page_num,to_page_num):
        url = r'http://college.koolearn.com/kaoyan/s/fs-0-0-0-0-0-0/?p='
        urls = []
        for i in range(from_page_num,to_page_num+1):
            urls.append(url + str(i))
        for url in urls:
            with(yield from sem):
                response = yield from aiohttp.request('GET',url) #该语句功能是发出http请求，并在此处等待请求的结果
            html = yield from response.read_and_close()
            bs = BeautifulSoup(html.decode('utf-8'))
        #以下的操作是抓取数据
    def main_test():
        page_ranges_lst = [
            (1,10),
            (11,20),
            (21,30),
            (31,40),
        ]
    
        #下面三行语句代码完成的功能就是启动事件循环机制，生成待运行的诸协程，然后调度运行
        loop = asyncio.get_event_loop()
        f = asyncio.wait([get_urls_in_pages(page_range[0],page_range[1]) for page_range in page_ranges_lst])
        loop.run_until_complete(f)
    
    if __name__ == '__main__':
        mian_test()