比requests还好用的Python高效库，效率提升一倍

news2024/11/15 12:13:15

Python开发者们最常用的库之一就是requests，它以简单易用的API和强大的功能深受喜爱。然而，随着爬取任务的复杂性增加，requests的局限性逐渐显现。一些更为高效、功能更强大的爬虫库开始进入开发者的视野

那么，有没有比requests更高效、更适合复杂爬虫任务的Python库？这些库又是如何帮助开发者更快速、更智能地进行网页数据采集的？

在众多Python爬虫库中，httpx和aiohttp无疑是两个突出的代表。httpx不仅保留了requests的简洁性，还提供了异步支持，让爬虫任务的并发处理能力大幅提升。而aiohttp则凭借其对异步I/O的出色支持，成为处理大规模爬取任务时的理想选择。比如在一个涉及百万级页面的爬取项目中，使用aiohttp的开发团队将原本需要数天的任务缩短到了数小时，大大提高了效率。

嗨，大家周二好呀，今天看到九月的中秋节和国庆节需要调休时，无语至极...
废话少说，今天来聊聊一个比requests还好用的Python高效爬虫库，httpx。

最近公司在搞Python后端项目重构，整个后端逻辑都变成了“异步”协程的方式实现。看着满屏的async和await，测试内心那叫一个崩溃。之前对协程有点了解，但这次是真正上手了，今天就把学习成果分享给大家。

什么是协程？

协程是一种比线程更轻量级的存在。简单来说，协程是用户空间内的“轻量级线程”，由程序员自己管理。和多线程相比，协程有几个显著的优势：

控制权在用户手中：线程的控制权在操作系统手中，而协程完全由用户控制，这减少了上下文切换的开销，提高了效率。
更轻量：线程需要分配较大的栈空间（通常是MB级别），而协程只需要几KB，因此可以在相同内存中运行更多的协程。
避免锁机制：因为协程是单线程的，不存在多线程的资源竞争问题，不需要加锁，避免了死锁和竞争状态，提高了执行效率。

协程特别适合用于I/O密集型任务，例如网络请求、文件读写等，但不适用于CPU密集型任务。因为协程的本质是单线程，在处理大量计算任务时，还是需要多进程或多线程来分担。

为什么选择httpx？

搞清楚了协程，咱们来看看今天的主角——httpx。httpx是一个支持异步HTTP请求的库，几乎继承了requests的所有特性，可以认为是requests的“升级版”。对于那些已经习惯了requests的开发者来说，httpx的上手几乎没有学习成本。

安装httpx非常简单，只需要在命令行中执行：

pip install httpx

使用httpx进行异步请求

先来看看如何用httpx实现异步HTTP请求吧。

同步请求示例

先回顾一下同步请求的写法，这里用requests库来做个对比：

import requests
import time


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

sync_start = time.time()
[sync_main('http://www.baidu.com', i) for i in range(200)]
sync_end = time.time()
print(f'Total time for sync requests: {sync_end - sync_start} seconds')

这段代码发送了200次同步HTTP请求，并记录了总耗时。运行结果可能如下：

sync_main: 0: 200
...
sync_main: 199: 200
Total time for sync requests: 16.6 seconds

可以看到，同步请求的总耗时约为16.6秒。

异步请求示例

现在，用httpx来实现相同的功能，但这次使用异步请求：

import asyncio
import httpx
import time


async def async_main(url, sign):    
    async with httpx.AsyncClient() as client:        
        response = await client.get(url)        
        print(f'async_main: {sign}: {response.status_code}')

async def main():    
    tasks = [async_main('http://www.baidu.com', i) for i in range(200)]    
    await asyncio.gather(*tasks)

async_start = time.time()
asyncio.run(main())
async_end = time.time()
print(f'Total time for async requests: {async_end - async_start} seconds')

这段代码用httpx实现了异步HTTP请求。运行结果可能如下：

async_main: 0: 200
...
async_main: 199: 200
Total time for async requests: 4.5 seconds

可以看到，异步请求的总耗时约为4.5秒，比同步请求缩短了近73%！

深入理解httpx

httpx不仅仅是requests的替代品，它还提供了一些额外的功能，使其在实际应用中更加强大和灵活。

1. 并发请求

使用httpx，我们可以轻松地实现并发请求，提高数据抓取效率。例如：

import asyncio
import httpx


async def fetch(url):
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.text

async def main(urls):
    tasks = [fetch(url) for url in urls]
    results = await asyncio.gather(*tasks)
    return results

urls = ['http://example.com' for _ in range(100)]
results = asyncio.run(main(urls))
print(f'Fetched {len(results)} pages')

这段代码演示了如何使用httpx并发地抓取多个网页，并将结果存储在一个列表中。通过这种方式，可以显著提高爬虫的效率。

2. 异步上下文管理

httpx提供了异步上下文管理器，确保在异步操作中正确地管理资源。例如：

import httpx
import asyncio


async def fetch(url):
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response

async def main():
    response = await fetch('http://example.com')
    print(response.text)

asyncio.run(main())

在这个示例中，async with语句确保在异步操作完成后正确关闭HTTP客户端。

3. 超时和重试机制

在网络请求中，超时和重试机制是非常重要的。httpx提供了灵活的超时和重试配置：

import httpx
import asyncio


async def fetch(url):
    async with httpx.AsyncClient(timeout=5.0) as client:
        response = await client.get(url)
        return response

async def main():
    try:
        response = await fetch('http://example.com')
        print(response.text)
    except httpx.TimeoutException:
        print('Request timed out')

asyncio.run(main())

这里设置了一个5秒的超时时间，如果请求超过这个时间未完成，将抛出一个TimeoutException。

httpx在实际项目中的应用

虎哥最近在实际项目中使用httpx进行数据抓取，不得不说，效率真的提升了不少。尤其是在处理需要大量并发请求的场景时，httpx简直就是利器。

1. 数据抓取

在一个数据抓取项目中，需要抓取上千个页面的数据。使用httpx后，不仅代码简洁了很多，效率也提高了一倍以上。例如：

import asyncio
import httpx


async def fetch_data(url):
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.json()

async def main(urls):
    tasks = [fetch_data(url) for url in urls]
    results = await asyncio.gather(*tasks)
    for result in results:
        print(result)

urls = [f'http://api.example.com/data/{i}' for i in range(100)]
asyncio.run(main(urls))

2. 接口测试

在接口测试中，httpx的异步特性也大有用武之地。例如，需要并发地测试多个API接口，可以使用以下代码：

import asyncio
import httpx


async def test_api(url):
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        assert response.status_code == 200

async def main(urls):
    tasks = [test_api(url) for url in urls]
    await asyncio.gather(*tasks)

urls = [f'http://api.example.com/test/{i}' for i in range(100)]
asyncio.run(main(urls))