爬蟲性能分析

對於爬蟲，python進行並發抓取的實現方式主要有以下幾種：進程，線程，協程。

性能的消耗主要在IO請求中，當單進程單線程模式下請求URL時必然會引起等待，從而使得請求整體變慢。

一 多進程執行

可以實現並發，但是，請求發送出去后和返回之前，中間時期進程空閑

編寫方式：
1- 多進程直接返回處理

from concurrent.futures import ProcessPoolExecutor
import requests
import time

def task(url):
    response = requests.get(url)
    print(url,response)
    # 寫正則表達式
    return response

pool = ProcessPoolExecutor(7)
url_list = [
    'http://www.cnblogs.com/wupeiqi',
    'http://huaban.com/favorite/beauty/',
    'http://www.bing.com',
    'http://www.zhihu.com',
    'http://www.sina.com',
    'http://www.baidu.com',
    'http://www.autohome.com.cn',
]

for url in url_list:
    pool.submit(task,url)

pool.shutdown(wait=True)

2- 多進程通過回調函數處理

from concurrent.futures import ProcessPoolExecutor
import requests
import time

def task(url):
    response = requests.get(url)
    return response

def done(future,*args,**kwargs):
    response = future.result()
    print(response.status_code,response.content)

pool = ProcessPoolExecutor(7)
url_list = [
    'http://www.cnblogs.com/wupeiqi',
    'http://huaban.com/favorite/beauty/',
    'http://www.bing.com',
    'http://www.zhihu.com',
    'http://www.sina.com',
    'http://www.baidu.com',
    'http://www.autohome.com.cn',
]
for url in url_list:
    v = pool.submit(task,url)
    v.add_done_callback(done)

pool.shutdown(wait=True)

二 多線程執行

爬蟲可以實現並發，但是，請求發送出去后和返回之前，中間時期線程空閑。

編寫方式：
1 多線程直接返回處理

from concurrent.futures import ThreadPoolExecutor
import requests
import time

def task(url):
    response = requests.get(url)
    print(url,response)
    # 寫正則表達式

pool = ThreadPoolExecutor(7)
url_list = [
    'http://www.cnblogs.com/wupeiqi',
    'http://huaban.com/favorite/beauty/',
    'http://www.bing.com',
    'http://www.zhihu.com',
    'http://www.sina.com',
    'http://www.baidu.com',
    'http://www.autohome.com.cn',
]
for url in url_list:
    pool.submit(task,url)

pool.shutdown(wait=True)

2 多線程通過回調函數處理

from concurrent.futures import ThreadPoolExecutor
import requests
import time

def task(url):
    """
    下載頁面
    :param url:
    :return:
    """
    response = requests.get(url)
    return response

def done(future,*args,**kwargs):
    response = future.result()
    print(response.status_code,response.content)

pool = ThreadPoolExecutor(7)
url_list = [
    'http://www.cnblogs.com/wupeiqi',
    'http://huaban.com/favorite/beauty/',
    'http://www.bing.com',
    'http://www.zhihu.com',
    'http://www.sina.com',
    'http://www.baidu.com',
    'http://www.autohome.com.cn',
]
for url in url_list:
    v = pool.submit(task,url)
    v.add_done_callback(done)

pool.shutdown(wait=True)

通過上述代碼均可以完成對請求性能的提高，對於多線程和多進行的缺點是在IO阻塞時會造成了線程和進程的浪費

三 異步非阻塞模塊

協程（微線程） + 異步IO---> 1個線程發送N個Http請求

import asyncio
@asyncio.coroutine
def task():
    print('before...task......')
    yield from asyncio.sleep(5) # 發送Http請求，支持TCP獲取結果..
    print('end...task......')


tasks = [task(), task()]

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.gather(*tasks))
loop.close()

import asyncio
@asyncio.coroutine
def task(host, url='/'):
    print('start',host,url)
    reader, writer = yield from asyncio.open_connection(host, 80)

    request_header_content = "GET %s HTTP/1.0\r\nHost: %s\r\n\r\n" % (url, host,)
    request_header_content = bytes(request_header_content, encoding='utf-8')

    writer.write(request_header_content)
    yield from writer.drain()
    text = yield from reader.read()
    print('end',host, url, text)
    writer.close()

tasks = [
    task('www.cnblogs.com', '/gregoryli/'),
    task('dig.chouti.com', '/pic/show?nid=4073644713430508&lid=10273091')
]

loop = asyncio.get_event_loop()
results = loop.run_until_complete(asyncio.gather(*tasks))
loop.close()

import aiohttp
import asyncio
@asyncio.coroutine
def fetch_async(url):
    print(url)
    response = yield from aiohttp.request('GET', url)
    print(url, response)
    response.close()

tasks = [fetch_async('http://www.baidu.com/'), fetch_async('http://www.chouti.com/')]

event_loop = asyncio.get_event_loop()
results = event_loop.run_until_complete(asyncio.gather(*tasks))
event_loop.close()

# -*- coding: utf-8 -*-
# 2017/11/17 14:04
import asyncio
import requests

@asyncio.coroutine
def task(func, *args):
    print(func,args)
    loop = asyncio.get_event_loop()
    future = loop.run_in_executor(None, func, *args) # requests.get('http://www.cnblogs.com/wupeiqi/')
    response = yield from future
    print(response.url, response.content)

tasks = [
    task(requests.get, 'http://www.cnblogs.com/gregoryli/'),
    task(requests.get, 'http://dig.chouti.com/pic/show?nid=4073644713430508&lid=10273091')
]

loop = asyncio.get_event_loop()
results = loop.run_until_complete(asyncio.gather(*tasks))
loop.close()

import gevent
import requests
from gevent import monkey

monkey.patch_all()

def task(method, url, req_kwargs):
    print(method, url, req_kwargs)
    response = requests.request(method=method, url=url, **req_kwargs)
    print(response.url, response.content)

# ##### 發送請求 #####
# gevent.joinall([
#     gevent.spawn(task, method='get', url='https://www.python.org/', req_kwargs={}),
#     gevent.spawn(task, method='get', url='https://www.yahoo.com/', req_kwargs={}),
#     gevent.spawn(task, method='get', url='https://github.com/', req_kwargs={}),
# ])

# ##### 發送請求（協程池控制最大協程數量） #####
from gevent.pool import Pool
pool = Pool(5)
gevent.joinall([
    pool.spawn(task, method='get', url='https://www.python.org/', req_kwargs={}),
    pool.spawn(task, method='get', url='https://www.yahoo.com/', req_kwargs={}),
    pool.spawn(task, method='get', url='https://www.github.com/', req_kwargs={}),
])

import grequests

request_list = [
    grequests.get('http://httpbin.org/delay/1', timeout=0.001),
    grequests.get('http://fakedomain/'),
    grequests.get('http://httpbin.org/status/500')
]

# ##### 執行並獲取響應列表 #####
response_list = grequests.map(request_list,size=5)
print(response_list)

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from twisted.internet import defer
from twisted.web.client import getPage
from twisted.internet import reactor

def one_done(arg):
    print(arg)

def all_done(arg):
    print('done')
    reactor.stop()

@defer.inlineCallbacks
def task(url):
    res = getPage(bytes(url, encoding='utf8')) # 發送Http請求
    res.addCallback(one_done)
    yield res

url_list = [
    'http://www.cnblogs.com',
    'http://www.cnblogs.com',
    'http://www.cnblogs.com',
    'http://www.cnblogs.com',
]

defer_list = [] # [特殊，特殊，特殊(已經向url發送請求)]
for url in url_list:
    v = task(url)
    defer_list.append(v)

d = defer.DeferredList(defer_list)
d.addBoth(all_done)
reactor.run() # 死循環,事件循環

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from tornado.httpclient import AsyncHTTPClient
from tornado.httpclient import HTTPRequest
from tornado import ioloop

COUNT = 0
def handle_response(response):
    global COUNT
    COUNT -= 1
    if response.error:
        print("Error:", response.error)
    else:
        print(response.body)
        # 方法同twisted
        # ioloop.IOLoop.current().stop()

    if COUNT == 0:
        ioloop.IOLoop.current().stop()

def func():
    url_list = [
        'http://www.baidu.com',
        'http://www.bing.com',
    ]
    global COUNT
    COUNT = len(url_list)
    for url in url_list:
        print(url)
        http_client = AsyncHTTPClient()
        http_client.fetch(HTTPRequest(url), handle_response)

ioloop.IOLoop.current().add_callback(func)
ioloop.IOLoop.current().start() # 死循環

- asyncio
- 示例1：asyncio.sleep(5)
- 示例2：自己封裝Http數據包
- 示例3：asyncio+aiohttp
aiohttp模塊：封裝Http數據包 pip3 install aiohttp
- 示例4：asyncio+requests
requests模塊：封裝Http數據包 pip3 install requests
- gevent，greenlet+異步IO
pip3 install greenlet
pip3 install gevent
- 示例1：gevent+requests
- 示例2：gevent（協程池，最多發多少個請求）+requests
- 示例3：gevent+requests => grequests
pip3 install grequests

效率：gevent > Twisted > Tornado > asyncio

四  socket

1. socket客戶端、服務端
連接阻塞
setblocking(0): 無數據（連接無響應；數據未返回）就報錯

------》http請求本質：阻塞

sk = socket.socket()
# 1.連接
sk.connect(('www.baidu.com',80,)) # IO阻塞
print('連接成功了...')

# 2. 連接成功發送消息
sk.send(b'GET / HTTP/1.0\r\nHost:www.baidu.com\r\n\r\n')
# sk.send(b'POST / HTTP/1.0\r\nHost:www.baidu.com\r\n\r\nk1=v1&k2=v2')

# 3. 等待着服務端響應
data = sk.recv(8096) # 響應頭，響應體，IO阻塞
print(data)

# 關閉連接
sk.close()

2. IO多路復用
客戶端：
try:
　　socket對象1.connet()
　　socket對象2.connet()
　　socket對象3.connet()
except Exception as e:
　　pass

while True:
r,w,e = select.select([socket對象1,socket對象2,socket對象3,],[socket對象1,socket對象2,socket對象3,],[],0.05)
r = [socket對象1,] # 表示有人給我發送數據
xx = socket對象1.recv()
w = [socket對象1,] # 表示我已經和別人創建連接成功:
socket對象1.send('"""GET /index HTTP/1.0\r\nHost: baidu.com\r\n\r\n"""')

--------->http請求本質：非阻塞

sk = socket.socket()
sk.setblocking(False)
# 1.連接
try:
    sk.connect(('www.baidu.com',80,)) # IO阻塞
    print('連接成功了...')
except BlockingIOError as e:
    print(e)
# 2. 連接成功發送消息
sk.send(b'GET / HTTP/1.0\r\nHost:www.baidu.com\r\n\r\n')
# sk.send(b'POST / HTTP/1.0\r\nHost:www.baidu.com\r\n\r\nk1=v1&k2=v2')

# 3. 等待着服務端響應
data = sk.recv(8096) # IO阻塞
print(data)

# 關閉連接
sk.close()

3. 
class Foo:
·def fileno(self):
·obj = socket()
·return obj.fileno()
r,w,e = select.select([socket對象？,對象？,對象？,Foo()],[],[])
# 對象必須有： fileno方法，並返回一個文件描述符

要點：

a. select內部：對象.fileno()

b. Foo()內部封裝socket文件描述符

IO多路復用： r,w,e = while 監聽多個socket對象；
異步IO： 非阻塞的socket+IO多路復用
- 非阻塞socket
- select[自己對象],w,r

import socket
import select

class HttpRequest:
    def __init__(self,sk,host,callback):
        self.socket = sk
        self.host = host
        self.callback = callback
    def fileno(self):
        return self.socket.fileno()

class HttpResponse:
    def __init__(self,recv_data):
        self.recv_data = recv_data
        self.header_dict = {}
        self.body = None
        self.initialize()

    def initialize(self):
        headers, body = self.recv_data.split(b'\r\n\r\n', 1)
        self.body = body
        header_list = headers.split(b'\r\n')
        for h in header_list:
            h_str = str(h,encoding='utf-8')
            v = h_str.split(':',1)
            if len(v) == 2:
                self.header_dict[v[0]] = v[1]


class AsyncRequest:
    def __init__(self):
        self.conn = []
        self.connection = [] # 用於檢測是否已經連接成功

    def add_request(self,host,callback):
        try:
            sk = socket.socket()
            sk.setblocking(0)#false
            sk.connect((host,80,))
        except BlockingIOError as e:
            pass
        request = HttpRequest(sk,host,callback)
        self.conn.append(request)
        self.connection.append(request)

    def run(self):

        while True:
            rlist,wlist,elist = select.select(self.conn,self.connection,self.conn,0.05)
            for w in wlist:
                print(w.host,'連接成功...')
                # 只要能循環到，表示socket和服務器端已經連接成功
                tpl = "GET / HTTP/1.0\r\nHost:%s\r\n\r\n"  %(w.host,)
                w.socket.send(bytes(tpl,encoding='utf-8'))
                self.connection.remove(w)
            for r in rlist:
                # r,是HttpRequest
                recv_data = bytes()
                while True:
                    try:
                        chunck = r.socket.recv(8096)
                        recv_data += chunck
                    except Exception as e:
                        break
                print(r.host,"有數據返回...",recv_data)

                response = HttpResponse(recv_data)
                r.callback(response)
                r.socket.close()
                self.conn.remove(r)
            if len(self.conn) == 0:
                break

def f1(response):
    print('保存到文件',response.header_dict)

def f2(response):
    print('保存到數據庫', response.header_dict)

url_list = [
    {'host':'www.baidu.com','callback': f1},
    {'host':'cn.bing.com','callback': f2},
    {'host':'www.cnblogs.com','callback': f2},
]

req = AsyncRequest()
for item in url_list:
    req.add_request(item['host'],item['callback'])

req.run()