第五篇：爬虫高性能相关

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

1、背景知识

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

须要强调的是：对于单线程下串行N个任务，并不彻底等同于低效，若是这N个任务都是纯计算的任务，那么该线程对cpu的利用率仍然会很高，之因此单线程下串行多个爬虫任务低效，是由于爬虫任务是明显的IO密集型程序。

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

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

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)

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

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

#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()

   该方案的问题是：

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

#不少程序员可能会考虑使用“线程池”或“链接池”。“线程池”旨在减小建立和销毁线程的频率，其维持必定合理数量的线程，
并让空闲的线程从新承担新的执行任务。“链接池”维持链接的缓存池，尽可能重用已有的链接、减小建立和关闭链接的频率。
这两种技术均可以很好的下降系统开销，都被普遍应用不少大型系统，如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接口带来的资源占用。并且，所谓“池”始终有其上限，
当请求大大超过上限时，“池”构成的系统对外界的响应并不比没有池的时候效果好多少。因此使用“池”必须考虑其面临的响应规模，
并根据响应规模调整“池”的大小

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

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

3、高性能

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

    解决这一问题的关键在于，咱们本身从应用程序级别检测IO阻塞而后切换到咱们本身程序的其余任务执行，这样把咱们程序的IO降到最低，咱们的程序处于就绪态就会增多，以此来迷惑操做系统，操做系统便觉得咱们的程序是IO比较少的程序，从而会尽量多的分配CPU给咱们，这样也就达到了提高程序执行效率的目的

    一、在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()  #一旦任务结束，就获取到任务的结果

  二、但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     #一、创建链接
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     #二、封装请求头
19     request_headers="""GET %s HTTP/1.0\r\nHost: %s\r\nUser-Agent: %s\r\n\r\n""" %(url,host,User_Agent)# http / 1.0省去了拼接太多的东西
20     request_headers=request_headers.encode('utf-8')
21 
22     #三、发送请求头
23     send.write(request_headers) #套接字不能发字符串，要发bytes
24     yield from send.drain() # 发送请求头 #遇到IO就切，而且保存状态
25     #四、接收响应头
26     # recv.read()  # 接收所有的，可是不能区分响应头和响应体
27     # recv.readline()  # 一次收一行，可是你也不肯定一次收几行,因此搞个循环
28     while True:
29         line = yield from recv.readline()
30         if line == b'\r\n':  # 最后一行是\r\n,就结束了
31             break
32     #五、接受响应体
33     text = yield from recv.read()
34     #六、调用回调函数，完成解析功能
35     #看一下效果，保存起来，吧返回的值给一个回调函数
36     callback(text)
37     #七、关闭链接
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()

爬虫应用asyncio模块

  三、自定义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]

asyncio+aiohttp

 四、此外，还能够将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]

asyncio+requests模块的方法

五、还有以前在协程时介绍的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) #拿到返回值

gevent+requests

六、封装了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)

grequests

 七、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+twisted\Twisted-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()

twisted

八、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()

Tornado