48異步編程_asyncio_aiohttp

?

目錄

asyncio. 1

事件循環(huán)：... 1

協(xié)程：... 2

aiohttp：... 4

?

?

?

asyncio

異步io，3.4version加入標(biāo)準(zhǔn)庫(kù)；

asyncio底層基于selectors實(shí)現(xiàn)，看似庫(kù)，其實(shí)是個(gè)框架，包含異步IO、事件循環(huán)、協(xié)程、任務(wù)等內(nèi)容；

?

例：

def a():

??? for x in range(3):

??????? time.sleep(0.001)?? #生產(chǎn)上不能加此句，僅在多線程下做實(shí)驗(yàn)用

??????? print('a.x', x)

?

def b():

??? for x in 'abc':

??????? time.sleep(0.001)

??????? print('b.x', x)

?

# a()?? #串行，兩件事如果有因果關(guān)系才需要串行，否則并行要好

# b()

?

# threading.Thread(target=a).start()?? #多線程下，由OS控制；輸出會(huì)亂

# threading.Thread(target=b).start()

?

if __name__ == '__main__':

??? multiprocessing.Process(target=a).start()?? #真正的同時(shí)進(jìn)行，由OS控制

??? multiprocessing.Process(target=b).start()

?

?

事件循環(huán)：

是asyncio提供的核心運(yùn)行機(jī)制；

?

loop = asyncio.get_event_loop()?? #返回一個(gè)事件循環(huán)對(duì)象，是asyncio.BaseEventLoop的實(shí)例；

loop.stop()、asyncio.AbstractEventLoop.stop()?? #停止運(yùn)行事件循環(huán)

loop.run_forever()、asyncio.AbstractEventLoop.run_forever()?? #一直運(yùn)行，直到stop

loop.run_until_complete(future)、asyncio.AbstractEventLoop.run_until_complete(future)?? #運(yùn)行直至Future對(duì)象運(yùn)行完

loop.close()、asyncio.AbstractEventLoop.close()?? #關(guān)閉事件循環(huán)

loop.is_running()、asyncio.AbstractEventLoop.is_running()?? #返回事件循環(huán)是否運(yùn)行

?

?

協(xié)程：

例：

協(xié)程，有返回值None，但沒(méi)有用；

由自己控制（類(lèi)似于并發(fā)的控制），與多線程、多進(jìn)程沒(méi)關(guān)系；

在線程內(nèi)通過(guò)生成器完成了調(diào)度，讓兩個(gè)函數(shù)幾乎都有在執(zhí)行，這樣的調(diào)度不是OS的進(jìn)程、線程完成的，而是用戶自己設(shè)計(jì)的；

編寫(xiě)此程序，要使用yield來(lái)讓出控制權(quán)，要用循環(huán)幫助執(zhí)行；

def a():

??? for x in range(3):

??????? time.sleep(0.001)

??????? print('a.x', x)

??????? yield

?

def b():

??? for x in 'abc':

??????? time.sleep(0.001)

??????? print('b.x', x)

??????? yield

?

m = a()

n = b()

for _ in range(3):

??? next(m)

??? next(n)

?

不是進(jìn)程，也不是線程，它是用戶空間調(diào)度完成并發(fā)處理的方式；

進(jìn)程、線程由OS完成調(diào)度，而協(xié)程是線程內(nèi)完成調(diào)度，它不需要更多的線程，自然也沒(méi)有多線程切換帶來(lái)的開(kāi)銷(xiāo)；

協(xié)程是非搶占式調(diào)度（串行），只有一個(gè)協(xié)程主動(dòng)讓出控制權(quán)，另一個(gè)協(xié)程才會(huì)被調(diào)度；

協(xié)程也需要使用鎖機(jī)制，因?yàn)槭窃谕粋€(gè)線程中執(zhí)行；

多cpu下，可使用多進(jìn)程+協(xié)程配合，既能進(jìn)程并發(fā)又能發(fā)揮協(xié)程在單線程中的優(yōu)勢(shì)；

py中協(xié)程是基于生成器的；

?

3.5version開(kāi)始，py提供關(guān)鍵字async、await，在語(yǔ)言上原生支持協(xié)程，支持async def、async with、async for；

async def用來(lái)定義協(xié)程函數(shù)，調(diào)用后即是協(xié)程對(duì)象（同生成器函數(shù)、生成器對(duì)象），協(xié)程函數(shù)中可以不包含await、async關(guān)鍵字，不能使用yield關(guān)鍵字；

?

asyncio.iscoroutinefunction(sleep)?? #判斷是否是協(xié)程函數(shù)

asyncio.iscoroutine(thread)?? #判斷是否是協(xié)程對(duì)象

?

例：

@asyncio.coroutine?? #使用裝飾器，3.4ver用法

def sleep(x):?? #生成器+函數(shù)（函數(shù)中有yield語(yǔ)句即為生成器函數(shù)）

??? for i in range(3):?? #有循環(huán)

??????? print('sleep {}'.format(i))

??????? yield from asyncio.sleep(x)?? #asyncio.sleep(x)，另一個(gè)生成器對(duì)象

?

loop = asyncio.get_event_loop()

loop.run_until_complete(sleep(10))?? #sleep(3)必須要有括號(hào)，sleep為生成器函數(shù)，可理解為拿到生成器對(duì)象，用返回的對(duì)象循環(huán)

loop.close()

?

例：

async def sleep(x):?? #3.5ver用法

??? for i in range(3):

??????? print('sleep {}'.format(i))

??????? await asyncio.sleep(x)

?

async def showthread(x):

??? for i in range(3):

??????? print(threading.enumerate())

??????? await asyncio.sleep(x)

?

loop = asyncio.get_event_loop()

tasks = [sleep(3), showthread(3)]?? #放協(xié)程對(duì)象

loop.run_until_complete(asyncio.wait(tasks))

loop.close()

?

print(asyncio.iscoroutinefunction(sleep))

print(asyncio.iscoroutine(showthread))

輸出：

[<_MainThread(MainThread, started 19172)>]

sleep 0

[<_MainThread(MainThread, started 19172)>]

sleep 1

[<_MainThread(MainThread, started 19172)>]

sleep 2

True

False

?

例：

TcpEchoServer

?

?

aiohttp：

>pip install aiohttp

?

對(duì)于socket，在accept()后，關(guān)鍵是recv()和send()；

對(duì)于http，關(guān)鍵是request和response；

?

異步的好處：

沒(méi)有多線程，并發(fā)用多進(jìn)程；

多進(jìn)程 + 協(xié)程，可完成很高的并發(fā)；

?

簡(jiǎn)單函數(shù)-->多線程-->IO復(fù)用-->異步io；

?

例，服務(wù)端：

from aiohttp import web

?

async def indexhandle(request: web.Request):

??? return web.Response(text=request.path, status=201)

?

async def handle(request: web.Request):

??? print(request.match_info)

??? print(request.query_string)

??? return web.Response(status=200,text=request.match_info.get('id', '0000'))

?

app = web.Application()

app.router.add_get('/', indexhandle)

app.router.add_get('/{id}', handle)

web.run_app(app, host='0.0.0.0', port=9999)

輸出：

======== Running on http://0.0.0.0:9999 ========

(Press CTRL+C to quit)

<MatchInfo {'id': '555'}: <ResourceRoute [GET] <DynamicResource? /{id}> -> <function handle at 0x00000000039989D8>>

?

?

例，客戶端：

from aiohttp import ClientSession

import asyncio

?

async def get_html(url: str):

??? async with ClientSession() as session:?? #client與server建立會(huì)話

??????? async with session.get(url) as res:?? #response

??????????? print(res.status)

??????????? print(await res.text())

?

url = 'http://127.0.0.1:9999/555'?? #服務(wù)端

?

loop = asyncio.get_event_loop()

loop.run_until_complete(get_html(url))?? #爬蟲(chóng)的第一步，拿到html頁(yè)面數(shù)據(jù)后，用第三方庫(kù)解析html變成dom樹(shù)，拿到想要的東西

loop.close()

輸出：

200

555

?