标签：obj map start executor list 多路复用 客户 异步 写入
在编写爬虫时，性能的消耗主要在IO请求中，当单进程单线程模式下请求URL时必然会引起等待，从而使得请求整体变慢。
1. 同步执行
1 import requests 2 3 def fetch_async(url): 4 response = requests.get(url) 5 return response 6 7 8 url_list = [‘http://www.github.com‘, ‘http://www.bing.com‘] 9 10 for url in url_list: 11 fetch_async(url)
2. 多线程执行（多个线程并发执行，时间长短取决于最长的URL请求）
1 from concurrent.futures import ThreadPoolExecutor 2 import requests 3 4 5 def fetch_async(url): 6 response = requests.get(url) 7 return response 8 9 10 url_list = [‘http://www.github.com‘, ‘http://www.bing.com‘] 11 pool = ThreadPoolExecutor(5) 12 for url in url_list: 13 pool.submit(fetch_async, url) 14 pool.shutdown(wait=True)
3. 多进程执行（在CPU核心数足够的情况下，多个进程并行执行，时间长短取决于最长的URL请求，理论上会快于多线程）
1 from concurrent.futures import ProcessPoolExecutor 2 import requests 3 4 def fetch_async(url): 5 response = requests.get(url) 6 return response 7 8 9 url_list = [‘http://www.github.com‘, ‘http://www.bing.com‘] 10 pool = ProcessPoolExecutor(5) 11 for url in url_list: 12 pool.submit(fetch_async, url) 13 pool.shutdown(wait=True)
4. 多线程+回调函数（实现了异步非阻塞，在IO等待的情况下可以做其它事情）
1 from concurrent.futures import ThreadPoolExecutor 2 import requests 3 4 def fetch_async(url): 5 response = requests.get(url) 6 return response 7 8 9 def callback(future): 10 print(future.result()) 11 12 13 url_list = [‘http://www.github.com‘, ‘http://www.bing.com‘] 14 pool = ThreadPoolExecutor(5) 15 for url in url_list: 16 v = pool.submit(fetch_async, url) 17 v.add_done_callback(callback) 18 pool.shutdown(wait=True)
5. 多进程+回调函数（实现了异步非阻塞，在IO等待的情况下可以做其它事情）
1 from concurrent.futures import ProcessPoolExecutor 2 import requests 3 4 5 def fetch_async(url): 6 response = requests.get(url) 7 return response 8 9 10 def callback(future): 11 print(future.result()) 12 13 14 url_list = [‘http://www.github.com‘, ‘http://www.bing.com‘] 15 pool = ProcessPoolExecutor(5) 16 for url in url_list: 17 v = pool.submit(fetch_async, url) 18 v.add_done_callback(callback) 19 pool.shutdown(wait=True)
通过上述代码均可以完成对请求性能的提高，对于多线程和多进程的缺点是在IO阻塞时会造成了线程和进程的浪费，所以异步IO会是首选：
1. asyncio示例一
import asyncio @asyncio.coroutine def func1(): print(‘before...func1......‘) yield from asyncio.sleep(5) print(‘end...func1......‘) tasks = [func1(), func1()] loop = asyncio.get_event_loop() loop.run_until_complete(asyncio.gather(*tasks)) loop.close()
2. asyncio示例二
1 import asyncio 2 3 4 @asyncio.coroutine 5 def fetch_async(host, url=‘/‘): 6 print(host, url) 7 reader, writer = yield from asyncio.open_connection(host, 80) 8 9 request_header_content = """GET %s HTTP/1.0\r\nHost: %s\r\n\r\n""" % (url, host,) 10 request_header_content = bytes(request_header_content, encoding=‘utf-8‘) 11 12 writer.write(request_header_content) 13 yield from writer.drain() 14 text = yield from reader.read() 15 print(host, url, text) 16 writer.close() 17 18 tasks = [ 19 fetch_async(‘www.cnblogs.com‘, ‘/wupeiqi/‘), 20 fetch_async(‘dig.chouti.com‘, ‘/pic/show?nid=4073644713430508&lid=10273091‘) 21 ] 22 23 loop = asyncio.get_event_loop() 24 results = loop.run_until_complete(asyncio.gather(*tasks)) 25 loop.close()
3. asyncio + aiohttp
1 import aiohttp 2 import asyncio 3 4 5 @asyncio.coroutine 6 def fetch_async(url): 7 print(url) 8 response = yield from aiohttp.request(‘GET‘, url) 9 # data = yield from response.read() 10 # print(url, data) 11 print(url, response) 12 response.close() 13 14 15 tasks = [fetch_async(‘http://www.google.com/‘), fetch_async(‘http://www.chouti.com/‘)] 16 17 event_loop = asyncio.get_event_loop() 18 results = event_loop.run_until_complete(asyncio.gather(*tasks)) 19 event_loop.close()
4. asyncio + requests
1 import asyncio 2 import requests 3 4 5 @asyncio.coroutine 6 def fetch_async(func, *args): 7 loop = asyncio.get_event_loop() 8 future = loop.run_in_executor(None, func, *args) 9 response = yield from future 10 print(response.url, response.content) 11 12 13 tasks = [ 14 fetch_async(requests.get, ‘http://www.cnblogs.com/wupeiqi/‘), 15 fetch_async(requests.get, ‘http://dig.chouti.com/pic/show?nid=4073644713430508&lid=10273091‘) 16 ] 17 18 loop = asyncio.get_event_loop() 19 results = loop.run_until_complete(asyncio.gather(*tasks)) 20 loop.close()
5. gevent + requests
1 import gevent 2 3 import requests 4 from gevent import monkey 5 6 monkey.patch_all() 7 8 9 def fetch_async(method, url, req_kwargs): 10 print(method, url, req_kwargs) 11 response = requests.request(method=method, url=url, **req_kwargs) 12 print(response.url, response.content) 13 14 # ##### 发送请求 ##### 15 gevent.joinall([ 16 gevent.spawn(fetch_async, method=‘get‘, url=‘https://www.python.org/‘, req_kwargs={}), 17 gevent.spawn(fetch_async, method=‘get‘, url=‘https://www.yahoo.com/‘, req_kwargs={}), 18 gevent.spawn(fetch_async, method=‘get‘, url=‘https://github.com/‘, req_kwargs={}), 19 ]) 20 21 # ##### 发送请求（协程池控制最大协程数量） ##### 22 # from gevent.pool import Pool 23 # pool = Pool(None) 24 # gevent.joinall([ 25 # pool.spawn(fetch_async, method=‘get‘, url=‘https://www.python.org/‘, req_kwargs={}), 26 # pool.spawn(fetch_async, method=‘get‘, url=‘https://www.yahoo.com/‘, req_kwargs={}), 27 # pool.spawn(fetch_async, method=‘get‘, url=‘https://www.github.com/‘, req_kwargs={}), 28 # ])
6. grequests（封装的 gevent + requests）
1 import grequests 2 3 4 request_list = [ 5 grequests.get(‘http://httpbin.org/delay/1‘, timeout=0.001), 6 grequests.get(‘http://fakedomain/‘), 7 grequests.get(‘http://httpbin.org/status/500‘) 8 ] 9 10 11 # ##### 执行并获取响应列表 ##### 12 # response_list = grequests.map(request_list) 13 # print(response_list) 14 15 16 # ##### 执行并获取响应列表（处理异常） ##### 17 # def exception_handler(request, exception): 18 # print(request,exception) 19 # print("Request failed") 20 21 # response_list = grequests.map(request_list, exception_handler=exception_handler) 22 # print(response_list)
7. Twisted示例
1 from twisted.web.client import getPage, defer 2 from twisted.internet import reactor 3 4 5 def all_done(arg): 6 reactor.stop() 7 8 9 def callback(contents): 10 print(contents) 11 12 13 deferred_list = [] 14 15 url_list = [‘http://www.bing.com‘, ‘http://www.baidu.com‘, ] 16 for url in url_list: 17 deferred = getPage(bytes(url, encoding=‘utf8‘)) 18 deferred.addCallback(callback) 19 deferred_list.append(deferred) 20 21 dlist = defer.DeferredList(deferred_list) 22 dlist.addBoth(all_done) 23 24 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()
9. Twisted更多
1 from twisted.internet import reactor 2 from twisted.web.client import getPage 3 import urllib.parse 4 5 6 def one_done(arg): 7 print(arg) 8 reactor.stop() 9 10 post_data = urllib.parse.urlencode({‘check_data‘: ‘adf‘}) 11 post_data = bytes(post_data, encoding=‘utf8‘) 12 headers = {b‘Content-Type‘: b‘application/x-www-form-urlencoded‘} 13 response = getPage(bytes(‘http://dig.chouti.com/login‘, encoding=‘utf8‘), 14 method=bytes(‘POST‘, encoding=‘utf8‘), 15 postdata=post_data, 16 cookies={}, 17 headers=headers) 18 response.addBoth(one_done) 19 20 reactor.run()
以上均是Python内置以及第三方模块提供异步IO请求模块，使用简便大大提高效率，而对于异步IO请求的本质则是【非阻塞Socket】+【IO多路复用】：
1 import select 2 import socket 3 import time 4 5 6 class AsyncTimeoutException(TimeoutError): 7 """ 8 请求超时异常类 9 """ 10 11 def __init__(self, msg): 12 self.msg = msg 13 super(AsyncTimeoutException, self).__init__(msg) 14 15 16 class HttpContext(object): 17 """封装请求和相应的基本数据""" 18 19 def __init__(self, sock, host, port, method, url, data, callback, timeout=5): 20 """ 21 sock: 请求的客户端socket对象 22 host: 请求的主机名 23 port: 请求的端口 24 port: 请求的端口 25 method: 请求方式 26 url: 请求的URL 27 data: 请求时请求体中的数据 28 callback: 请求完成后的回调函数 29 timeout: 请求的超时时间 30 """ 31 self.sock = sock 32 self.callback = callback 33 self.host = host 34 self.port = port 35 self.method = method 36 self.url = url 37 self.data = data 38 39 self.timeout = timeout 40 41 self.__start_time = time.time() 42 self.__buffer = [] 43 44 def is_timeout(self): 45 """当前请求是否已经超时""" 46 current_time = time.time() 47 if (self.__start_time + self.timeout) < current_time: 48 return True 49 50 def fileno(self): 51 """请求sockect对象的文件描述符，用于select监听""" 52 return self.sock.fileno() 53 54 def write(self, data): 55 """在buffer中写入响应内容""" 56 self.__buffer.append(data) 57 58 def finish(self, exc=None): 59 """在buffer中写入响应内容完成，执行请求的回调函数""" 60 if not exc: 61 response = b‘‘.join(self.__buffer) 62 self.callback(self, response, exc) 63 else: 64 self.callback(self, None, exc) 65 66 def send_request_data(self): 67 content = """%s %s HTTP/1.0\r\nHost: %s\r\n\r\n%s""" % ( 68 self.method.upper(), self.url, self.host, self.data,) 69 70 return content.encode(encoding=‘utf8‘) 71 72 73 class AsyncRequest(object): 74 def __init__(self): 75 self.fds = [] 76 self.connections = [] 77 78 def add_request(self, host, port, method, url, data, callback, timeout): 79 """创建一个要请求""" 80 client = socket.socket() 81 client.setblocking(False) 82 try: 83 client.connect((host, port)) 84 except BlockingIOError as e: 85 pass 86 # print(‘已经向远程发送连接的请求‘) 87 req = HttpContext(client, host, port, method, url, data, callback, timeout) 88 self.connections.append(req) 89 self.fds.append(req) 90 91 def check_conn_timeout(self): 92 """检查所有的请求，是否有已经连接超时，如果有则终止""" 93 timeout_list = [] 94 for context in self.connections: 95 if context.is_timeout(): 96 timeout_list.append(context) 97 for context in timeout_list: 98 context.finish(AsyncTimeoutException(‘请求超时‘)) 99 self.fds.remove(context) 100 self.connections.remove(context) 101 102 def running(self): 103 """事件循环，用于检测请求的socket是否已经就绪，从而执行相关操作""" 104 while True: 105 r, w, e = select.select(self.fds, self.connections, self.fds, 0.05) 106 107 if not self.fds: 108 return 109 110 for context in r: 111 sock = context.sock 112 while True: 113 try: 114 data = sock.recv(8096) 115 if not data: 116 self.fds.remove(context) 117 context.finish() 118 break 119 else: 120 context.write(data) 121 except BlockingIOError as e: 122 break 123 except TimeoutError as e: 124 self.fds.remove(context) 125 self.connections.remove(context) 126 context.finish(e) 127 break 128 129 for context in w: 130 # 已经连接成功远程服务器，开始向远程发送请求数据 131 if context in self.fds: 132 data = context.send_request_data() 133 context.sock.sendall(data) 134 self.connections.remove(context) 135 136 self.check_conn_timeout() 137 138 139 if __name__ == ‘__main__‘: 140 def callback_func(context, response, ex): 141 """ 142 :param context: HttpContext对象，内部封装了请求相关信息 143 :param response: 请求响应内容 144 :param ex: 是否出现异常（如果有异常则值为异常对象；否则值为None） 145 :return: 146 """ 147 print(context, response, ex) 148 149 obj = AsyncRequest() 150 url_list = [ 151 {‘host‘: ‘www.google.com‘, ‘port‘: 80, ‘method‘: ‘GET‘, ‘url‘: ‘/‘, ‘data‘: ‘‘, ‘timeout‘: 5, 152 ‘callback‘: callback_func}, 153 {‘host‘: ‘www.baidu.com‘, ‘port‘: 80, ‘method‘: ‘GET‘, ‘url‘: ‘/‘, ‘data‘: ‘‘, ‘timeout‘: 5, 154 ‘callback‘: callback_func}, 155 {‘host‘: ‘www.bing.com‘, ‘port‘: 80, ‘method‘: ‘GET‘, ‘url‘: ‘/‘, ‘data‘: ‘‘, ‘timeout‘: 5, 156 ‘callback‘: callback_func}, 157 ] 158 for item in url_list: 159 print(item) 160 obj.add_request(**item) 161 162 obj.running()
本文转载自： 银角大王
http://www.cnblogs.com/wupeiqi/articles/6229292.html
标签：obj map start executor list 多路复用 客户 异步 写入
原文地址：http://www.cnblogs.com/OldJack/p/7465140.html
