前言

最近在看部分Python源碼時(shí), 發(fā)現(xiàn)了async 這個(gè)關(guān)鍵字. 查了一下發(fā)現(xiàn)了Python中的協(xié)程.

協(xié)程這玩意, 在GO中我用過啊, 簡(jiǎn)單說, 就是一個(gè)輕量級(jí)的線程嘛, 由語(yǔ)言自己來實(shí)現(xiàn)不同協(xié)程的調(diào)度. 想著Python中可能也是差不多的東西吧. 但是我Google搜了一下, 前面的說明都給出了下面的例子:

def consumer():r = ''while True:n = yield rif not n:returnprint('[CONSUMER] Consuming %s...' % n)r = '200 OK'def produce(c):c.send(None)n = 0while n < 5:n = n + 1print('[PRODUCER] Producing %s...' % n)r = c.send(n)print('[PRODUCER] Consumer return: %s' % r)c.close()c = consumer() produce(c)

這這這, 這是協(xié)程咩? 這不就是一個(gè)生成器咩?

不過你想一下生成器的特征:

• 需要的時(shí)候, 返回一個(gè)值, 并將當(dāng)前函數(shù)內(nèi)部的狀態(tài)保存
• 下次執(zhí)行的時(shí)候, 會(huì)恢復(fù)上次執(zhí)行的環(huán)境并繼續(xù)執(zhí)行

能夠保存運(yùn)行狀態(tài)并在下次執(zhí)行時(shí)恢復(fù), 說他是協(xié)程貌似也沒什么問題哈. 而且, Python中的協(xié)程是跑在同一個(gè)線程中, 也就是串行執(zhí)行的, 所以也不需要加鎖.

協(xié)程

看了上面的生成器, 是不是有一種想法? 這玩意不就是個(gè)生成器么? 為什么要叫協(xié)程? 沒錯(cuò), 就是生成器.

上面也說了, 協(xié)程的特點(diǎn), 就是可以停止當(dāng)前函數(shù)的執(zhí)行并保存當(dāng)前狀態(tài), 并在下次執(zhí)行時(shí)進(jìn)行恢復(fù). 對(duì)于單線程的運(yùn)行來說, 什么時(shí)候需要這種操作呢? 等待的時(shí)候. 比如等待文件打開, 等待鎖, 等待網(wǎng)絡(luò)返回等等. 這時(shí)程序運(yùn)行著也沒什么事做, 就可以先去做其他事情, 等這邊好了再繼續(xù)回來執(zhí)行. 下面以單純的sleep舉例.

自己實(shí)現(xiàn)

我們?nèi)绾问褂迷膟ield生成器來實(shí)現(xiàn)一個(gè)任務(wù)隊(duì)列呢? 我隨便寫了一下:

import timedef yield_sleep(delay):start_time = time.time()while True:if time.time() - start_time < delay:yieldelse:breakdef hello(name):print(f"{name}-1-{time.strftime('%X')}")yield from yield_sleep(1)print(f"{name}-2-{time.strftime('%X')}")# 創(chuàng)建任務(wù) tasks = [hello('one'), hello('two')] while True:if len(tasks) <= 0:break# 復(fù)制數(shù)組, 下方刪除時(shí)能夠正常遍歷copy_tasks = tasks[:]for task in copy_tasks:try:next(task)except StopIteration:# 迭代完成, 刪除元素tasks.remove(task)

簡(jiǎn)單解釋一下, 我們?cè)诿看稳蝿?wù)調(diào)用yield臨時(shí)返回時(shí), 進(jìn)行任務(wù)的輪換. 這樣, 原本需要2s 執(zhí)行的操作, 一共只需要1s 即可. (這里為了說明效果, 只是簡(jiǎn)單實(shí)現(xiàn)了一下. )

哎, 如此一來, 所有支持yield生成器的語(yǔ)言, 其實(shí)都是支持coroutine的呀, 比如我大 PHP, 嘿嘿.

asyncio

在Python 3.4中, 引入了asyncio包. 將異步 IO 的操作進(jìn)行了封裝.

簡(jiǎn)單說, asyncio內(nèi)部, 維護(hù)了一個(gè)任務(wù)隊(duì)列, 在函數(shù)執(zhí)行yield讓出執(zhí)行權(quán)時(shí), 切換到下一個(gè)任務(wù)繼續(xù)執(zhí)行. 嗯, 大概就是這樣.

import asyncio import time@asyncio.coroutine def hello(name):print(f"{name}-1-{time.strftime('%X')}")yield from asyncio.sleep(1)print(f"{name}-2-{time.strftime('%X')}")# 獲取事件隊(duì)列 loop = asyncio.get_event_loop() # 并發(fā)執(zhí)行任務(wù) tasks = [hello('one'), hello('two')] loop.run_until_complete(asyncio.wait(tasks)) loop.close()

可以看到, 函數(shù)在首次執(zhí)行到y(tǒng)ield時(shí), 進(jìn)行了中斷并將執(zhí)行權(quán)讓了出去.

通過asyncio.get_event_loop()方法, 生成一個(gè)事件隊(duì)列

調(diào)用loop.run_until_complete方法運(yùn)行指定的任務(wù)隊(duì)列

asyncio.sleep函數(shù)和我們上面實(shí)現(xiàn)的yield_sleep是一樣的效果. 在yield中斷時(shí), 會(huì)從隊(duì)列中找到另一個(gè)任務(wù)并執(zhí)行

需要注意的一點(diǎn), Python的協(xié)程是需要手動(dòng)讓出執(zhí)行權(quán)的. 這點(diǎn)與Go不同. 也就是說, 發(fā)生協(xié)程切換的時(shí)機(jī)為:

任務(wù)主動(dòng)讓出執(zhí)行權(quán)

任務(wù)執(zhí)行完成

舉個(gè)例子(將上面的例子簡(jiǎn)單修改):

import asyncio import time@asyncio.coroutine def hello(name, delay, not_yield):print(f"{name}-1-{time.strftime('%X')}")# 占用協(xié)程等待if not_yield:time.sleep(delay)else:yield from asyncio.sleep(1)print(f"{name}-2-{time.strftime('%X')}")# 獲取事件隊(duì)列 loop = asyncio.get_event_loop() # 并發(fā)執(zhí)行任務(wù) tasks = [hello('one', 2, False), hello('two', 5, True)] loop.run_until_complete(asyncio.wait(tasks)) loop.close()

可以看到, 盡管協(xié)程1只是想等待1s, 但因?yàn)閰f(xié)程2一直占用這執(zhí)行權(quán)沒有放出來, 故協(xié)程1等到協(xié)程2執(zhí)行結(jié)束后才再次獲得執(zhí)行權(quán), 既5s 后

async/await

在Python 3.5中, 增加了async/await語(yǔ)法糖.

簡(jiǎn)單來說, 將上面的@asyncio.coroutine換成async, 將yield from換成await就行了. 其他不變. 替換后, 上面的代碼就變成了這樣, 意思是一樣的.

import asyncio import timeasync def hello(name):print(f"{name}-1-{time.strftime('%X')}")await asyncio.sleep(1)print(f"{name}-2-{time.strftime('%X')}")# 獲取事件隊(duì)列 loop = asyncio.get_event_loop() # 并發(fā)執(zhí)行任務(wù) tasks = [hello('one'), hello('two')] loop.run_until_complete(asyncio.wait(tasks)) loop.close()

通用方式

前面說的方式是簡(jiǎn)單介紹下實(shí)現(xiàn), 協(xié)程在Python中較為平常的使用方式如下:

import asyncio import timeasync def hello(name):print(f"{name}-1-{time.strftime('%X')}")await asyncio.sleep(1)print(f"{name}-2-{time.strftime('%X')}")async def main():"""創(chuàng)建任務(wù)并執(zhí)行"""# 方案1: 添加并執(zhí)行任務(wù)task1 = asyncio.create_task(hello('one'))task2 = asyncio.create_task(hello('tow'))# 調(diào)用`asyncio.create_task`方法調(diào)用后, 任務(wù)就已經(jīng)存在調(diào)度隊(duì)列中了# 即使沒有手動(dòng)`await`等待, 在協(xié)程切換時(shí)也會(huì)被執(zhí)行. 我們添加`await`只是為了等待所有任務(wù)完成await task1await task2# 等待其中所有協(xié)程執(zhí)行完成, 與分開 await 相同await asyncio.wait({task1, task2})# 等待協(xié)程執(zhí)行. 若指定時(shí)間后還沒有執(zhí)行完畢, 則會(huì)拋出異常await asyncio.wait_for(task1, 1)# 方案2: 批量執(zhí)行協(xié)程. 方案1的簡(jiǎn)化版# 返回值為所有協(xié)程的集合await asyncio.gather(hello('one'),hello('tow'))"""獲取任務(wù)信息"""# 獲取當(dāng)前執(zhí)行的任務(wù)current_task = asyncio.current_task()# 獲取事件循環(huán)中所有未完成的任務(wù)all_task = asyncio.all_tasks()# 關(guān)閉一個(gè)協(xié)程, 不再執(zhí)行task1.cancel()# 獲取任務(wù)的結(jié)果. 若協(xié)程被關(guān)閉了, 會(huì)拋出異常ret = task1.result()asyncio.run(main())

創(chuàng)建并執(zhí)行一個(gè)協(xié)程任務(wù), 這樣就不用操心事件隊(duì)列的問題了, 我們?cè)趍ain函數(shù)中要做的就是創(chuàng)建任務(wù)并執(zhí)行.

注意, 使用關(guān)鍵字async定義的方法, 是一個(gè)協(xié)程對(duì)象, 不能單純調(diào)用hello('1234'), 其實(shí)現(xiàn)是一個(gè)裝飾器, 返回的是一個(gè)coroutine對(duì)象.

---

對(duì)于Python中的協(xié)程, 差不多就是這么個(gè)東西了. 簡(jiǎn)單說, 就是在執(zhí)行 IO 耗時(shí)操作時(shí), 將執(zhí)行權(quán)暫時(shí)讓出, 以活動(dòng)更好的執(zhí)行效率.

但是問題來了, 對(duì)于這種耗時(shí)操作, 我們總不能每次都自己實(shí)現(xiàn)一遍吧. 勿慌, 其實(shí)很多異步操作, 都已經(jīng)有了實(shí)現(xiàn), 具體可見: https://github.com/aio-libs, 列出來當(dāng)前已經(jīng)實(shí)現(xiàn)異步操作的大部分庫(kù). 當(dāng)然了, 系統(tǒng)asyncio庫(kù)中也有部分簡(jiǎn)單的異步操作實(shí)現(xiàn).

以后再寫耗時(shí)操作的時(shí)候, 就可以用上協(xié)程了, 比如爬蟲. 爬蟲在發(fā)起請(qǐng)求的時(shí)候, 是需要等待返回的, 這時(shí)候同時(shí)發(fā)起 n 個(gè)請(qǐng)求, 就可以極大的提高爬蟲的效率.

總結(jié)

以上是生活随笔為你收集整理的Python 的协程的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。