八、asynicio模块以及爬虫应用asynicio模块(高性能爬虫)

asynicio模块以及爬虫应用asynicio模块(高性能爬虫)

一、背景知识

 爬虫的本质就是一个socket客户端与服务端的通信过程，如果我们有多个url待爬取，只用一个线程且采用串行的方式执行，那只能等待爬取一个结束后才能继续下一个，效率会非常低。

需要强调的是：对于单线程下串行N个任务，并不完全等同于低效，如果这N个任务都是纯计算的任务，那么该线程对cpu的利用率仍然会很高，之所以单线程下串行多个爬虫任务低效，是因为爬虫任务是明显的IO密集型程序。

二、同步、异步、回调机制

1、同步调用：即提交一个任务后就在原地等待任务结束，等到拿到任务的结果后再继续下一行代码，效率低下

import requests

def parse_page(res):
    print('解析 %s' %(len(res)))

def get_page(url):
    print('下载 %s' %url)
    response=requests.get(url)
    if response.status_code == 200:
        return response.text

urls=['https://www.baidu.com/','http://www.sina.com.cn/','https://www.python.org']
for url in urls:
    res=get_page(url) #调用一个任务，就在原地等待任务结束拿到结果后才继续往后执行
    parse_page(res)

2、一个简单的解决方案：多线程或多进程

#在服务器端使用多线程（或多进程）。多线程（或多进程）的目的是让每个连接都拥有独立的线程（或进程），
这样任何一个连接的阻塞都不会影响其他的连接。

#IO密集型程序应该用多线程
import requests
from threading import Thread,current_thread

def parse_page(res):
    print('%s 解析 %s' %(current_thread().getName(),len(res)))

def get_page(url,callback=parse_page):
    print('%s 下载 %s' %(current_thread().getName(),url))
    response=requests.get(url)
    if response.status_code == 200:
        callback(response.text)

if __name__ == '__main__':
    urls=['https://www.baidu.com/','http://www.sina.com.cn/','https://www.python.org']
    for url in urls:
        t=Thread(target=get_page,args=(url,))
        t.start()

   该方案的问题是：

开启多进程或都线程的方式，我们是无法无限制地开启多进程或多线程的：在遇到要同时响应成百上千路的连接请求，
则无论多线程还是多进程都会严重占据系统资源，降低系统对外界响应效率，而且线程与进程本身也更容易进入假死状态。
3、改进方案： 线程池或进程池+异步调用：提交一个任务后并不会等待任务结束，而是继续下一行代码

#很多程序员可能会考虑使用“线程池”或“连接池”。“线程池”旨在减少创建和销毁线程的频率，其维持一定合理数量的线程，
并让空闲的线程重新承担新的执行任务。“连接池”维持连接的缓存池，尽量重用已有的连接、减少创建和关闭连接的频率。
这两种技术都可以很好的降低系统开销，都被广泛应用很多大型系统，如websphere、tomcat和各种数据库等。

 1 #IO密集型程序应该用多线程，所以此时我们使用线程池
 2 import requests
 3 from threading import current_thread
 4 from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
 5 
 6 def parse_page(res):
 7     res=res.result()
 8     print('%s 解析 %s' %(current_thread().getName(),len(res)))
 9 
10 def get_page(url):
11     print('%s 下载 %s' %(current_thread().getName(),url))
12     response=requests.get(url)
13     if response.status_code == 200:
14         return response.text
15 
16 if __name__ == '__main__':
17     urls=['https://www.baidu.com/','http://www.sina.com.cn/','https://www.python.org']
18 
19     pool=ThreadPoolExecutor(50)
20     # pool=ProcessPoolExecutor(50)
21     for url in urls:
22         pool.submit(get_page,url).add_done_callback(parse_page)
23 
24     pool.shutdown(wait=True)

    改进后方案其实也存在着问题：

#“线程池”和“连接池”技术也只是在一定程度上缓解了频繁调用IO接口带来的资源占用。而且，所谓“池”始终有其上限，
当请求大大超过上限时，“池”构成的系统对外界的响应并不比没有池的时候效果好多少。所以使用“池”必须考虑其面临的响应规模，
并根据响应规模调整“池”的大小

对应上例中的所面临的可能同时出现的上千甚至上万次的客户端请求，“线程池”或“连接池”或许可以缓解部分压力，但是不能解决所有问题。

总之，多线程模型可以方便高效的解决小规模的服务请求，但面对大规模的服务请求，多线程模型也会遇到瓶颈，可以用非阻塞接口来尝试解决这个问题。

三、高性能

    上述无论哪种解决方案其实没有解决一个性能相关的问题：IO阻塞，无论是多进程还是多线程，在遇到IO阻塞时都会被操作系统强行剥夺走CPU的执行权限，程序的执行效率因此就降低了下来。

    解决这一问题的关键在于，我们自己从应用程序级别检测IO阻塞然后切换到我们自己程序的其他任务执行，这样把我们程序的IO降到最低，我们的程序处于就绪态就会增多，以此来迷惑操作系统，操作系统便以为我们的程序是IO比较少的程序，从而会尽可能多的分配CPU给我们，这样也就达到了提升程序执行效率的目的

    1、在python3.3之后新增了asyncio模块，可以帮我们检测IO（只能是网络IO），实现应用程序级别的切换

import asyncio
#当程序遇到IO的时候不阻塞了，让这个装饰器去检测有没有IO，当有IO的时候提醒一下，切到其他的地方去
@asyncio.coroutine
def task(task_id,seconds):
    print("%s is start"%task_id)
    yield from asyncio.sleep(seconds)  #自动检测IO，  #遇到IO就切，并且保存状态
    print("%s id end" %task_id)

tasks = [
    task(task_id="任务1",seconds=3),
    task(task_id="任务2",seconds=2),
    task(task_id="任务3",seconds=1),
]
loop = asyncio.get_event_loop()  #创建事件循环
loop.run_until_complete(asyncio.wait(tasks))  #运行事件循环，直到任务完成
loop.close()  #一旦任务结束，就获取到任务的结果

  2、但asyncio模块只能发tcp级别的请求，不能发http协议，因此，在我们需要发送http请求的时候，需要我们自定义http报头

 1 import requests
 2 import asyncio
 3 import uuid
 4 User_Agent='Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/63.0.3239.132 Safari/537.36'
 5 
 6 def parse_page(res):
 7     with open("%s.html"%uuid.uuid1(),"wb") as f:
 8         f.write(res)
 9 
10 def get_pager(host,port=80,url="/",ssl=False,callback=parse_page):
11 
12     #1、建立连接
13     if ssl:
14         port = 443
15     print("下载：https:%s:%s:%s"%(host,port,url))
16     recv,send = yield from asyncio.open_connection(host=host,port=port,ssl=ssl)
17 
18     #2、封装请求头
19     request_headers="""GET %s HTTP/1.0
Host: %s
User-Agent: %s

""" %(url,host,User_Agent)# http / 1.0省去了拼接太多的东西
20     request_headers=request_headers.encode('utf-8')
21 
22     #3、发送请求头
23     send.write(request_headers) #套接字不能发字符串，要发bytes
24     yield from send.drain() # 发送请求头 #遇到IO就切，并且保存状态
25     #4、接收响应头
26     # recv.read()  # 接收全部的，但是不能区分响应头和响应体
27     # recv.readline()  # 一次收一行，但是你也不确定一次收几行,所以搞个循环
28     while True:
29         line = yield from recv.readline()
30         if line == b'
':  # 最后一行是
,就结束了
31             break
32     #5、接受响应体
33     text = yield from recv.read()
34     #6、调用回调函数，完成解析功能
35     #看一下效果，保存起来，吧返回的值给一个回调函数
36     callback(text)
37     #7、关闭连接
38     send.close()
39     # 三次握手建立好之后，一定是四次之后才断开连接
40     # 发送端决定接受数据的什么时候关闭，
41     # 没有recv.close()
42 
43 if __name__ == '__main__':
44     tasks = [
45         get_pager(host='www.baidu.com', url='/s?wd=唐诗三百首', ssl=True),
46         get_pager(host='www.cnblogs.com', url='/haiyan123/p/7445542.html', ssl=True)
47     ]
48     loop = asyncio.get_event_loop()
49     loop.run_until_complete(asyncio.wait(tasks))
50     loop.close()

  3、自定义http报头多少有点麻烦，于是有了aiohttp模块，专门帮我们封装http报头，然后我们还需要用asyncio检测IO实现切换

 1 import aiohttp
 2 import asyncio
 3 
 4 @asyncio.coroutine
 5 def get_page(url):
 6     print('GET:%s' %url)
 7     response=yield from aiohttp.request('GET',url)
 8 
 9     data=yield from response.read()
10 
11     print(url,data)
12     response.close()
13     return 1
14 
15 tasks=[
16     get_page('https://www.python.org/doc'),
17     get_page('https://www.cnblogs.com/linhaifeng'),
18     get_page('https://www.openstack.org')
19 ]
20 
21 loop=asyncio.get_event_loop()
22 results=loop.run_until_complete(asyncio.gather(*tasks))
23 loop.close()
24 
25 print('=====>',results) #[1, 1, 1]

 4、此外，还可以将requests.get函数传给asyncio，就能够被检测了

 1 import requests
 2 import asyncio
 3 
 4 @asyncio.coroutine
 5 def get_page(func,*args):
 6     print('GET:%s' %args[0])
 7     loog=asyncio.get_event_loop()
 8     furture=loop.run_in_executor(None,func,*args)
 9     response=yield from furture
10 
11     print(response.url,len(response.text))
12     return 1
13 
14 tasks=[
15     get_page(requests.get,'https://www.python.org/doc'),
16     get_page(requests.get,'https://www.cnblogs.com/linhaifeng'),
17     get_page(requests.get,'https://www.openstack.org')
18 ]
19 
20 loop=asyncio.get_event_loop()
21 results=loop.run_until_complete(asyncio.gather(*tasks))
22 loop.close()
23 
24 print('=====>',results) #[1, 1, 1]

5、还有之前在协程时介绍的gevent模块

 1 from gevent import monkey;monkey.patch_all()
 2 import gevent
 3 import requests
 4 
 5 def get_page(url):
 6     print('GET:%s' %url)
 7     response=requests.get(url)
 8     print(url,len(response.text))
 9     return 1
10 
11 # g1=gevent.spawn(get_page,'https://www.python.org/doc')
12 # g2=gevent.spawn(get_page,'https://www.cnblogs.com/linhaifeng')
13 # g3=gevent.spawn(get_page,'https://www.openstack.org')
14 # gevent.joinall([g1,g2,g3,])
15 # print(g1.value,g2.value,g3.value) #拿到返回值
16 
17 
18 #协程池
19 from gevent.pool import Pool
20 pool=Pool(2)
21 g1=pool.spawn(get_page,'https://www.python.org/doc')
22 g2=pool.spawn(get_page,'https://www.cnblogs.com/linhaifeng')
23 g3=pool.spawn(get_page,'https://www.openstack.org')
24 gevent.joinall([g1,g2,g3,])
25 print(g1.value,g2.value,g3.value) #拿到返回值

6、封装了gevent+requests模块的grequests模块

1 #pip3 install grequests
 2 
 3 import grequests
 4 
 5 request_list=[
 6     grequests.get('https://wwww.xxxx.org/doc1'),
 7     grequests.get('https://www.cnblogs.com/linhaifeng'),
 8     grequests.get('https://www.openstack.org')
 9 ]
10 
11 
12 ##### 执行并获取响应列表 #####
13 # response_list = grequests.map(request_list)
14 # print(response_list)
15 
16 ##### 执行并获取响应列表（处理异常） #####
17 def exception_handler(request, exception):
18     # print(request,exception)
19     print("%s Request failed" %request.url)
20 
21 response_list = grequests.map(request_list, exception_handler=exception_handler)
22 print(response_list)

 7、twisted：是一个网络框架，其中一个功能是发送异步请求，检测IO并自动切换

 1 from twisted.web.client import getPage,defer
 2 from twisted.internet import reactor
 3 #pip install pypiwin32
 4 
 5 
 6 def all_done(res):  #这里的res ，接收的是所有函D:pywin32+twistedTwisted-17.9.0-cp36-cp36m-win_amd64.whl数的返回值，
 7     '''等到所有的任务都结束了才触发这个函数'''
 8     print(res) # #打印结果[(回调函数是否抛出异常<True,False>,回调函数的返回值),(),()]
 9     reactor.stop()
10 
11 def callback(res):
12     print(len(res))  #obj
13     return 1
14 
15 
16 urls = [
17     'http://www.baidu.com',
18     'http://www.bing.com',
19     'http://www.python.org',
20 ]
21 task = []
22 for url in urls:
23     obj = getPage(url.encode("utf-8"),)  #请求url页面，要传bytes类型的
24     obj.addCallback(callback)  #吧结果给了回调函数
25     task.append(obj)
26 # defer.Deferred(task)  #创建循环，开始检测IO
27 defer.DeferredList(task).addBoth(all_done)  #给所有任务绑定一个回调函数，等所有的任务都结束了以后关闭连接
28 reactor.run()

8、tornado

 1 from tornado.httpclient import AsyncHTTPClient
 2 from tornado.httpclient import HTTPRequest
 3 from tornado import ioloop
 4 
 5 
 6 def handle_response(response):
 7     """
 8     处理返回值内容（需要维护计数器，来停止IO循环），调用 ioloop.IOLoop.current().stop()
 9     :param response:
10     :return:
11     """
12     if response.error:
13         print("Error:", response.error)
14     else:
15         print(response.body)
16 
17 
18 def func():
19     url_list = [
20         'http://www.baidu.com',
21         'http://www.bing.com',
22     ]
23     for url in url_list:
24         print(url)
25         http_client = AsyncHTTPClient()
26         http_client.fetch(HTTPRequest(url), handle_response)
27 
28 
29 ioloop.IOLoop.current().add_callback(func)
30 ioloop.IOLoop.current().start()
31 
32 
33 
34 
35 #发现上例在所有任务都完毕后也不能正常结束，为了解决该问题，让我们来加上计数器
36 from tornado.httpclient import AsyncHTTPClient
37 from tornado.httpclient import HTTPRequest
38 from tornado import ioloop
39 
40 count=0
41 
42 def handle_response(response):
43     """
44     处理返回值内容（需要维护计数器，来停止IO循环），调用 ioloop.IOLoop.current().stop()
45     :param response:
46     :return:
47     """
48     if response.error:
49         print("Error:", response.error)
50     else:
51         print(len(response.body))
52 
53     global count
54     count-=1 #完成一次回调，计数减1
55     if count == 0:
56         ioloop.IOLoop.current().stop() 
57 
58 def func():
59     url_list = [
60         'http://www.baidu.com',
61         'http://www.bing.com',
62     ]
63 
64     global count
65     for url in url_list:
66         print(url)
67         http_client = AsyncHTTPClient()
68         http_client.fetch(HTTPRequest(url), handle_response)
69         count+=1 #计数加1
70 
71 ioloop.IOLoop.current().add_callback(func)
72 ioloop.IOLoop.current().start()