一、背景知识
爬虫的本质就是一个socket客户端与服务端的通信过程,如果我们有多个url待爬取,只用一个线程且采用串行的方式执行,那只能等待爬取一个结束后才能继续下一个,效率会非常低。
需要强调的是:对于单线程下串行N个任务,并不完全等同于低效,如果这N个任务都是纯计算的任务,那么该线程对cpu的利用率仍然会很高,之所以单线程下串行多个爬虫任务低效,是因为爬虫任务是明显的IO密集型程序。
关于IO模型详见链接:http://www.cnblogs.com/linhaifeng/articles/7454717.html
那么该如何提高爬取性能呢?且看下述概念
二、同步、异步、回调机制
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.一个简单的解决方案:多线程或多进程
#在服务器端使用多线程(或多进程)。多线程(或多进程)的目的是让每个连接都拥有独立的线程(或进程),这样任何一个连接的阻塞都不会影响其他的连接。
import requests from threading import Thread,current_thread IO密集型用多线程,计算密集型用多进程 def parse_page(res): print('%s 解析 %s' % (current_thread().getName(),len(res))) def get_page(url): print('%s下载 %s' %(current_thread().getName(),url)) response=requests.get(url) if response.status_code==200: return 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和各种数据库等。
进程池或线程池:异步调用+回调机制
#IO密集型程序应该用多线程,所以此时我们使用线程池 import requests from threading import current_thread from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor def parse_page(res): res=res.result() print('%s 解析 %s' %(current_thread().getName(),len(res))) #current_thread().getName() 获取当前线程名
def get_page(url): print('%s 下载 %s' %(current_thread().getName(),url)) response=requests.get(url) if response.status_code == 200: #响应成功状态码为200时,返回 return response.text if __name__ == '__main__': urls=['https://www.baidu.com/','http://www.sina.com.cn/','https://taobao.com'] pool=ThreadPoolExecutor(50) #开启50个线程或进程 # pool=ProcessPoolExecutor(50) for url in urls: pool.submit(get_page,url).add_done_callback(parse_page) #使用回调函数 pool.shutdown(wait=True)
结果如下:
改进后方案其实也存在问题
线程池和进程池只是在一定程度上缓解了频繁调用IO接口带来的资源占用问题,
而且,所谓‘池’,也始终有上限,当池大大超过上限时,“池”构成的系统对外界的响应并不比没有池的时候效果好多少。
所以使用“池”必须考虑其面临的响应规模,并根据响应规模调整“池”的大小。
对于上面例子中可能会出现成千上百个客户请求时,“”线程池“”和“进程池”或许可以缓解部分压力,但不能解决所有问题,总之,多线程模型可以方便高效的解决小规模的服务请求,但面对大规模的服务请求,多线程模型也会遇到瓶颈,可以用非阻塞接口来尝试解决这个问题。
补充:gevent模块
from gevent import joinall,spawn,monkey;monkey.patch_all() import requests from threading import current_thread def parse_page(res): print('%s PARSE %s' %(current_thread().getName(),len(res))) def get_page(url,callback=parse_page): print('%s GET %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', 'https://www.taobao.com', 'https://www.openstack.org', ] tasks=[] for url in urls: tasks.append(spawn(get_page,url)) joinall(tasks)
结果如下
三、高性能
上面无论哪一种解决方案都没有解决一个性能相关的问题:IO阻塞,无论是多线程还是多进程,在遇到IO时都会被操作系统强行夺走cpu的执行权限,
程序的执行效率因此就降低了下来。
解决这一问题的关键在于我们自己从应用程序级别检测IO阻塞,然后切换到我们程序的其他任务执行,这样把我们程序的IO降到最低,我们的程序处于就绪态就会增多,以此来迷惑操作系统,操作系统便以为我们的程序是IO比较少的程序,从而会尽可能多的分配CPU给我们,这样也就达到了提升程序执行效率的目的。
1、在python3.3之后新增了asyncio模块,可以帮我们检测IO(只能是网络IO),实现应用程序级别的切换
import asyncio @asyncio.coroutine def task(task_id,senconds): print('%s is start' %task_id) yield from asyncio.sleep(senconds) #只能检测网络IO,检测到IO后切换到其他任务执行(一旦遇到IO,用yield from) print('%s is end' %task_id) tasks=[task(task_id="任务1",senconds=3),task("任务2",2),task(task_id="任务3",senconds=1)] loop=asyncio.get_event_loop() loop.run_until_complete(asyncio.wait(tasks)) loop.close()
2、但asyncio模块只能发tcp级别的请求,不能发http协议,因此,在我们需要发送http请求的时候,需要我们自定义http报头
import asyncio import requests import uuid user_agent='Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/49.0.2623.221 Safari/537.36 SE 2.X MetaSr 1.0' def parse_page(host,res): print('%s 解析结果 %s' %(host,len(res))) with open('%s.html' %(uuid.uuid1()),'wb') as f: f.write(res) @asyncio.coroutine def get_page(host,port=80,url='/',callback=parse_page,ssl=False): print('下载 http://%s:%s%s' %(host,port,url)) #步骤一(IO阻塞):发起tcp链接,是阻塞操作,因此需要yield from if ssl: port=443 recv,send=yield from asyncio.open_connection(host=host,port=443,ssl=ssl) # 步骤二:封装http协议的报头,因为asyncio模块只能封装并发送tcp包,因此这一步需要我们自己封装http协议的包 request_headers="""GET %s HTTP/1.0 Host: %s User-agent: %s """ %(url,host,user_agent) # requset_headers="""POST %s HTTP/1.0 Host: %s name=egon&password=123""" % (url, host,) request_headers=request_headers.encode('utf-8') # 步骤三(IO阻塞):发送http请求包 send.write(request_headers) yield from send.drain() # 步骤四(IO阻塞):接收响应头 while True: line=yield from recv.readline() if line == b' ': break print('%s Response headers:%s' %(host,line)) # 步骤五(IO阻塞):接收响应体 text=yield from recv.read() # 步骤六:执行回调函数 callback(host,text) # 步骤七:关闭套接字 send.close() #没有recv.close()方法,因为是四次挥手断链接,双向链接的两端,一端发完数据后执行send.close()另外一端就被动地断开 if __name__ == '__main__': tasks=[ get_page('www.baidu.com',url='/s?wd=美女',ssl=True), get_page('www.cnblogs.com',url='/',ssl=True), ] loop=asyncio.get_event_loop() loop.run_until_complete(asyncio.wait(tasks)) loop.close() asyncio+自定义http协议报头