Bruce_Liuxiaowei

Bruce_Liuxiaowei

关注
发信
关注(28)粉丝(399)

Python编程异步爬虫——aiohttp的使用

异步爬虫—aiohttp的使用

  1. 基本介绍

    asyncio模块其内部实现了对TCP、UDP、SSL协议的异步操作，但是对于HTTP请求来说，就需要用aiohttp实现了。

aiohttp是一个基于asyncio的异步HTTP网络模块，它既提供了服务端，又提供了客户端。requests发起的是同步网络请求，aiohttp则是异步。 aiohttp 模块是一个基于 asyncio 的 HTTP 客户端和服务器框架，可以用于异步处理 HTTP 请求和响应。以下是一个简单的示例代码展示 aiohttp 模块的基本用法：

首先，你需要安装 aiohttp 模块：

pip install aiohttp
  1. 基本示例

接下来，以下是一个简单的示例代码，演示了如何使用 aiohttp 发送一个简单的 GET 请求并获取响应：

import aiohttp
import asyncio

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text(), response.status

async def main():
    async with aiohttp.ClientSession() as session:
        html, status = await fetch(session, 'https://cuiqingcai.com')
        print(f'html:{html[:100]}....')
        print(f'status: {status}')

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
    
  
 运行结果如下：
 html:<!DOCTYPE html>
<html lang="zh-CN">

<head>
  <meta charset="UTF-8">
  <meta name="viewport" content....
status: 200

在这个示例中，我们定义了一个 fetch 函数来发送 GET 请求并返回响应文本，然后使用 main 函数调用 fetch 函数。最后，创建事件循环并运行主函数。

由于网页源码过长，这里只截取了输出的一部分，可以看到，成功获取了网页的源代码及响应状态码200，成功使用aiohttp通过异步方式完成了网页爬取。当然，这个操作用requests也可以做到。

  1. URL参数设置

对于URL参数的设置，可以借助params参数，传入一个字典即可，示例如下：

import aiohttp
import asyncio

async def main():
    params = {'name': 'germey', 'age':25}
    async with aiohttp.ClientSession() as session:
        async with session.get('https://www.httpbin.org/get', params=params) as response:
            print(await response.text())

if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())
    
    
  运行结果如下：
  {
  "args": {
    "age": "25", 
    "name": "germey"
  }, 
  "headers": {
    "Accept": "*/*", 
    "Accept-Encoding": "gzip, deflate", 
    "Host": "www.httpbin.org", 
    "User-Agent": "Python/3.9 aiohttp/3.8.3", 
    "X-Amzn-Trace-Id": "Root=1-65ffa6c9-6d501fa57391d9fc1d7b5f28"
  }, 
  "origin": "58.21.224.175", 
  "url": "https://www.httpbin.org/get?name=germey&age=25"
}

这里看到，实际请求的URL为https://www.httpbin.org/get?name=germey&age=25，其中的参数对应于params的内容。

  1. 其他请求类型

aiohttp还支持其他请求类型，如POST、PUT、DELETE等，这和requests的使用方式有点类似，示例如下：

session.post('http://www.httpbin.org/post', data=b'data')
session.put('http://www.httpbin.org/put', data=b'data')
session.delete('http://www.httpbin.org/delete')
session.head('http://www.httpbin.org/get')
session.options('http://www.httpbin.org/get')
session.patch('http://www.httpbin.org/patch', data=b'data')
  1. POST请求

当使用 aiohttp 模块时，除了发送简单的 GET 请求之外，你还可以处理 POST 请求、设置请求头、处理异常等。以下是一个稍微复杂一点的示例代码，展示了如何发送 POST 请求并处理异常：

import aiohttp
import asyncio

async def main():
    data = {'name': 'germey', 'age':25}
    async with aiohttp.ClientSession() as session:
        async with session.post('https://www.httpbin.org/post', data=data) as response:
            print(await response.text())

if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())
    
    
   运行结果如下：
   {
  "args": {}, 
  "data": "", 
  "files": {}, 
  "form": {
    "age": "25", 
    "name": "germey"
  }, 
  "headers": {
    "Accept": "*/*", 
    "Accept-Encoding": "gzip, deflate", 
    "Content-Length": "18", 
    "Content-Type": "application/x-www-form-urlencoded", 
    "Host": "www.httpbin.org", 
    "User-Agent": "Python/3.9 aiohttp/3.8.3", 
    "X-Amzn-Trace-Id": "Root=1-65ffabc5-3bbf3cdb7cb413a3475ece45"
  }, 
  "json": null, 
  "origin": "58.21.224.175", 
  "url": "https://www.httpbin.org/post"
}

在上面的示例中，我们定义了一个 fetch 函数来发送一个包含 JSON 数据的 POST 请求，并在发生异常时进行处理。在 main 函数中调用 fetch 函数，并根据返回的状态码和响应文本输出相应信息。

通过这个示例，你可以进一步了解如何使用 aiohttp 模块处理 POST 请求和异常情况。你可以根据自己的需求进行进一步的定制和扩展。希望这能帮助你更深入地理解 aiohttp 模块的用法。

对于POST J SON数据提交，其对应的请求头中的Content-Type为application/json，我们只需将post方法里的data参数改成json即可，实例代码如下：

import json
import aiohttp
import asyncio

async def main():
    data = {'name': 'germey', 'age': 25}
    async with aiohttp.ClientSession() as session:
        async with session.post('https://www.httpbin.org/post', json=data) as response:
            print(await response.text())

if __name__ == '__main__':
    print(asyncio.get_event_loop().run_until_complete(main()))

运行结果如下：
    {
  "args": {}, 
  "data": "{\"name\": \"germey\", \"age\": 25}", 
  "files": {}, 
  "form": {}, 
  "headers": {
    "Accept": "*/*", 
    "Accept-Encoding": "gzip, deflate", 
    "Content-Length": "29", 
    "Content-Type": "application/json", 
    "Host": "www.httpbin.org", 
    "User-Agent": "Python/3.9 aiohttp/3.8.4", 
    "X-Amzn-Trace-Id": "Root=1-65ffc207-7d035c6f2b342778797881c9"
  }, 
  "json": {
    "age": 25, 
    "name": "germey"
  }, 
  "origin": "58.21.224.175", 
  "url": "https://www.httpbin.org/post"
}
  1. 响应

我们可以用如下方法分别获取其中的状态码、响应头、响应体、响应体二进制内容、响应体JSON结果，实例代码如下：

import aiohttp
import asyncio

async def main():
    data = {'name': 'germey', 'age': 25}
    async with aiohttp.ClientSession() as session:
        async with session.post('https://www.httpbin.org/post', data=data) as response:
            print('status:', response.status)
            print('headers:', response.headers)
            print('body:', await response.text())
            print('bytes:', await response.read())
            print('json:', await response.json())

if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())

运行结果如下：
status: 200
headers: <CIMultiDictProxy('Date': 'Sun, 24 Mar 2024 06:34:54 GMT', 'Content-Type': 'application/json', 'Content-Length': '510', 'Connection': 'keep-alive', 'Server': 'gunicorn/19.9.0', 'Access-Control-Allow-Origin': '*', 'Access-Control-Allow-Credentials': 'true')>
body: {
  "args": {}, 
  "data": "", 
  "files": {}, 
  "form": {
    "age": "25", 
    "name": "germey"
  }, 
  "headers": {
    "Accept": "*/*", 
    "Accept-Encoding": "gzip, deflate", 
    "Content-Length": "18", 
    "Content-Type": "application/x-www-form-urlencoded", 
    "Host": "www.httpbin.org", 
    "User-Agent": "Python/3.9 aiohttp/3.8.4", 
    "X-Amzn-Trace-Id": "Root=1-65ffc98e-153aad5f0e536d773cd978b4"
  }, 
  "json": null, 
  "origin": "58.21.224.175", 
  "url": "https://www.httpbin.org/post"
}

bytes: b'{\n  "args": {}, \n  "data": "", \n  "files": {}, \n  "form": {\n    "age": "25", \n    "name": "germey"\n  }, \n  "headers": {\n    "Accept": "*/*", \n    "Accept-Encoding": "gzip, deflate", \n    "Content-Length": "18", \n    "Content-Type": "application/x-www-form-urlencoded", \n    "Host": "www.httpbin.org", \n    "User-Agent": "Python/3.9 aiohttp/3.8.4", \n    "X-Amzn-Trace-Id": "Root=1-65ffc98e-153aad5f0e536d773cd978b4"\n  }, \n  "json": null, \n  "origin": "58.21.224.175", \n  "url": "https://www.httpbin.org/post"\n}\n'
json: {'args': {}, 'data': '', 'files': {}, 'form': {'age': '25', 'name': 'germey'}, 'headers': {'Accept': '*/*', 'Accept-Encoding': 'gzip, deflate', 'Content-Length': '18', 'Content-Type': 'application/x-www-form-urlencoded', 'Host': 'www.httpbin.org', 'User-Agent': 'Python/3.9 aiohttp/3.8.4', 'X-Amzn-Trace-Id': 'Root=1-65ffc98e-153aad5f0e536d773cd978b4'}, 'json': None, 'origin': '58.21.224.175', 'url': 'https://www.httpbin.org/post'}

可以看到，有些字段前需要加await，有些则不需要。其原则是，如果返回一个协程对象（如async修饰的方法），那么前面就要加await，具体可以看aiohttp的API，链接为https://docs.aiohttp.org/en/stable/client_reference.html。

  1. 超时设置

我们可以借助ClientTimeout对象设置超时，例如要设置1秒的超时时间，可以这么实现：

import asyncio
import aiohttp

async def main():
    timeout = aiohttp.ClientTimeout(total=2)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        async with session.get('https://www.httpbin.org/get') as response:
            print('status:', response.status)

if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())
    
运行结果如下：
200
  1. 并发限制

由于aiohttp可以支持非常高的并发量，目标网站可能无法在短时间内响应，而且有瞬间将目标网站爬挂掉的危险。

一般情况下，可以借助asyncio的Semaphore来控制并发量，示例代码如下：

import asyncio
import aiohttp

CONCURRENCY = 5
URL = 'https://www.baidu.com'

semaphore = asyncio.Semaphore(CONCURRENCY)
# session = None

async def scrape_api():
    async with semaphore:
        print('scraping',URL)
        async with session.get(URL) as response:
            await asyncio.sleep(1)
            return await response.text()

async def main():
    global session
    session = aiohttp.ClientSession()
    scrape_index_tasks = [asyncio.ensure_future(scrape_api()) for _ in range(100000)]
    await asyncio.gather(*scrape_index_tasks)

if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())\
  
运行结果：
scraping https://www.baidu.com
scraping https://www.baidu.com
scraping https://www.baidu.com
scraping https://www.baidu.com
scraping https://www.baidu.com
....

这里我们声明CONCURRENCY（代表爬取的最大并发量）为5， 同时声明爬取的目标URL为百度。接着，借助Semaphore创建了一个信号量对象，将其赋值为semaphore，这样就可以用它来控制最大并发量了。这里我们把semaphore直接放置在了对应的爬取方法里，使用async with语句将semaphore作为上下文对象即可。信号量便可以控制进入爬取的最大协程数量，即我们声明的CONCURRENCY的值。在main方法里， 我们声明10 000个task，将其传递给gather方法运行，倘若不加以限制，那这10000个task会被同时执行，并发数量相当大。有了信号量的控制后，同时运行的task数量最大会被控制5个，这样就能给aiohttp限制速度了。

03-29 01:48
Powered by LMLPHP ©2024 Bruce_Liuxiaowei 0.001826