快捷導(dǎo)航

python異步IO的項目實踐

更新時間：2023年06月07日 09:01:35 作者：快樂江小魚

本文主要介紹了python異步IO的項目實踐，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧

asyncio是python3.4引入的標準庫，內(nèi)置對異步IO的支持。asyncio的編程模型是一個消息循環(huán)，從asyncio模塊中直接獲取一個EventLoop的引用，然后把需要執(zhí)行的協(xié)程放入EventLoop中執(zhí)行，就實現(xiàn)了異步IO。

協(xié)程又稱為微線程。子程序在所有語言中都是層級調(diào)用，子程序調(diào)用通過棧實現(xiàn)。一個線程就是執(zhí)行一個子程序。協(xié)程最大的優(yōu)勢是執(zhí)行效率高，子程序切換不是線程切換，而是由程序自身控制。第二個優(yōu)勢是不需要多線程的鎖機制。python對協(xié)程的支持通過generator實現(xiàn)，generator中，可以通過for循環(huán)來迭代，也可以不斷調(diào)用next()函數(shù)獲取由yield語句返回的下一個值。

協(xié)程示例，@asyncio.coroutine把一個生成器標記為coroutine類型，自python3.8棄用，使用async def替代。

import threading
import asyncio
@asyncio.coroutine
def hello():
? ? print('Hello World! (%s)' % threading.current_thread())
? ? # yield from調(diào)用另一個生成器
? ? r = yield from asyncio.sleep(1)
? ? print('Hello Again! (%s)' % threading.current_thread())
loop = asyncio.get_event_loop()
# 兩個協(xié)程是由同一個線程并發(fā)執(zhí)行的
tasks = [hello(), hello()]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()

結(jié)果：

Hello World! (<_MainThread(MainThread, started 49300)>)
Hello World! (<_MainThread(MainThread, started 49300)>)
Hello again! (<_MainThread(MainThread, started 49300)>)
Hello again! (<_MainThread(MainThread, started 49300)>)

推薦使用async/await語法編寫協(xié)程應(yīng)用。

# 直接調(diào)用main()并不會執(zhí)行協(xié)程應(yīng)用。
>>> import asyncio
>>> async def main():
...        print('hello')    
...        await asyncio.sleep(1)
...        print('world')
>>> asyncio.run(main())
hello
world

傳統(tǒng)生產(chǎn)消費模型，一個線程寫消息，一個線程讀消息，通過鎖機制控制隊列和等待，但一不小心就可能死鎖。改用協(xié)程生產(chǎn)消息后直接通過yield跳轉(zhuǎn)到消費者開始執(zhí)行，待消費者執(zhí)行完畢后，切換回生產(chǎn)者，效率極高。

# consumer函數(shù)是一個生成器
def consumer():
? ? r = ''
? ? while True:
? ? ? ? n = yield r
? ? ? ? if not n:
? ? ? ? ? ? return?
? ? ? ? print('[COUSUMER] Consuming %s...' % n)
? ? ? ? r = '200 OK'
def produce(c):
? ? # 啟動生成器
? ? c.send(None)
? ? n = 0
? ? while n < 5:
? ? ? ? n = n + 1
? ? ? ? print('[PRODUCER] Producing %s...' % n)
? ? ? ? r = c.send(n)
? ? ? ? print('[PRODUCER] Consumer return:%s' % r)
? ? c.close()
# 注意到consumer函數(shù)是一個generator，把一個consumer傳入produce后：
# 首先調(diào)用c.send(None)啟動生成器；
# 然后，一旦生產(chǎn)了東西，通過c.send(n)切換到consumer執(zhí)行；
# consumer通過yield拿到消息處理，又通過yield把結(jié)果傳回；
# produce拿到consumer處理的結(jié)果，繼續(xù)生產(chǎn)下一條消息；
# produce決定不生產(chǎn)了，通過c.close()關(guān)閉consumer，整個過程結(jié)束。
# 整個流程無鎖，由一個線程執(zhí)行，produce和consumer協(xié)作完成任務(wù)，所以稱為“協(xié)程”，而非線程的搶占式多任務(wù)。 ? ?
c = consumer()
produce(c)

asyncio可以實現(xiàn)單線程并發(fā)IO操作，用單線程+coroutine實現(xiàn)多用戶的高并發(fā)支持，asyncio實現(xiàn)了TCP、UDP、SSL等協(xié)議，aiohttp則是基于asyncio實現(xiàn)的HTTP框架。

import asyncio
from aiohttp import web
def index(request):
? ? return web.Response(body=b'<h1>Index</h1>')
def hello(request):
? ? yield from asyncio.sleep(0.5)
? ? text = '<h1>Hello, %s!</h1>' % request.match_info['name']
? ? return web.Response(body=text.encode('utf-8'))
@asyncio.coroutine
def init(loop):
? ? app = web.Application(loop=loop)
? ? app.router.add_route('GET', '/', index)
? ? app.router.add_router('GET', '/hello/{name}', hello)
? ? srv = yield from loop.create_server(app.make_handler(), '127.0.0.1', 8000)
? ? print('Server started at http://127.0.0.1:8000')
? ? return srv
loop = asyncio.get_event_loop()
loop.run_until_complete(init(loop))
loop.run_forever()

到此這篇關(guān)于python異步IO的項目實踐的文章就介紹到這了,更多相關(guān)python異步IO內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: