需求：抓取网站https://www.edge.org/library所有的书名、作者以及作者的维基百科简介

工具：PyCharm
第三方库：requests、BeautifulSoup

import requests
from bs4 import BeautifulSoup

爬虫有两种方式，一种是模拟请求获取返回数据即可，另一种需要解析HTML。本次爬虫方式就是后者。
BeautifulSoup是解析HTML的库。

初次写爬虫，一开始傻不拉几的用同步方式，依次获取所有的页面的数据，总共耗时2个多小时，令人发指。后来自己摸索，分别采用多进程/多线程/多协程的方式，效率惊人。

说到进程池，Python提供了两种创建方式，直接上代码：

import re
import time
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
from multiprocessing.pool import Pool

import requests
import sys
from bs4 import BeautifulSoup

from book.ExcelUtils import ExcelUtils

headers = {
    'user-agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/67.0.3396.99 Safari/537.36'}


# 获取书名、作者、作者wiki简介
def get_book_info(url):
    book_dict = {'name': [], 'author': [], 'info': []}
    res = requests.get(url, headers=headers)
    if res.status_code == 200:
        find_book_name_list = re.findall('<div class="views-field views-field-title">.*?<a href=.*?>(.*?)</a>',
                                         res.text)
        book_dict['name'] = find_book_name_list

        find_author_list = re.findall(
            '<span class="views-field views-field-field-edge-author">.*?<a href=".*?">(.*?)</a>', res.text)
        book_dict['author'] = find_author_list

        # 获取作者wiki简介
        get_wiki_author_info(find_author_list, book_dict)

    else:
        print('get_book_info is failed, url:[%s]' % url)

    return book_dict


# 根据作者名字获取作者wiki简介
def get_wiki_author_info(find_author_list, book_dict):
    for name in find_author_list:
        url = 'https://en.wikipedia.org/wiki/%s' % name.replace(' ', '_')
        try:
            res = requests.get(url, headers=headers)
            if res.status_code == 200:
                soup = BeautifulSoup(res.text, "html.parser")
                soup_select = soup.select('#mw-content-text p')
                if str.strip(soup_select[0].get_text()) != '':
                    if soup_select[0].get_text().find('may refer to') != -1:
                        # 存在多种选项
                        name_list = soup.select('#mw-content-text li a')
                        for index, item in enumerate(name_list):
                            if item.get_text().find(name) != -1:
                                # 递归，默认只取第一个匹配的
                                get_wiki_author_info([item.get_text()], book_dict)
                                break
                    else:
                        book_dict['info'].append(soup_select[0].get_text())
                else:
                    book_dict['info'].append(soup_select[1].get_text())
            else:
                book_dict['info'].append('get failed, url:[%s]' % url)
                print('get_wiki_author_info is failed, url:[%s]' % url)

        except:
            book_dict['info'].append('get exception, url:[%s]' % url)

# 进程池的一种创建方式
def pool_test(url_list):
    book_list = []
    # 创建进程池
    pool = Pool(20)
    start = time.time()
    for url in url_list:
        time.sleep(0.5)
        result = pool.apply_async(get_book_info, args=(url,))
        book_list.append(result)

    # 关闭进程池，不再接受新的进程，依旧处理未处理完的任务
    pool.close()
    # 主进程等待所有子进程执行完毕，必须在close或terminate之后
    pool.join()
    print('time: ', time.time() - start)

    book_name_list = []
    author_list = []
    author_info_list = []
    print('book_list: ', len(book_list))
    for v in book_list:
        book_name_list.extend(v.get()['name'])
        author_list.extend(v.get()['author'])
        author_info_list.extend(v.get()['info'])

    ExcelUtils.write_data_to_excel('bookInfo', book_name_list, author_list, author_info_list)

# 线程池的创建方式
def thread_pool_test(url_list):
    book_list = []
    # 创建线程池
    pool = ThreadPoolExecutor(max_workers=20)
    start = time.time()
    for url in url_list:
        time.sleep(0.5)
        result = pool.submit(get_book_info, url)
        book_list.append(result)

    pool.shutdown()
    print('time: ', time.time() - start)

    book_name_list = []
    author_list = []
    author_info_list = []
    print('book_list: ', len(book_list))
    for future in book_list:
        book_name_list.extend(future.result()['name'])
        author_list.extend(future.result()['author'])
        author_info_list.extend(future.result()['info'])

    ExcelUtils.write_data_to_excel('bookInfo', book_name_list, author_list, author_info_list)

# 进程池的另外一种创建方式，跟线程池的创建方式一样。其方法等也相同。
def process_pool_test(url_list):
    book_list = []
    # 创建进程池
    pool = ProcessPoolExecutor(max_workers=20)
    start = time.time()
    for url in url_list:
        time.sleep(0.5)
        result = pool.submit(get_book_info, url)
        book_list.append(result)

    pool.shutdown()
    print('time: ', time.time() - start)

    book_name_list = []
    author_list = []
    author_info_list = []
    print('book_list: ', len(book_list))
    for future in book_list:
        book_name_list.extend(future.result()['name'])
        author_list.extend(future.result()['author'])
        author_info_list.extend(future.result()['info'])

    ExcelUtils.write_data_to_excel('bookInfo', book_name_list, author_list, author_info_list)


if __name__ == '__main__':
    sys.setrecursionlimit(10000)
    url_list = ['https://www.edge.org/library']
    for i in range(1, 52):
        url_list.append('https://www.edge.org/library?page=%s' % i)

    thread_pool_test(url_list)

Python协程爬虫，代码相对于进程线程，稍微麻烦了一点，如下：

import asyncio
import aiohttp
import re
import time

import sys
from bs4 import BeautifulSoup

from spider.ExcelUtils import ExcelUtils

headers = {
    'user-agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/67.0.3396.99 Safari/537.36'}


# 获取书名、作者
async def get_book_info(url, semaphore):
    book_dict = {'name': [], 'author': []}
    async with semaphore:
        async with aiohttp.ClientSession() as session:
            async with session.get(url, headers=headers) as res:
                if res.status == 200:
                    text = await res.text()
                    find_book_name_list = re.findall(
                        '<div class="views-field views-field-title">.*?<a href=.*?>(.*?)</a>', text)
                    book_dict['name'] = find_book_name_list

                    find_author_list = re.findall(
                        '<span class="views-field views-field-field-edge-author">.*?<a href=".*?">(.*?)</a>', text)
                    book_dict['author'] = find_author_list

                else:
                    print('get_book_info is failed, url:[%s]' % url)

    return book_dict


# 根据作者名字获取作者wiki简介
async def get_wiki_author_info(name, semaphore):
    url = 'https://en.wikipedia.org/wiki/%s' % name.replace(' ', '_')
    try:
        async with semaphore:
            async with aiohttp.ClientSession() as session:
                async with session.get(url, headers=headers) as res:
                    if res.status == 200:
                        text = await res.text()
                        soup = BeautifulSoup(text, "html.parser")
                        soup_select = soup.select('#mw-content-text p')
                        if str.strip(soup_select[0].get_text()) != '':
                            if soup_select[0].get_text().find('may refer to') != -1:
                                # 存在多种选项
                                name_list = soup.select('#mw-content-text li a')
                                for index, item in enumerate(name_list):
                                    if item.get_text().find(name) != -1:
                                        # 递归，默认只取第一个匹配的
                                        await get_wiki_author_info([item.get_text(), semaphore])
                                        break
                            else:
                                return soup_select[0].get_text()
                        else:
                            return soup_select[1].get_text()
                    else:
                        print('get_wiki_author_info is failed, url:[%s]' % url)
                        return 'get failed, url:[%s]' % url

    except:
        return 'get exception, url:[%s]' % url


if __name__ == '__main__':
    sys.setrecursionlimit(10000)
    url_list = ['https://www.edge.org/library']
    for i in range(1, 52):
        url_list.append('https://www.edge.org/library?page=%s' % i)

    # 信号量，限制并发数，类比20个线程
    semaphore = asyncio.Semaphore(20)
    loop = asyncio.get_event_loop()
    tasks = [asyncio.ensure_future(get_book_info(url, semaphore)) for url in url_list]

    start = time.time()
    loop.run_until_complete(asyncio.wait(tasks))

    book_name_list = []
    author_list = []
    print('book_list: ', len(tasks))
    for future in tasks:
        book_name_list.extend(future.result()['name'])
        author_list.extend(future.result()['author'])
    # 获取作者wiki简介
    wiki_tasks = [asyncio.ensure_future(get_wiki_author_info(name, semaphore)) for name in author_list]
    loop.run_until_complete(asyncio.wait(wiki_tasks))
    author_info_list = [future.result() for future in wiki_tasks]
    print('time: ', time.time() - start)

    ExcelUtils.write_data_to_excel('bookInfo', book_name_list, author_list, author_info_list)

ExcelUtils工具类的代码如下：

import xlwt


class ExcelUtils:

    @staticmethod
    def def_style():
        style = xlwt.XFStyle()
        alignment = xlwt.Alignment()
        alignment.horz = xlwt.Alignment.HORZ_CENTER  # 水平居中
        alignment.vert = xlwt.Alignment.VERT_CENTER  # 垂直居中
        style.alignment = alignment

        return style

    @staticmethod
    def write_data_to_excel(excel_name, book_name_list, author_list, author_info_list):
        # 实例化一个Workbook()对象(即excel文件)
        wbk = xlwt.Workbook()
        # 新建一个名为Sheet1的excel sheet。此处的cell_overwrite_ok =True是为了能对同一个单元格重复操作。
        sheet = wbk.add_sheet('Sheet1', cell_overwrite_ok=True)
        col_1 = sheet.col(0)
        col_2 = sheet.col(1)
        col_3 = sheet.col(2)
        col_1.width = 256 * 45
        col_2.width = 256 * 30
        col_3.width = 256 * 200

        sheet.write(0, 0, '书名', ExcelUtils.def_style())
        sheet.write(0, 1, '作者', ExcelUtils.def_style())
        sheet.write(0, 2, '作者简介', ExcelUtils.def_style())

        for i in range(len(author_info_list)):
            sheet.write(i + 1, 0, book_name_list[i])
            sheet.write(i + 1, 1, author_list[i], ExcelUtils.def_style())
            sheet.write(i + 1, 2, author_info_list[i])

        wbk.save(excel_name + '.xls')

四种方式每次的实验时间都不同，这里只取其中一次的执行时间，如下：

多进程 Pool(20)
time: 215.41232109069824秒
数据总数：1421

多进程 ProcessPoolExecutor(max_workers=20)
time:  202.88900017738342秒
数据总数：1481


多线程：ThreadPoolExecutor(max_workers=20)
time: 198.67899990081787秒
数据总数：1481

协程： async
time:  166.4319999217987秒
数据总数：1481

可以看出这四者的效率差别并不大，由于每次时间都不同，也不能确切的说哪种方式更优。

不过根据前面讲的进程、线程、协程，一般情况下。协程执行效率是高于线程的，Pool的效率高于ProcessPoolExecutor。但是协程是单线程的，无法利用多核CPU，组合应用的时候，推荐Pool + 协程。