Python3网络爬虫开发实战（6）异步爬虫

爬虫是 IO 密集型任务，使用 requests 库来爬取某个站点，当发出一个请求后，程序必须等待网站返回响应才能进行下一步操作，我们可以使用异步爬虫的方式来优化这一步骤

一、基础知识

使用协程实现加速，这种方法对 IO 密集型任务非常有效，应用在爬虫中，爬取效率可以得到大大提升；

• 阻塞：阻塞状态指程序未得到所需计算资源时被挂起的状态，程序在等待某个操作完成期间，自身无法继续干别的事情，则称该程序在该操作上是阻塞的；
• 同步：不同程序单元为了共同完成某个任务，在执行过程中需要靠某种通信方式保持协调一致，此时这些程序单元是同步执行的；同步意味着有序；
• 异步：为了完成某个任务，不同程序单元之间无需通信协调也能完成任务，此时不相关的程序单元之间是异步的；异步意味着无序；
• 多进程：多进程利用 CPU 的多核优势，在同一时间并行执行多个任务，可以大大提高执行效率；
• 协程：协程不是进程或线程，其执行过程类似于自主控制的，可以暂停执行的，不带返回值的函数调用，协程拥有自己的寄存器上下文和栈。协程调度切换时，将寄存器上下文和栈保存到其他地方，在切回来的时候，恢复先前保存的寄存器

协程中的四个概念：

• event_loop：事件循环，相当于一个无限循环，我们可以把一些函数注册到这个事件循环上，当满足发生条件的时候，就调用对应的处理方法；
• coroutine：在 Python 中常指代协程对象类型，我们可以将协程对象注册到时间循环中，它会被事件循环调用，我们可以使用 async 关键字来定义一个方法，这个方法在调用时不会立即被执行，而是会返回一个协程对象；
• task：任务，这是对协程对象的进一步封装，包含协程对象的各个状态；
• future：代表将来执行或者没有执行的任务的结果，实际上和 task 没有本质区别；

二、定义协程

协程在处理等待操作时，具有很大的优势，遇见等待操作时程序可以暂时挂起，转而执行其他操作，从而避免因一直等待一个程序而耗费过多的时间，能够充分利用资源；耗时等待操作一般都是 IO 操作，例如文件读取，网络请求等等；

要实现挂起，我们需要使用 await 关键字来进行操作，但是 await 关键字后面接收的对象必须是如下格式之一：

• 一个原生协程对象；
• 一个由 types.coroutine 修饰的生成器，这个生成器可以返回协程对象；
• 有一个包含 __await__ 方法的对象返回的一个迭代器

首先需要引入 asyncio 包，这样才可以使用 async 和 await 关键字，然后使用 async 定义了一个方法，利用该方法接收参数并不会立即执行，而是变成一个协程 coroutine 对象，之后我们使用 asyncio 的 get_event_loop 方法创建一个事件循环 loop，利用 loop 的 create_task 函数将这个 coroutine 转化为 task，然后调用 loop 的 run_until_complete 方法去执行这个 task；

Nest_asyncio的核心是它对原始的asyncio模块的增强，当导入 nest_asyncio.apply 并应用于事件循环时，它会修改asyncio的行为，允许在同一个事件循环中安全地运行不同异步库的任务；这意味着可以混合使用多个基于asyncio的库，而无需担心它们之间的冲突或者事件循环的混乱；这意味着我们可以在 Jupyter 中使用

• asyncio 中的 task 在未加入事件循环中执行时，处于 pending 状态，在执行完毕后，状态变为 finished，同时如果携程函数具有返回值，可以在 task.result() 中体现；
• asyncio 中的 coroutine 中的值的处理方法有三种，1. 定义一个全局变量在全局变量中存储；2. 在 task 或 future 还未加入事件循环时，使用 add_done_callback 函数添加回调，回调函数中的参数就是 处于 finished 的 task 本身；3. 等待事件全部执行完毕后，直接调用 task.result() 来获取结果；

1. 单任务协程

import asyncio
import nest_asyncio
nest_asyncio.apply()async def execute(x):print('numer:', x)coroutine = execute(1)# 显式
# loop = asyncio.get_event_loop()
# task = loop.create_task(coroutine)
# loop.run_until_complete(task)# 隐式
# loop = asyncio.get_event_loop()
# loop.run_until_complete(coroutine)# 提前注册 
future = asyncio.ensure_future(coroutine)
loop = asyncio.get_event_loop()
loop.run_until_complete(future)

2. 多任务协程

import asyncio
import nest_asyncio
nest_asyncio.apply()async def execute(x):await asyncio.sleep(10-x)print('numer', x)# 只能显示
loop = asyncio.get_event_loop()
tasks = [loop.create_task(execute(item)) for item in range(0, 10)]
## 这里必须要添加 asyncio.wait
loop.run_until_complete(asyncio.wait(tasks))

3. Queue

4. aiohttp 协程实现

aiohttp 是一个支持异步请求的库，它和 asyncio 配合使用，可以使我们非常方便地实现异步请求操作，aiohttp 和 request 的请求和响应类似，关键是其函数结果会有协程对象需要使用 await 关键字接受；

并发量很大的时候，目标网站很可能无法在短时间内响应，而且有瞬间将目标网站爬挂掉的危险，因此我们需要使用 semaphore = asyncio.Semaphore(CONCURRENCY) 和 async with semaphore 控制爬取的并发量；

import aiohttp
import asyncio
import nest_asyncio
nest_asyncio.apply()# 设置并发限制
CONCURRENCY = 5
semaphore = asyncio.Semaphore(CONCURRENCY)async def fetch(session, url):async with session.get(url) as response:return await response.text(), response.statusasync def main():# 设置超时timeout = aiohttp.ClientTimeout(total=1)# 使用并发async with semaphore:async with aiohttp.ClientSession(timeout=timeout) as session:for i in range(100):html, status = await fetch(session, 'http://www.baidu.com')print(f'html :{html[:100]}')print(f'status:{status}')if __name__ == '__main__':# mode: 1loop = asyncio.get_event_loop()loop.run_until_complete(main())# mode: 2 python > 3.7asyncio.run(main())

5. httpx 协程实现

import asyncio
import httpx
import nest_asyncio
nest_asyncio.apply()# 设置并发限制
CONCURRENCY = 5
semaphore = asyncio.Semaphore(CONCURRENCY)async def fetch(client, url):# 不可以使用 async with as 方法构建response = await client.get(url)return response.text, response.status_codeasync def main():# 使用并发async with semaphore:async with httpx.AsyncClient() as client:for i in range(10):html, status = await fetch(client, 'http://www.baidu.com')print(f'html :{html[:100]}')print(f'status:{status}')if __name__ == '__main__':# loop = asyncio.get_event_loop()# loop.run_until_complete(main())asyncio.run(main())

6. 其他异步环境

1. asyncio：asyncio — Asynchronous I/O — Python 3.12.4 documentation
2. Trio：Trio: a friendly Python library for async concurrency and I/O — Trio 0.26.0 documentation
3. anyio：agronholm/anyio: High level asynchronous concurrency and networking framework that works on top of either trio or asyncio (github.com)

7. 实战

大学教务处课表数据采集（以北京师范大学为例）课表采集 课表爬虫_爬虫爬课表-CSDN博客

import pandas as pd
from tqdm import tqdm
from lxml import etree
import aiohttp
import asyncio
import nest_asyncio
import json
import numpy as np
nest_asyncio.apply()df = df.reset_index(drop=True).reset_index()
lst = list(df.to_dict('index').values())
lst[:1]
class GetCurriculum:def __init__(self, lst_dic, cookies, headers, params):self.que = asyncio.Queue()[self.que.put_nowait(dic) for dic in lst_dic]self.cookies = cookiesself.headers = headersself.params = paramsself.result = {}self.eventloop()async def scrape_url(self, session, dic):xn = 2023xq = 0pycc = dic['val_1']nj = dic['val_2']yxb = dic['val_3']zydm = dic['val_4']url = 'https://onevpn.bnu.edu.cn/http/77726476706e69737468656265737421eaee478b69326645300d8db9d6562d/taglib/DataTable.jsp'data = f'initQry=0&xktype=2&xh=202261291404&xn=2023&xq=0&nj=2022&pycc=2&dwh=AF&zydm=AF025200221000&kclb1=&kclb2=&isbyk=&items=&xnxq={xn}%2C{xq}&btnFilter=%C0%E0%B1%F0%B9%FD%C2%CB&btnSubmit=%CC%E1%BD%BB&sel_pycc={pycc}&sel_nj={nj}&sel_yxb={yxb}&sel_zydm={zydm}&kkdw_range=self&sel_cddwdm=&menucode_current=JW130417'async with session.post(url=url, params=self.params, data=data) as response:if response.status == 200:content = await response.text()data = pd.read_html(content)[0]self.result[dic['index']] = dataasync def main(self):pbar = tqdm(total=self.que.qsize())while True:if self.que.empty():print('任务完成！')breakelse:dic = await self.que.get()async with aiohttp.ClientSession(headers=self.headers, cookies=self.cookies) as session:await self.scrape_url(session, dic)pbar.update(1)def eventloop(self):loop = asyncio.get_event_loop()loop.run_until_complete(self.main())cookies = {}
headers = {}
params = {}data = GetCurriculum(lst, cookies=cookies, headers=headers, params=params)for key, val in data.result.items():val['index'] = key
df_5 = pd.concat(data.result.values())df = df.merge(df_5, on='index', how='outer')
df.to_excel('完整数据集.xlsx', index=False)