高性能的异步爬虫

三种方式:

　　1.多进程多线程(不建议)

　　2.进程池或者线程池(适当)

　　3.单线程+异步协程(推荐)

多进程多线程

占用cpu资源,不建议使用

基于线程池的异步爬虫

from multiprocessing.dummy import Pool
import time

def request(url):
    print('downloading...')
    time.sleep(2)
    print('download!!!')

pool = Pool(3)  #线程池
urls = [
    'www.baidu.com',
    'www.sogou.com',
    'www.taobao.com'
]
start = time.time()
pool.map(request,urls)
print('total times:',time.time()-start)

结果:

downloading...
downloading...
downloading...

download!!!
download!!!
download!!!
total times: 2.0691184997558594

单线程+异步协程

- event_loop: 事件循环,相当于一个无限循环,我们可以把一些特殊的函数注册到事件循环中,当满足某些条件时,函数就会被执行.在编写异步程序时,必然有部分程序的耗时是比较久的,需要先让出当前程序的控制权,让其在背后运行,让后面的程序先运行起来.当背后运行的程序完成后,再通知主程序进行下一步操作.
- coroutine: 中文翻译叫协程.在Python中常代指为协程对象类型.使用async关键字修饰一个普通函数,这个函数在调用时就不会立即执行,而是返回一个协程对象.我们就可以将协程对象(特殊函数)注册到事件循环中.
- task: 任务,是对协程对象的进一步封装,包含了任务的各个状态.
- future: 和task没有本质区别,创建方法不一样而已.
- 另外我们还需要了解 async/await 关键字，它是从 Python 3.5 才出现的，专门用于定义协程。其中，async 定义一个协程，await 用来挂起阻塞方法的执行。

 基本使用

import asyncio
async def hello(name):
    print('hello to :',name)
#获取了一个协程对象
c = hello('bobo')

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

#将协程对象注册到事件循环中，然后启动事件循环对象
loop.run_until_complete(c)

执行结果:

hello to : bobo

task的使用

import asyncio
async def hello(name):
    print('hello to :',name)

c = hello('bobo')
loop = asyncio.get_event_loop()
#就协程进行进一步的封装，封装到了task对象中
task = loop.create_task(c) #基于事件循环对象实现
print(task)
loop.run_until_complete(task)
print(task)


执行结果:

<Task pending coro=<hello() running at <ipython-input-15-250865fd4d0b>:3>>
hello to : bobo
<Task finished coro=<hello() done, defined at <ipython-input-15-250865fd4d0b>:3> result=None>

future的使用

import asyncio
async def hello(name):
    print('hello to :',name)

c = hello('bobo')

task = asyncio.ensure_future(c)
loop = asyncio.get_event_loop()

loop.run_until_complete(task)

绑定回调

def callback(task):
    print('i am callback:',task.result())

import asyncio
async def hello(name):
    print('hello to :',name)
    return name

c = hello('bobo')

task = asyncio.ensure_future(c)
#给任务对象绑定一个回调函数
task.add_done_callback(callback)
loop.run_until_complete(task)


执行结果:

hello to : bobo
i am callback: bobo

多任务异步协程

import asyncio
async def request(url):
    print('正在下载：',url)
    sleep(2) #非异步模块的代码：在此处如果存在非异步操作代码，则会彻底让asyncio失去异步的效果
    print('下载成功：',url)
urls = [
    'www.baidu.com',
    'www.taobao.com',
    'www.sogou.com'
]
start = time.time()
loop = asyncio.get_event_loop()
tasks = [] #任务列表，放置多个任务对象
for url in urls:
    c = request(url)
    task = asyncio.ensure_future(c)
    tasks.append(task)
    
#将多个任务对象对应的列表注册到事件循环中
loop.run_until_complete(asyncio.wait(tasks))
print('总耗时：',time.time()-start)

执行结果:

正在下载： www.baidu.com
下载成功： www.baidu.com
正在下载： www.taobao.com
下载成功： www.taobao.com
正在下载： www.sogou.com
下载成功： www.sogou.com
总耗时： 6.001343250274658

结果发现,并没有实现异步,是因为 time.sleep() 是非异步模块的代码,在协程对象中存在非异步操作代码，则会彻底让asyncio失去异步的效果

asyncio中也有sleep()方法,这个是异步的

import asyncio
async def request(url):
    print('正在下载：',url)
    await asyncio.sleep(2)
    print('下载成功：',url)
urls = [
    'www.baidu.com',
    'www.taobao.com',
    'www.sogou.com'
]
start = time.time()
loop = asyncio.get_event_loop()
tasks = [] #任务列表，放置多个任务对象
for url in urls:
    c = request(url)
    task = asyncio.ensure_future(c)
    tasks.append(task)
    
#将多个任务对象对应的列表注册到事件循环中
loop.run_until_complete(asyncio.wait(tasks))
print('总耗时：',time.time()-start)

执行结果:

正在下载： www.baidu.com
正在下载： www.taobao.com
正在下载： www.sogou.com
下载成功： www.baidu.com
下载成功： www.taobao.com
下载成功： www.sogou.com
总耗时： 2.0011146068573  #生效了

多任务异步操作应用到爬虫中

因为影响爬虫效率的因素有很多,所以为了避免网速等因素的影响,单单测试多任务异步的效果,这里我们自己搭建一个服务器.

服务器端:

 

import requests
async def get_page(url):
    print('正在下载：',url)
    #之所以没有实现异步操作，原因是因为requests模块是一个非异步的模块
    response = requests.get(url=url)
    print('响应数据：',response.text)
    print('下载成功：',url)
start = time.time()
urls = [
    'http://127.0.0.1:5000/jack',
    'http://127.0.0.1:5000/jay',
    'http://127.0.0.1:5000/tom'
]
tasks = []
loop = asyncio.get_event_loop()
for url in urls:
    c = get_page(url)
    task = asyncio.ensure_future(c)
    tasks.append(task)
loop.run_until_complete(asyncio.wait(tasks))
print('总耗时：',time.time()-start)

执行结果:

正在下载： http://127.0.0.1:5000/bobo
响应数据： Hello bobo
下载成功： http://127.0.0.1:5000/bobo
正在下载： http://127.0.0.1:5000/jay
响应数据： Hello jay
下载成功： http://127.0.0.1:5000/jay
正在下载： http://127.0.0.1:5000/tom
响应数据： Hello tom
下载成功： http://127.0.0.1:5000/tom
总耗时： 6.0263447761535645  #无效,原因是因为requests模块是一个非异步的模块

这个时候就要使用另外一个支持异步的网络请求模块了:aiohttp

import aiohttp
import asyncio

async def get_page(url):
    async with aiohttp.ClientSession() as session:
        async with await session.get(url=url) as response:
            page_text = await response.text() #read()  json()
            print(page_text)
start = time.time()
urls = [
    'http://127.0.0.1:5000/jack',
    'http://127.0.0.1:5000/jay',
    'http://127.0.0.1:5000/tom',
    'http://127.0.0.1:5000/jack',
    'http://127.0.0.1:5000/jay',
    'http://127.0.0.1:5000/tom',
    'http://127.0.0.1:5000/jack',
    'http://127.0.0.1:5000/jay',
    'http://127.0.0.1:5000/tom'
]
tasks = []
loop = asyncio.get_event_loop()
for url in urls:
    c = get_page(url)
    task = asyncio.ensure_future(c)
    tasks.append(task)
loop.run_until_complete(asyncio.wait(tasks))
print('总耗时：',time.time()-start)

执行结果:

Hello jack
Hello jay
Hello tom
Hello jay
Hello jack
Hello tom
Hello jack
Hello tom
Hello jay
总耗时： 2.031116008758545

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

import aiohttp
import asyncio
#回调函数：解析响应数据
def callback(task):
    print('this is callback()')
    #获取响应数据
    page_text = task.result()
    print('在回调函数中，实现数据解析')
    
async def get_page(url):
    async with aiohttp.ClientSession() as session:
        async with await session.get(url=url) as response:
            page_text = await response.text() #read()  json()
            return page_text
start = time.time()
urls = [
    'http://127.0.0.1:5000/jack',
    'http://127.0.0.1:5000/jay',
    'http://127.0.0.1:5000/tom',
    'http://127.0.0.1:5000/jack',
    'http://127.0.0.1:5000/jay',
    'http://127.0.0.1:5000/tom',
    'http://127.0.0.1:5000/jack',
    'http://127.0.0.1:5000/jay',
    'http://127.0.0.1:5000/tom'
]
tasks = []
loop = asyncio.get_event_loop()
for url in urls:
    c = get_page(url)
    task = asyncio.ensure_future(c)
    #给任务对象绑定回调函数用于解析响应数据
    task.add_done_callback(callback)
    tasks.append(task)
loop.run_until_complete(asyncio.wait(tasks))
print('总耗时：',time.time()-start)


执行结果:

this is callback()
在回调函数中，实现数据解析
this is callback()
在回调函数中，实现数据解析
this is callback()
在回调函数中，实现数据解析
this is callback()
在回调函数中，实现数据解析
this is callback()
在回调函数中，实现数据解析
this is callback()
在回调函数中，实现数据解析
this is callback()
在回调函数中，实现数据解析
this is callback()
在回调函数中，实现数据解析
this is callback()
在回调函数中，实现数据解析
总耗时： 2.018115758895874