如何使用Python協程

小編給大家分享一下如何使用Python協程，希望大家閱讀完這篇文章之后都有所收獲，下面讓我們一起去探討吧！

前言

從語法上來看，協程和生成器類似，都是定義體中包含yield關鍵字的函數。
yield在協程中的用法：

• 在協程中yield通常出現在表達式的右邊，例如：datum = yield,可以產出值，也可以不產出--如果yield關鍵字后面沒有表達式，那么生成器產出None.

• 協程可能從調用方接受數據，調用方是通過send(datum)的方式把數據提供給協程使用，而不是next(...)函數，通常調用方會把值推送給協程。

• 協程可以把控制器讓給中心調度程序，從而激活其他的協程

所以總體上在協程中把yield看做是控制流程的方式。

在Python早期的版本中協程也是通過生成器來實現的，也就是基于生成器的協程（Generator-based Coroutines）。在前一篇介紹生成器的文章末尾舉了一個生產者-消費者的例子，就是基于生成器的協程來實現的。

def producer(c):
 n = 0
 while n < 5:
 n += 1
 print('producer {}'.format(n))
 r = c.send(n)
 print('consumer return {}'.format(r))


def consumer():
 r = ''
 while True:
 n = yield r
 if not n:
 return
 print('consumer {} '.format(n))
 r = 'ok'


if __name__ == '__main__':
 c = consumer()
 next(c) # 啟動consumer
 producer(c)

看了這段代碼，相信很多初學者和我一樣對基于生成器的協程實現其實很難馬上就能夠根據業務寫出自己的協程代碼。Python實現者們也注意到這個問題，因為它太不Pythonic了。而基于生成器的協程也將被廢棄，因此本文將重點介紹asyncio包的使用，以及涉及到的一些相關類概念。

注：我使用的Python環境是3.7。

0x00 何為協程(Coroutine)

協程(Coroutine)是在線程中執行的，可理解為微線程，但協程的切換沒有上下文的消耗，它比線程更加輕量些。一個協程可以隨時中斷自己讓另一個協程開始執行，也可以從中斷處恢復并繼續執行，它們之間的調度是由程序員來控制的（可以看本文開篇處生產者-消費者的代碼）。

定義一個協程

在Python3.5+版本新增了aysnc和await關鍵字，這兩個語法糖讓我們非常方便地定義和使用協程。

在函數定義時用async聲明就定義了一個協程。

import asyncio

# 定義了一個簡單的協程
async def simple_async():
 print('hello')
 await asyncio.sleep(1) # 休眠1秒
 print('python')
 
# 使用asynio中run方法運行一個協程
asyncio.run(simple_async())

# 執行結果為
# hello
# python

在協程中如果要調用另一個協程就使用await。要注意await關鍵字要在async定義的函數中使用，而反過來async函數可以不出現await

# 定義了一個簡單的協程
async def simple_async():
 print('hello')
 
asyncio.run(simple_async())

# 執行結果
# hello

asyncio.run()將運行傳入的協程，負責管理asyncio事件循環。

除了run()方法可直接執行協程外，還可以使用事件循環loop

async def do_something(index):
 print(f'start {time.strftime("%X")}', index)
 await asyncio.sleep(1)
 print(f'finished at {time.strftime("%X")}', index)


def test_do_something():
 # 生成器產生多個協程對象
 task = [do_something(i) for i in range(5)]

 # 獲取一個事件循環對象
 loop = asyncio.get_event_loop()
 # 在事件循環中執行task列表
 loop.run_until_complete(asyncio.wait(task))
 loop.close()

test_do_something()

# 運行結果
# start 00:04:03 3
# start 00:04:03 4
# start 00:04:03 1
# start 00:04:03 2
# start 00:04:03 0
# finished at 00:04:04 3
# finished at 00:04:04 4
# finished at 00:04:04 1
# finished at 00:04:04 2
# finished at 00:04:04 0

可以看出幾乎同時啟動了所有的協程。

其實翻閱源碼可知asyncio.run()的實現也是封裝了loop對象及其調用。而asyncio.run()每次都會創建一個新的事件循環對象用于執行協程。

0x01 Awaitable對象

在Python中可等待(Awaitable)對象有：協程(corountine)、任務(Task)、Future。即這些對象可以使用await關鍵字進行調用

await awaitable_object

1. 協程(Coroutine)

協程由async def聲明定義，一個協程可由另一個協程使用await進行調用

async def nested():
 print('in nested func')
 return 13


async def outer():

 # 要使用await 關鍵字 才會執行一個協程函數返回的協程對象
 print(await nested())

asyncio.run(outer())

# 執行結果
# in nested func
# 13

如果在outer()方法中直接調用nested()而不使用await,將拋出一個RuntimeWarning

async def outer():
 # 直接調用協程函數不會發生執行，只是返回一個 coroutine 對象
 nested()
 
asyncio.run(outer())

運行程序，控制臺將輸出以下信息

RuntimeWarning: coroutine 'nested' was never awaited
  nested()
RuntimeWarning: Enable tracemalloc to get the object allocation traceback

2. 任務(Task)

任務(Task)是可以用來并發地執行協程。可以使用asyncio.create_task()將一個協程對象封裝成任務，該任務將很快被排入調度隊列并執行。

async def nested():
 print('in nested func')
 return 13

async def create_task():
 # create_task 將一個協程對象打包成一個 任務時，該協程就會被自動調度運行
 task = asyncio.create_task(nested())
 # 如果要看到task的執行結果
 # 可以使用await等待協程執行完成，并返回結果
 ret = await task
 print(f'nested return {ret}')

asyncio.run(create_task())

# 運行結果
# in nested func
# nested return 13

注：關于并發下文還會詳細說明。

3. Future

Future是一種特殊的低層級(low-level)對象，它是異步操作的最終結果(eventual result)。
當一個 Future 對象 被等待，這意味著協程將保持等待直到該 Future 對象在其他地方操作完畢。

通常在應用層代碼不會直接創建Future對象。在某些庫和asyncio模塊中的會使用到該對象。

async def used_future_func():
 await function_that_returns_a_future_object()

0x02 并發

1. Task

前面我們知道Task可以并發地執行。  asyncio.create_task()就是一個把協程封裝成Task的方法。

async def do_after(what, delay):
 await asyncio.sleep(delay)
 print(what)

# 利用asyncio.create_task創建并行任務
async def corun():
 task1 = asyncio.create_task(do_after('hello', 1)) # 模擬執行1秒的任務
 task2 = asyncio.create_task(do_after('python', 2)) # 模擬執行2秒的任務

 print(f'started at {time.strftime("%X")}')
 # 等待兩個任務都完成，兩個任務是并行的，所以總時間兩個任務中最大的執行時間
 await task1
 await task2

 print(f'finished at {time.strftime("%X")}')

asyncio.run(corun())

# 運行結果
# started at 23:41:08
# hello
# python
# finished at 23:41:10

task1是一個執行1秒的任務，task2是一個執行2秒的任務，兩個任務并發的執行，總共消耗2秒。

2. gather

除了使用asyncio.create_task()外還可以使用asyncio.gather()，這個方法接收協程參數列表

async def do_after(what, delay):
 await asyncio.sleep(delay)
 print(what)
 
async def gather():
 print(f'started at {time.strftime("%X")}')
 # 使用gather可將多個協程傳入
 await asyncio.gather(
 do_after('hello', 1),
 do_after('python', 2),
 )
 print(f'finished at {time.strftime("%X")}')

asyncio.run(gather())

# 運行結果
# started at 23:47:50
# hello
# python
# finished at 23:47:52

兩個任務消耗的時間為其中消耗時間最長的任務。

看完了這篇文章，相信你對“如何使用Python協程”有了一定的了解，如果想了解更多相關知識，歡迎關注億速云行業資訊頻道，感謝各位的閱讀！