爬虫之高性能相关

　　问题：我给你10个的url，你帮我去把10url的网址下载。

传统方式

# 传统串行方式
import requests
import time

urls = ['https://github.com/' for _ in range(10)]
start = time.time()
for url in urls:
    response = requests.get(url)
    # print(response)

spend_time = time.time() - start
print(spend_time)
# 12.493084907531738

一、多进程和多线程实现并发

import time
from concurrent.futures import ProcessPoolExecutor

import requests

start = time.time()


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


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


if __name__ == '__main__':
    url_list = ['https://www.douban.com/' for _ in range(100)]
    with ProcessPoolExecutor(max_workers=10) as pool:
        for url in url_list:
            v = pool.submit(task, url)
            v.add_done_callback(done)
    print(time.time() - start)
    # 11.862671136856079

　　花费时间 11.862671136856079秒

#########编写方式二#########
import requests
from concurrent.futures import ThreadPoolExecutor

pool = ThreadPoolExecutor()


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


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


if __name__ == '__main__':
    start = time.time()
    with ThreadPoolExecutor(max_workers=os.cpu_count()) as pool:
        url_list = ['https://www.douban.com/' for _ in range(100)]
        for url in url_list:
            v = pool.submit(task,url)
            v.add_done_callback(done)
    print(time.time() - start)
    # 8.904985904693604

　　花费时间 8.904985904693604秒

　　由于GIL限制，建议：IO密集的任务，用ThreadPoolExecutor；CPU密集任务，用ProcessPoolExcutor。

二、基于事件循环的异步IO

 1. asyncio + aiohttp

import time
import asyncio

import aiohttp


async def start_request(session, url):
    sem = asyncio.Semaphore(10, loop=loop)
    async with sem:
        print(f'make request to {url}')
        with async_timeout.timeout(60):
            async with session.get(url, verify_ssl=False) as response:
                if response.status == 200:
                    print(response.status)


async def run(urls):
    conn = aiohttp.TCPConnector(ssl=False,
                                limit=60,  # 连接池在windows下不能太大, <500
                                use_dns_cache=True)
    async with aiohttp.ClientSession(connector=conn, loop=loop) as session:
        datas = await asyncio.gather(*[start_request(session, url) for url in urls], return_exceptions=True)
        for ind, url in enumerate(urls):
            if isinstance(datas[ind], Exception):
                print(f"{ind}, {url}: 下载失败 请重新下载:")


if __name__ == '__main__':
    start = time.time()
    urls = (('http://www.baidu.com/') for _ in range(100))
    loop = asyncio.get_event_loop()
    loop.run_until_complete(run(urls))
    print(time.time() - start)
    # 1.4860568046569824

　2. Twisted　

import time

from twisted.internet import defer
from twisted.internet import reactor
from twisted.web.client import getPage

start = time.time()


def one_done(content, arg):
    response = content.decode('utf-8')
    # print(response)
    print(arg)


def all_done(arg):
    reactor.stop()
    print(time.time() - start)


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


url_list = ('http://www.cnblogs.com' for _ in range(100))

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()  # 死循环
# 5.039534091949463

　3. tornado

import time

from tornado.httpclient import AsyncHTTPClient
from tornado.httpclient import HTTPRequest
from tornado import ioloop

COUNT = 0
start = time.time()


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


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


if __name__ == '__main__':
    ioloop.IOLoop.current().add_callback(func)
    ioloop.IOLoop.current().start()  # 死循环
    print(time.time() - start)
    # 3.0621743202209473

以上均是Python内置以及第三方模块提供异步IO请求模块，使用简便大大提高效率，而对于异步IO请求的本质则是【非阻塞Socket】+【IO多路复用】

"""
########http请求本质，IO阻塞########
sk = socket.socket()
#1.连接
sk.connect(('www.baidu.com',80,)) #阻塞
print('连接成功了')
#2.连接成功后发送消息
sk.send(b"GET / HTTP/1.0
Host: baidu.com

")

#3.等待服务端响应
data = sk.recv(8096)#阻塞
print(data) #

区分响应头和影响体

#关闭连接
sk.close()
"""
"""
########http请求本质，IO非阻塞########
sk = socket.socket()
sk.setblocking(False)
#1.连接
try:
    sk.connect(('www.baidu.com',80,)) #非阻塞，但会报错
    print('连接成功了')
except BlockingIOError as e:
    print(e)

#2.连接成功后发送消息
sk.send(b"GET / HTTP/1.0
Host: baidu.com

")

#3.等待服务端响应
data = sk.recv(8096)#阻塞
print(data) #

区分响应头和影响体

#关闭连接
sk.close()
"""

自定义异步非阻塞IO

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'

', 1)
        self.body = body
        header_list = headers.split(b'
')
        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)
            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
Host:%s

"  %(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
                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.youku.com','callback': f1},
    {'host':'v.qq.com','callback': f2},
    {'host':'www.cnblogs.com','callback': f2},
]


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