python爬虫-提高爬虫效率

httpx

1	比 requests 更强大 Python 库，让你的爬虫效率提高一倍

什么是协程？

简单来说，协程是一种基于线程之上，但又比线程更加轻量级的存在。对于系统内核来说，协程具有不可见的特性，所以这种由 程序员自己写程序来管理 的轻量级线程又常被称作 "用户空间线程"

协程比多线程好在哪呢？

1
2
3

1. 线程的控制权在操作系统手中，而 协程的控制权完全掌握在用户自己手中，因此利用协程可以减少程序运行时的上下文切换，有效提高程序运行效率。
2. 建立线程时，系统默认分配给线程的 栈 大小是 1 M，而协程更轻量，接近 1 K 。因此可以在相同的内存中开启更多的协程。
3. 由于协程的本质不是多线程而是单线程，所以不需要多线程的锁机制。因为只有一个线程，也不存在同时写变量而引起的冲突。在协程中控制共享资源不需要加锁，只需要判断状态即可。所以协程的执行效率比多线程高很多，同时也有效避免了多线程中的竞争关系

协程的适用 & 不适用场景

1
2
3

适用场景：协程适用于被阻塞的，且需要大量并发的场景。

不适用场景：协程不适用于存在大量计算的场景（因为协程的本质是单线程来回切换），如果遇到这种情况，还是应该使用其他手段去解决

初探异步 http 框架 httpx

至此我们对 "协程" 应该有了个大概的了解，但故事说到这里，相信有朋友还是满脸疑问："协程" 对于接口测试有什么帮助呢？不要着急，答案就在下面。

相信用过 Python 做接口测试的朋友都对 requests 库不陌生。requests 中实现的 http 请求是同步请求，但其实基于 http 请求 IO 阻塞的特性，非常适合用协程来实现 "异步" http 请求从而提升测试效率。 

相信早就有人注意到了这点，于是在 Github 经过了一番探索后，果不其然，最终寻找到了支持协程 "异步" 调用 http 的开源库: httpx

什么是 httpx

1	httpx 是一个几乎继承了所有 requests 的特性并且支持 "异步" http 请求的开源库。简单来说，可以认为 httpx 是强化版 requests。

安装

1 2	httpx 的安装非常简单，在 Python 3.6 以上的环境执行 pip install httpx

最佳实践

1
2
3

俗话说得好，效率决定成败。我分别使用了 httpx 异步 和 同步 的方式对批量 http 请求进行了耗时比较，来一起看看结果吧～

首先来看看同步 http 请求的耗时表现

同步 http 请求

import asyncio
import httpx
import threading
import time


def sync_main(url, sign):
    response = httpx.get(url).status_code
    print(f'sync_main: {threading.current_thread()}: {sign}: {response}')


sync_start = time.time()
[sync_main(url='http://www.baidu.com', sign=i) for i in range(200)]
sync_end = time.time()
print(sync_end - sync_start)

代码比较简单，可以看到在 sync_main 中则实现了同步 http 访问百度 200 次
sync_main: <_MainThread(MainThread, started 4471512512)>: 192: 200
sync_main: <_MainThread(MainThread, started 4471512512)>: 193: 200
sync_main: <_MainThread(MainThread, started 4471512512)>: 194: 200
sync_main: <_MainThread(MainThread, started 4471512512)>: 195: 200
sync_main: <_MainThread(MainThread, started 4471512512)>: 196: 200
sync_main: <_MainThread(MainThread, started 4471512512)>: 197: 200
sync_main: <_MainThread(MainThread, started 4471512512)>: 198: 200
sync_main: <_MainThread(MainThread, started 4471512512)>: 199: 200
16.56578803062439

1
2
3

可以看到在上面的输出中, 主线程没有进行切换（因为本来就是单线程啊喂！）请求按照顺序执行（因为是同步请求）。

程序运行共耗时 16.6 秒

异步 http请求

import asyncio
import httpx
import threading
import time


client = httpx.AsyncClient()


async def async_main(url, sign):
    response = await client.get(url)
    status_code = response.status_code
    print(f'async_main: {threading.current_thread()}: {sign}:{status_code}')



loop = asyncio.get_event_loop()
tasks = [async_main(url='http://www.baidu.com', sign=i) for i in range(200)]
async_start = time.time()
loop.run_until_complete(asyncio.wait(tasks))
async_end = time.time()
loop.close()
print(async_end - async_start)

1	上述代码在 async_main 中用 async await 关键字实现了"异步" http，通过 asyncio ( 异步 io 库请求百度首页 200 次并打印出了耗时

async_main: <_MainThread(MainThread, started 4471512512)>: 56: 200
async_main: <_MainThread(MainThread, started 4471512512)>: 99: 200
async_main: <_MainThread(MainThread, started 4471512512)>: 67: 200
async_main: <_MainThread(MainThread, started 4471512512)>: 93: 200
async_main: <_MainThread(MainThread, started 4471512512)>: 125: 200
async_main: <_MainThread(MainThread, started 4471512512)>: 193: 200
async_main: <_MainThread(MainThread, started 4471512512)>: 100: 200

            4.518340110778809

1	可以看到顺序虽然是乱的（56，99，67...） (这是因为程序在协程间不停切换) 但是主线程并没有切换（协程本质还是单线程）

程序共耗时 4.5 秒

比起同步请求耗时的 16.6 秒 缩短了接近 73 %！

俗话说得好，一步快，步步快。 在耗时方面，"异步" http 确实比同步 http 快了很多。当然，"协程" 不仅仅能在请求效率方面赋能接口测试， 掌握 "协程"后，相信小伙伴们的技术水平也能提升一个台阶，从而设计出更优秀的测试框架

多线程和多进程(不建议使用)

线程池或进程池(适当使用)

import re
import time
import random
import requests
from lxml import etree
from multiprocessing.dummy import Pool


start_time = time.time()

url = "https://www.pearvideo.com/category_3"
headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.88 Safari/537.36"
}

ex = 'srcUrl="(.*?)",vdoUrl=srcUrl'

def request_video(url):
    """
    向视频链接发送请求
    """
    return requests.get(url=url, headers=headers).content

def save_video(content):
    """
    将视频的二进制数据保存到本地
    """
    video_name = str(random.randint(100, 999)) + ".mp4"
    with open(video_name, 'wb') as f:
        f.write(content)

        
# 获取首页源码
page_text = requests.get(url=url, headers=headers).text

tree = etree.HTML(page_text)
li_list = tree.xpath('//ul[@class="listvideo-list clearfix"]/li')

video_url_list = list()
for li in li_list:
    detail_url = "https://www.pearvideo.com/" + li.xpath('./div/a/@href')[0]
    
    # 获取该视频页面的源码
    detail_page_text = requests.get(url=detail_url, headers=headers).text
    
    # 正则匹配视频的URL
    video_url = re.findall(ex, detail_page_text, re.S)[0]
    video_url_list.append(video_url)
    
pool = Pool(4)
    
#使用线程池将视频的二进制数据下载下来
content_list = pool.map(request_video, video_url_list)
    
# 使用线程池将视频的二进制数据保存到本地
pool.map(save_video, content_list)    

print("执行耗时: ", time.time() - start_time)

单线程+异步协程(爬虫推荐使用）

单线程

import time

start_time = time.time()

def request(url):
    print("正在下载", url)
    time.sleep(2)
    print("下载完成", url)

url_list = [
    "https://www.baidu.com",
    "https://www.taobao.com",
    "https://www.jd.com"
]

for url in url_list:
    request(url)

print("执行耗时:", time.time() - start_time)

进程池或线程池

import time
from multiprocessing.dummy import Pool

start_time = time.time()

def request(url):
    print("正在下载", url)
    time.sleep(2)
    print("下载完成", url)

url_list = [
    "https://www.baidu.com",
    "https://www.taobao.com",
    "https://www.jd.com"
]

pool = Pool(3)

pool.map(request, url_list)

print("执行耗时:", time.time() - start_time)

协程

优质项目

协程相关的概念

- event_loop：事件循环，相当于一个无限循环，我们可以把一些函数注册到这个事件循环上，当满足某些条件的时候，函数就会被循环执行。程序是按照设定的顺序从头执行到尾，运行的次数也是完全按照设定。当在编写异步程序时，必然其中有部分程序的运行耗时是比较久的，需要先让出当前程序的控制权，让其在背后运行，让另一部分的程序先运行起来。当背后运行的程序完成后，也需要及时通知主程序已经完成任务可以进行下一步操作，但这个过程所需的时间是不确定的，需要主程序不断的监听状态，一旦收到了任务完成的消息，就开始进行下一步。loop就是这个持续不断的监视器。

- coroutine：中文翻译叫协程，在 Python 中常指代为协程对象类型，我们可以将协程对象注册到事件循环中，它会被事件循环调用。我们可以使用 async 关键字来定义一个方法，这个方法在调用时不会立即被执行，而是返回一个协程对象。

- task：任务，它是对协程对象的进一步封装，包含了任务的各个状态。

- future：代表将来执行或还没有执行的任务，实际上和 task 没有本质区别。

- 另外我们还需要了解 async/await 关键字，它是从 Python 3.5 才出现的，专门用于定义协程。其中，async 定义一个协程，await 用来使挂起阻塞方法的执行。

创建一个协程对象

创建一个协程对象，只需要在函数名前加上一个async关键字即可
import time

async def request():
    print("正在下载")
    time.sleep(2)
    print("下载完成")

c = request()

基本使用

import time
import asyncio


async def request():
    print("正在下载")
    time.sleep(2)
    print("下载完成")

# 创建一个协程对象
c = request()

# 创建一个事件循环对象
loop = asyncio.get_event_loop()


# 1. 将协程对象注册到事件循环中
# 2. 执行事件循环
loop.run_until_complete(c)

task的使用

import time
import asyncio


async def request():
    print("下载成功~~")

# 创建一个协程对象
c = request()

# 创建一个事件循环对象
loop = asyncio.get_event_loop()

# 通过事件循环对象创建一个任务, 并将协程对象封装进这个任务里面 //作用：可以监听它的状态
task = loop.create_task(c)
print(task)

# 1. 将协程对象注册到事件循环中
# 2. 执行事件循环
loop.run_until_complete(task)
print(task)

future的使用

import time
import asyncio


async def request():
    print("下载成功~~")

# 创建一个协程对象
c = request()

# 创建一个事件循环对象
loop = asyncio.get_event_loop()

# 通过future创建任务，就不再依附于loop对象，通过asyncio模块来创建
future = asyncio.ensure_future(c)
print(future)

# 1. 将协程对象注册到事件循环中
# 2. 执行事件循环
loop.run_until_complete(future)
print(future)

回调函数

import time
import asyncio

def call_back(res):
    # 通过res.result()就可以接收到requests发送请求返回的响应数据
    print("在这里进行数据解析", res.result())
    

async def request():
    print("下载成功~~")
    return "永灵大神"

# 创建一个协程对象
c = request()

# 创建一个事件循环对象
loop = asyncio.get_event_loop()

# 通过future创建任务，就不再依附于loop对象，通过asyncio模块来创建
future = asyncio.ensure_future(c)

# 通过future定义一个回调函数，这样的话，协程对象的返回值就可以传递给我们的回调函数
future.add_done_callback(call_back)


# 1. 将协程对象注册到事件循环中
# 2. 执行事件循环
loop.run_until_complete(future)

回调函数(真正发送请求)

import time
import requests
import asyncio

headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.88 Safari/537.36"
}

def call_back(res):
    # 通过res.result()就可以接收到requests发送请求返回的响应数据
    page_text = res.result()
    print("在这里进行数据解析", page_text)
    

async def request():
    return requests.get(url="https://www.baidu.com", headers=headers).text

# 创建一个协程对象
c = request()

# 创建一个事件循环对象
loop = asyncio.get_event_loop()

# 通过future创建任务，就不再依附于loop对象，通过asyncio模块来创建
future = asyncio.ensure_future(c)

# 通过future定义一个回调函数，这样的话，协程对象的返回值就可以传递给我们的回调函数
future.add_done_callback(call_back)


# 1. 将协程对象注册到事件循环中
# 2. 执行事件循环
loop.run_until_complete(future)

多任务异步协程

多任务异步协程的基本使用

import time
import asyncio

start_time = time.time()

# 特殊函数
async def request(url):
    print("正在下载: ", url)
    time.sleep(2)  # time模块为非异步的模块，在整个异步的代码中，如果出现了非异步的模块，则会让整个异步失去效果
    print("下载完成: ", url)
    
url_list = [
    "https://www.baidu.com",
    "https://www.taobao.com",
    "https://www.jd.com"
]

# 创建一个事件循环对象
loop = asyncio.get_event_loop()

task_list = list()
for url in url_list:
    c = request(url)
    
    # 创建一个任务对象
    task = asyncio.ensure_future(c)
    task_list.append(task)
    
# 如果传过来的是一个列表的话，需要再加上一层封装, 使用asyncio.wait()方法进行封装
loop.run_until_complete(asyncio.wait(task_list))
    
print("执行耗时:", time.time() - start_time)

多任务异步协程(解决方案)

import time
import asyncio

start_time = time.time()

# 特殊函数
async def request(url):
    print("正在下载: ", url)
#     time.sleep(2)  # time模块为非异步的模块，在整个异步的代码中，如果出现了非异步的模块，则会让整个异步失去效果
    await asyncio.sleep(2)
    print("下载完成: ", url)
    
url_list = [
    "https://www.baidu.com",
    "https://www.taobao.com",
    "https://www.jd.com"
]

# 创建一个事件循环对象
loop = asyncio.get_event_loop()

task_list = list()
for url in url_list:
    c = request(url)
    
    # 创建一个任务对象
    task = asyncio.ensure_future(c)
    task_list.append(task)
    
# 如果传过来的是一个列表的话，需要再加上一层封装, 使用asyncio.wait()方法进行封装
loop.run_until_complete(asyncio.wait(task_list))
    
print("执行耗时:", time.time() - start_time)

将多任务异步协程应用到爬虫中

import time
import asyncio
import requests

start_time = time.time()

headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.88 Safari/537.36"
}

# 特殊函数
async def request(url):
    print("正在下载: ", url)
    # requests模块为非异步的请求模块，如果使用它，将会使整个异步失去效果
    page_text = requests.get(url=url, headers=headers).text
    print("返回的数据", page_text)
    print("下载完成: ", url)
    
url_list = [
    "http://127.0.0.1:5000/tiger",
    "http://127.0.0.1:5000/tom",
    "http://127.0.0.1:5000/jay"
]

# 创建一个事件循环对象
loop = asyncio.get_event_loop()

task_list = list()
for url in url_list:
    c = request(url)
    
    # 创建一个任务对象
    task = asyncio.ensure_future(c)
    task_list.append(task)
    
# 如果传过来的是一个列表的话，需要再加上一层封装, 使用asyncio.wait()方法进行封装
loop.run_until_complete(asyncio.wait(task_list))
    
print("执行耗时:", time.time() - start_time)

将多任务异步协程应用到爬虫中(解决方案)

import time
import asyncio
import aiohttp
import requests

start_time = time.time()

headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.88 Safari/537.36"
}

# 特殊函数
async def request(url):
    print("正在下载: ", url)
    # 此session非requests.Session, 这个session支持异步的网络并发请求
    
    async with aiohttp.ClientSession() as session:
        async with await session.get(url=url) as response:
            page_text = await response.text()  # text()页面源码 read()二进制数据 json()
            print("返回的数据", page_text)
            
    print("下载完成: ", url)

url_list = [
    "http://127.0.0.1:5000/tiger",
    "http://127.0.0.1:5000/tom",
    "http://127.0.0.1:5000/jay"
]

# 创建一个事件循环对象
loop = asyncio.get_event_loop()

task_list = list()
for url in url_list:
    c = request(url)
    
    # 创建一个任务对象
    task = asyncio.ensure_future(c)
    task_list.append(task)
    
# 如果传过来的是一个列表的话，需要再加上一层封装, 使用asyncio.wait()方法进行封装
loop.run_until_complete(asyncio.wait(task_list))
    
print("执行耗时:", time.time() - start_time)

如何实现数据解析（任务的绑定回调机制）

import time
import asyncio
import aiohttp
import requests

start_time = time.time()

headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.88 Safari/537.36"
}

def call_back(task):
    print("在这里进行数据解析")
    print("I am call_back")
    print(task.result())

# 特殊函数
async def request(url):
    # 此session非requests.Session, 这个session支持异步的网络并发请求
    
    async with aiohttp.ClientSession() as session:
        async with await session.get(url=url) as response:
            page_text = await response.text()  # text()页面源码 read()二进制数据 json()
            print("返回的数据", page_text)
            return page_text

url_list = [
    "http://127.0.0.1:5000/tiger",
    "http://127.0.0.1:5000/tom",
    "http://127.0.0.1:5000/jay"
]

# 创建一个事件循环对象
loop = asyncio.get_event_loop()

task_list = list()
for url in url_list:
    c = request(url)
    
    # 创建一个任务对象
    task = asyncio.ensure_future(c)
    task.add_done_callback(call_back)
    task_list.append(task)
    
# 如果传过来的是一个列表的话，需要再加上一层封装, 使用asyncio.wait()方法进行封装
loop.run_until_complete(asyncio.wait(task_list))
    
print("执行耗时:", time.time() - start_time)

参考文献

1
2
3

https://www.cnblogs.com/youhongliang/p/12177380.html
https://blog.csdn.net/weixin_36691991/article/details/118722041
https://www.sohu.com/a/393563980_120686110

你给世界什么姿态，世界将还你什么样的人生。