高性能异步爬虫-阿里云开发者社区

引子：看到这个你是不是想到了多线程，多进程。

1.问题一：

"""
问题1
线程池”或“连接池”或许可以缓解部分压力，
但是不能解决所有问题。
总之，多线程模型可以方便高效的解决小规模的服务请求，
但面对大规模的服务请求，多线程模型也会遇到瓶颈，
可以用非阻塞接口来尝试解决这个问题。
"""

View Code

2.问题二：

"""
问题二：
在 Python 3.5 则增加了 async/await，使得协程的实现更加方便。
首先我们需要了解下面几个概念：

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

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

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

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

View Code

3最简单的一个协程的例子

import asyncio
async def fun(x):
    print(f"协程{x}执行")
coroutine = fun(1) #协程任务
loop = asyncio.get_event_loop() #创建了一个循环事件
loop.run_until_complete(coroutine) #将任务注册到循环事件中

View Code

3.1版本更新

# 在python3.7版本以后
# asyncio.run(coroutine)
=
# loop = asyncio.get_event_loop() #创建了一个循环事件
# loop.run_until_complete(coroutine) #将任务注册到循环事件中

View Code

4.原理解析

"""
函数-coroutine对象-task对象-交给管理器
有三种：
1.  coroutine = fun(1)
    loop = asyncio.get_event_loop()
    loop.run_until_complete(coroutine)
2.  coroutine = fun(1)
    loop = asyncio.get_event_loop()
    task = loop.create_task(coroutine)
    loop.run_until_complete(task)
3.  coroutine = fun(1)
    task = asyncio.ensure_future(coroutine)
    loop = asyncio.get_event_loop()
    loop.run_until_complete(task)
"""

View Code

5.绑定回调函数，代码

import asyncio
async def fun(x):
    print(f"协程{x}执行")
    return "wusen"
def back(task):
    print(f"回调函数{task.result()}")
task = asyncio.ensure_future(fun(1))
task.add_done_callback(back)
loop = asyncio.get_event_loop()
loop.run_until_complete(task)

View Code

6.问题三：多任务异步

"""
多任务协程异步，不可以出现非异步模块的代码，
否则就无法实现真正的异步了。上述案例中的time.sleep就是非异步模块中的代码
"""
# 多任务协程异步
# import asyncio
# import time
# async def request(url):
#     print('正在下载',url)
#     #在异步协程中如果出现了同步模块相关的代码，那么就无法实现异步。
#     # time.sleep(2)
#     #当在asyncio中遇到阻塞操作必须进行手动挂起
#     await asyncio.sleep(2)
#     print('下载完毕',url)
# start = time.time()
# urls = [
#     'www.baidu.com',
#     'www.sogou.com',
#     'www.goubanjia.com'
# ]
# #任务列表：存放多个任务对象
# stasks = []
# for url in urls:
#     c = request(url)
#     task = asyncio.ensure_future(c)
#     stasks.append(task)
# loop = asyncio.get_event_loop()
# #需要将任务列表封装到wait中
# loop.run_until_complete(asyncio.wait(stasks))
# print(time.time()-start)

View Code

7.aiohttp:为了解决requests是非异步模块的问题

"""
requests模块是非异步模块，
要想实现真正的异步必须使用基于异步的网络请求模块
所以这里就需要 aiohttp 派上用场
pip install aiohttp
"""

View Code

8 aiohttp用法

8.1发起请求

async def fetch():
    async with aiohttp.ClientSession() as session:
        async with session.get('https://www.baidu.com') as resposne:
            print(await resposne.text())
loop = asyncio.get_event_loop()
tasks = [fetch(),]
loop.run_until_complete(asyncio.wait(tasks))

View Code

8.2添加请求参数

params = {'key': 'value', 'page': 10}
async def fetch():
    async with aiohttp.ClientSession() as session:
        async with session.get('https://www.baidu.com/s',params=params) as resposne:
            print(await resposne.url)
loop = asyncio.get_event_loop()
tasks = [fetch(),]
loop.run_until_complete(asyncio.wait(tasks))
loop.run_until_complete(asyncio.wait(tasks))

View Code

8.3UA伪装

url = 'http://httpbin.org/user-agent'
headers = {'User-Agent': 'test_user_agent'}
async def fetch():
    async with aiohttp.ClientSession() as session:
        async with session.get(url,headers=headers) as resposne:
            print(await resposne.text())
loop = asyncio.get_event_loop()
tasks = [fetch(),]
loop.run_until_complete(asyncio.wait(tasks))

View Code

8.4自定义cookies

url = 'http://httpbin.org/cookies'
cookies = {'cookies_name': 'test_cookies'}
async def fetch():
    async with aiohttp.ClientSession() as session:
        async with session.get(url,cookies=cookies) as resposne:
            print(await resposne.text())
loop = asyncio.get_event_loop()
tasks = [fetch(),]
loop.run_until_complete(asyncio.wait(tasks))

View Code

8.5post请求参数

url = 'http://httpbin.org'
payload = {'username': 'zhang', 'password': '123456'}
async def fetch():
    async with aiohttp.ClientSession() as session:
        async with session.post(url, data=payload) as resposne:
            print(await resposne.text())
loop = asyncio.get_event_loop()
tasks = [fetch(), ]
loop.run_until_complete(asyncio.wait(tasks))

View Code

8.6设置代理

url = "http://python.org"
async def fetch():
    async with aiohttp.ClientSession() as session:
        async with session.get(url, proxy="http://some.proxy.com") as resposne:
        print(resposne.status)
loop = asyncio.get_event_loop()
tasks = [fetch(), ]
loop.run_until_complete(asyncio.wait(tasks))

View Code

8.7aiohttp多任务例子

#环境安装：pip install aiohttp
#使用该模块中的ClientSession
import requests
import asyncio
import time
import aiohttp
start = time.time()
urls = [
    'http://127.0.0.1:5000/bobo','http://127.0.0.1:5000/jay','http://127.0.0.1:5000/tom',
    'http://127.0.0.1:5000/bobo', 'http://127.0.0.1:5000/jay', 'http://127.0.0.1:5000/tom',
    'http://127.0.0.1:5000/bobo', 'http://127.0.0.1:5000/jay', 'http://127.0.0.1:5000/tom',
    'http://127.0.0.1:5000/bobo', 'http://127.0.0.1:5000/jay', 'http://127.0.0.1:5000/tom',
]
async def get_page(url):
    async with aiohttp.ClientSession() as session:
        #get()、post():
        #headers,params/data,proxy='http://ip:port'
        async with await session.get(url) as response:
            #text()返回字符串形式的响应数据
            #read()返回的二进制形式的响应数据
            #json()返回的就是json对象
            #注意：获取响应数据操作之前一定要使用await进行手动挂起
            page_text = await response.text()
            print(page_text)
tasks = []
for url in urls:
    c = get_page(url)
    task = asyncio.ensure_future(c)
    tasks.append(task)
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))
end = time.time()
print('总耗时:',end-start)

View Code

8.8感悟

代码里面我们使用了 await，后面跟了 get() 方法，在执行这五个协程的时候，如果遇到了 await，那么就会将当前协程挂起，转而去执行其他的协程，直到其他的协程也挂起或执行完毕，再进行下一个协程的执行。

开始运行时，时间循环会运行第一个 task，针对第一个 task 来说，当执行到第一个 await 跟着的 get() 方法时，它被挂起，但这个 get() 方法第一步的执行是非阻塞的，挂起之后立马被唤醒，所以立即又进入执行，创建了 ClientSession 对象，接着遇到了第二个 await，调用了 session.get() 请求方法，然后就被挂起了，由于请求需要耗时很久，所以一直没有被唤醒，好第一个 task 被挂起了，那接下来该怎么办呢？事件循环会寻找当前未被挂起的协程继续执行，于是就转而执行第二个 task 了，也是一样的流程操作，直到执行了第五个 task 的 session.get() 方法之后，全部的 task 都被挂起了。所有 task 都已经处于挂起状态，那咋办？只好等待了。3 秒之后，几个请求几乎同时都有了响应，然后几个 task 也被唤醒接着执行，输出请求结果，最后耗时，3 秒！

怎么样？这就是异步操作的便捷之处，当遇到阻塞式操作时，任务被挂起，程序接着去执行其他的任务，而不是傻傻地等着，这样可以充分利用 CPU 时间，而不必把时间浪费在等待 IO 上。

可见，使用了异步协程之后，我们几乎可以在相同的时间内实现成百上千倍次的网络请求，把这个运用在爬虫中，速度提升可谓是非常可观了。

高性能异步爬虫

热门文章

最新文章

相关课程

相关电子书

相关实验场景