高性能的異步爬蟲

三種方式:

　　1.多進程多線程(不建議)

　　2.進程池或者線程池(適當)

　　3.單線程+異步協程(推薦)

多進程多線程

占用cpu資源,不建議使用

基於線程池的異步爬蟲

from multiprocessing.dummy import Pool
import time

def request(url):
    print('downloading...')
    time.sleep(2)
    print('download!!!')

pool = Pool(3)  #線程池
urls = [
    'www.baidu.com',
    'www.sogou.com',
    'www.taobao.com'
]
start = time.time()
pool.map(request,urls)
print('total times:',time.time()-start)

結果:

downloading...
downloading...
downloading...

download!!!
download!!!
download!!!
total times: 2.0691184997558594

單線程+異步協程

- event_loop: 事件循環,相當於一個無限循環,我們可以把一些特殊的函數注冊到事件循環中,當滿足某些條件時,函數就會被執行.在編寫異步程序時,必然有部分程序的耗時是比較久的,需要先讓出當前程序的控制權,讓其在背后運行,讓后面的程序先運行起來.當背后運行的程序完成后,再通知主程序進行下一步操作.
- coroutine: 中文翻譯叫協程.在Python中常代指為協程對象類型.使用async關鍵字修飾一個普通函數,這個函數在調用時就不會立即執行,而是返回一個協程對象.我們就可以將協程對象(特殊函數)注冊到事件循環中.
- task: 任務,是對協程對象的進一步封裝,包含了任務的各個狀態.
- future: 和task沒有本質區別,創建方法不一樣而已.
- 另外我們還需要了解 async/await 關鍵字，它是從 Python 3.5 才出現的，專門用於定義協程。其中，async 定義一個協程，await 用來掛起阻塞方法的執行。

基本使用

import asyncio
async def hello(name):
    print('hello to :',name)
#獲取了一個協程對象
c = hello('bobo')

#創建一個事件循環對象
loop = asyncio.get_event_loop()

#將協程對象注冊到事件循環中，然后啟動事件循環對象
loop.run_until_complete(c)

執行結果:
hello to : bobo

task的使用

import asyncio
async def hello(name):
    print('hello to :',name)

c = hello('bobo')
loop = asyncio.get_event_loop()
#就協程進行進一步的封裝，封裝到了task對象中
task = loop.create_task(c) #基於事件循環對象實現
print(task)
loop.run_until_complete(task)
print(task)


執行結果:
<Task pending coro=<hello() running at <ipython-input-15-250865fd4d0b>:3>>
hello to : bobo
<Task finished coro=<hello() done, defined at <ipython-input-15-250865fd4d0b>:3> result=None>

future的使用

import asyncio
async def hello(name):
    print('hello to :',name)

c = hello('bobo')

task = asyncio.ensure_future(c)
loop = asyncio.get_event_loop()

loop.run_until_complete(task)

綁定回調

def callback(task):
    print('i am callback:',task.result())

import asyncio
async def hello(name):
    print('hello to :',name)
    return name

c = hello('bobo')

task = asyncio.ensure_future(c)
#給任務對象綁定一個回調函數
task.add_done_callback(callback)
loop.run_until_complete(task)


執行結果:
hello to : bobo
i am callback: bobo

多任務異步協程

import asyncio
async def request(url):
    print('正在下載：',url)
    sleep(2) #非異步模塊的代碼：在此處如果存在非異步操作代碼，則會徹底讓asyncio失去異步的效果
    print('下載成功：',url)
urls = [
    'www.baidu.com',
    'www.taobao.com',
    'www.sogou.com'
]
start = time.time()
loop = asyncio.get_event_loop()
tasks = [] #任務列表，放置多個任務對象
for url in urls:
    c = request(url)
    task = asyncio.ensure_future(c)
    tasks.append(task)
    
#將多個任務對象對應的列表注冊到事件循環中
loop.run_until_complete(asyncio.wait(tasks))
print('總耗時：',time.time()-start)

執行結果:

正在下載： www.baidu.com
下載成功： www.baidu.com
正在下載： www.taobao.com
下載成功： www.taobao.com
正在下載： www.sogou.com
下載成功： www.sogou.com
總耗時： 6.001343250274658

結果發現,並沒有實現異步,是因為 time.sleep() 是非異步模塊的代碼,在協程對象中存在非異步操作代碼，則會徹底讓asyncio失去異步的效果

asyncio中也有sleep()方法,這個是異步的

import asyncio
async def request(url):
    print('正在下載：',url)
    await asyncio.sleep(2)
    print('下載成功：',url)
urls = [
    'www.baidu.com',
    'www.taobao.com',
    'www.sogou.com'
]
start = time.time()
loop = asyncio.get_event_loop()
tasks = [] #任務列表，放置多個任務對象
for url in urls:
    c = request(url)
    task = asyncio.ensure_future(c)
    tasks.append(task)
    
#將多個任務對象對應的列表注冊到事件循環中
loop.run_until_complete(asyncio.wait(tasks))
print('總耗時：',time.time()-start)

執行結果:

正在下載： www.baidu.com
正在下載： www.taobao.com
正在下載： www.sogou.com
下載成功： www.baidu.com
下載成功： www.taobao.com
下載成功： www.sogou.com
總耗時： 2.0011146068573  #生效了

多任務異步操作應用到爬蟲中

因為影響爬蟲效率的因素有很多,所以為了避免網速等因素的影響,單單測試多任務異步的效果,這里我們自己搭建一個服務器.

服務器端:

import requests
async def get_page(url):
    print('正在下載：',url)
    #之所以沒有實現異步操作，原因是因為requests模塊是一個非異步的模塊
    response = requests.get(url=url)
    print('響應數據：',response.text)
    print('下載成功：',url)
start = time.time()
urls = [
    'http://127.0.0.1:5000/jack',
    'http://127.0.0.1:5000/jay',
    'http://127.0.0.1:5000/tom'
]
tasks = []
loop = asyncio.get_event_loop()
for url in urls:
    c = get_page(url)
    task = asyncio.ensure_future(c)
    tasks.append(task)
loop.run_until_complete(asyncio.wait(tasks))
print('總耗時：',time.time()-start)

執行結果:

正在下載： http://127.0.0.1:5000/bobo
響應數據： Hello bobo
下載成功： http://127.0.0.1:5000/bobo
正在下載： http://127.0.0.1:5000/jay
響應數據： Hello jay
下載成功： http://127.0.0.1:5000/jay
正在下載： http://127.0.0.1:5000/tom
響應數據： Hello tom
下載成功： http://127.0.0.1:5000/tom
總耗時： 6.0263447761535645  #無效,原因是因為requests模塊是一個非異步的模塊

這個時候就要使用另外一個支持異步的網絡請求模塊了:aiohttp

import aiohttp
import asyncio

async def get_page(url):
    async with aiohttp.ClientSession() as session:
        async with await session.get(url=url) as response:
            page_text = await response.text() #read()  json()
            print(page_text)
start = time.time()
urls = [
    'http://127.0.0.1:5000/jack',
    'http://127.0.0.1:5000/jay',
    'http://127.0.0.1:5000/tom',
    'http://127.0.0.1:5000/jack',
    'http://127.0.0.1:5000/jay',
    'http://127.0.0.1:5000/tom',
    'http://127.0.0.1:5000/jack',
    'http://127.0.0.1:5000/jay',
    'http://127.0.0.1:5000/tom'
]
tasks = []
loop = asyncio.get_event_loop()
for url in urls:
    c = get_page(url)
    task = asyncio.ensure_future(c)
    tasks.append(task)
loop.run_until_complete(asyncio.wait(tasks))
print('總耗時：',time.time()-start)

執行結果:

Hello jack
Hello jay
Hello tom
Hello jay
Hello jack
Hello tom
Hello jack
Hello tom
Hello jay
總耗時： 2.031116008758545

如何實現數據解析--任務的綁定回調機制

import aiohttp
import asyncio
#回調函數：解析響應數據
def callback(task):
    print('this is callback()')
    #獲取響應數據
    page_text = task.result()
    print('在回調函數中，實現數據解析')
    
async def get_page(url):
    async with aiohttp.ClientSession() as session:
        async with await session.get(url=url) as response:
            page_text = await response.text() #read()  json()
            return page_text
start = time.time()
urls = [
    'http://127.0.0.1:5000/jack',
    'http://127.0.0.1:5000/jay',
    'http://127.0.0.1:5000/tom',
    'http://127.0.0.1:5000/jack',
    'http://127.0.0.1:5000/jay',
    'http://127.0.0.1:5000/tom',
    'http://127.0.0.1:5000/jack',
    'http://127.0.0.1:5000/jay',
    'http://127.0.0.1:5000/tom'
]
tasks = []
loop = asyncio.get_event_loop()
for url in urls:
    c = get_page(url)
    task = asyncio.ensure_future(c)
    #給任務對象綁定回調函數用於解析響應數據
    task.add_done_callback(callback)
    tasks.append(task)
loop.run_until_complete(asyncio.wait(tasks))
print('總耗時：',time.time()-start)


執行結果:
this is callback()
在回調函數中，實現數據解析
this is callback()
在回調函數中，實現數據解析
this is callback()
在回調函數中，實現數據解析
this is callback()
在回調函數中，實現數據解析
this is callback()
在回調函數中，實現數據解析
this is callback()
在回調函數中，實現數據解析
this is callback()
在回調函數中，實現數據解析
this is callback()
在回調函數中，實現數據解析
this is callback()
在回調函數中，實現數據解析
總耗時： 2.018115758895874