python協程

協程本質上就是一個執行緒，不過它是協作式的非搶占式的程式，面向的是IO操作，python有GIL的限制，不能充分利用多執行緒的實作高并發，行程和執行緒都是通過cpu的調度實作不同任務的有序執行，協程則要在寫代碼的時候確定執行順序，由于協程在一個執行緒中，所以協程不能阻塞，

優缺點：

• 無需執行緒背景關系切換的開銷
• 在一個執行緒中，不需要加鎖
• 無法利用多核資源：協程的本質是單執行緒，需要和行程配合才能運行在多CPU上
• 進行阻塞(Blocking)操作(如IO時)會阻塞掉整個程式

python協程的發展時間較長：

• python2.5 為生成器參考.send()、.throw()、.close()方法
• python3.3 為引入yield from，可以接識訓傳值，可以使用yield from定義協程
• Python3.4 加入了asyncio模塊
• Python3.5 增加async、await關鍵字，在語法層面的提供支持
• python3.7 使用async def + await的方式定義協程
• 此后asyncio模塊更加完善和穩定，對底層的API進行的封裝和擴展
• python將于 3.10版本中移除 以yield from的方式定義協程 （目前版本是3.9.1）

由于asyncio每個版本都會新增功能，對一些舊的底層的API進行封裝，極大地方便的使用者，但正因為此，網上有很多教程使用的介面官方已經不建議直接使用，應該改而使用更加高級的API，所以在這里記錄一下如何使用這些API，

簡單例子

要點：

1. 使用async def的形式定義
2. 在協程中可以使用await關鍵字，注意其后跟的是"可等待物件"(協程, 任務 和 Future)
3. 協程不能直接執行，需要在asyncio.run()中執行，也可以跟在await后面
4. async和await這兩個關鍵字只能在協程中使用

   import asyncio
   
   
   async def foo(name):
   
   	await asyncio.sleep(1)      # 這是一個不會阻塞的sleep，是一個協程
   	print(f"name = {name}")
   
   
   async def main():
   	# 協程本身就是一個可等待物件
   	await foo("lczmx")  # 執行協程
   	print("done")
   
   if __name__ == '__main__':
   	# 使用asyncio.run運行
   	asyncio.run(main())
   

asyncio.run(main, *, debug=False)方法就是對run_until_complete進行了封裝：
loop = events.new_event_loop()
return loop.run_until_complete(main)

關于可等待物件
可等待物件(awaitable)是能在 await 運算式中使用的物件，可以是 協程 或是具有__await__() 方法的物件，

那么協程是如何成為可等待物件的呢？

1. collections.abc.Awaitable類，這是為可等待物件提供的類，可被用于 await 運算式中

   class Awaitable(metaclass=ABCMeta):
   	__slots__ = ()
   
   	@abstractmethod
   	def __await__(self):	# __await__方法必須回傳一個 iterator
   		yield
   
   	@classmethod
   	def __subclasshook__(cls, C):
   		if cls is Awaitable:
   			return _check_methods(C, "__await__")
   		return NotImplemented

2. 用async def復合陳述句創建的函式，它回傳的是一個Coroutine物件，而Coroutine繼承Awaitable，

并發

使用協程進行并發操作
方法一
使用asyncio.create_task(coro)方法，回傳一個Task物件，Task類繼承Future，在python3.7以下版本中使用asyncio.ensure_future(coro_or_future)，

import asyncio


async def foo(char:str, count: int):
    for i in range(count):
        print(f"{char}-{i}")
        await asyncio.sleep(.5)


async def main():
    task1 = asyncio.create_task(foo("A", 2))
    task2 = asyncio.create_task(foo("B", 3))
    task3 = asyncio.create_task(foo("C", 2))

    await task1
    await task2
    await task3


if __name__ == '__main__':
    asyncio.run(main())

執行結果

A-0
B-0
C-0
A-1
B-1
C-1
B-2

方法二
使用asyncio.gather()方法，其內部呼叫的是asyncio.ensure_future()方法

import asyncio


async def foo(char:str, count: int):
    for i in range(count):
        print(f"{char}-{i}")
        await asyncio.sleep(.5)


async def main():

    await asyncio.gather(foo("A", 2), foo("B", 3), foo("C", 2))

if __name__ == '__main__':
    asyncio.run(main())

回呼、回傳值等操作

要完成這些功能需要Task物件，即asyncio.create_task()的回傳值，由于Task繼承Future，實作了除Future.set_result() 和 Future.set_exception()外的全部API，而asyncio.Future模仿的是 concurrent.futures.Future類，所以Task很多方法和 在使用線/行程池時用到的方法類似（有細微差別），

Task的方法，見官方檔案
使用回呼函式和取得回傳值的例子：

import asyncio


def callback(future):
    # 唯一引數是一個Task物件
    # print(type(future))     # <class '_asyncio.Task'>

    print(future)
    # <Task finished name='Task-2' coro=<foo() done, defined at E: ... xxx.py:11> result=123>

    print(future.result())      # 123   # 接識訓傳值
    print(future.get_name())    # foo


async def foo():
    print("running")
    return 123


async def main():
    task = asyncio.create_task(foo(), name="foo")   # name形參3.8及以上版本可用
    task.add_done_callback(callback)                # 添加回呼函式
    await task


if __name__ == '__main__':
    asyncio.run(main())

與執行緒結合

我們知道，協程本身就只有一個執行緒，假如這協程阻塞了，那么整個程式也就阻塞了，為此我們在執行一些必然會產生阻塞的代碼時，可以把代碼放入到其它執行緒/行程中，這樣可以繼續執行協程的其它代碼了，

方法一
coroutine asyncio.to_thread(func, /, *args, **kwargs)
這是python3.9的新方法，3.9以下版本看方法二
在不同的執行緒中異步地運行函式 func，向此函式提供的任何*args 和 **kwargs 會被直接傳給 func，其回傳值是一個協程，所以假如有回呼等操作，使用asyncio.create_task(coro)方法，再呼叫Task物件的方法，

import asyncio
import time


def block_func(name: str):
    time.sleep(2)       # 模擬阻塞時間
    print(f"name = {name}")


async def foo():
    # 一個協程
    print("async foo")
    await asyncio.sleep(1)


async def main():
    await asyncio.gather(
        asyncio.to_thread(block_func, name="lczmx"),
        foo()
    )

if __name__ == '__main__':
    asyncio.run(main())

方法二
awaitable loop.run_in_executor(executor, func, *args)
安排在指定的執行器(線/行程池)中呼叫 func，該方法的回傳值awaitable物件，其實就是一個asyncio.Future物件，這個方法使用起來也比較簡單，不過要注意傳參方式：位置引數可以直接傳入，而關鍵字引數需要使用functools.partial()，

from concurrent.futures import ThreadPoolExecutor
from functools import partial
import asyncio
import time


def block_func(name: str, age: int):
    time.sleep(2)  # 模擬阻塞時間
    print(f"name = {name}, age = {age}")


async def foo():
    # 一個協程
    print("async foo")
    await asyncio.sleep(1)


async def main():
    loop = asyncio.get_running_loop()

    with ThreadPoolExecutor(5) as pool:
        task = loop.run_in_executor(pool, partial(block_func, "lczmx", age=18))
        # task可以添加回呼等操作

    await asyncio.gather(foo(), task)


if __name__ == '__main__':
    asyncio.run(main())

這個兩個方法的關系
asyncio.to_thread()方法實質上就是對loop.run_in_executor()方法進行了封裝：

async def to_thread(func, /, *args, **kwargs):
    loop = events.get_running_loop()
    ctx = contextvars.copy_context()
    func_call = functools.partial(ctx.run, func, *args, **kwargs)
    return await loop.run_in_executor(None, func_call)

假如B執行緒想要把協程c放入到A執行緒的事件回圈中執行，使用syncio.run_coroutine_threadsafe(coro_func(), loop)方法，具體怎么使用見范例

異步背景關系管理器

異步背景關系管理器使用的是async with語法， 是 背景關系管理器 的一種，它能夠在其__aenter__ 和 __aexit__ 方法中暫停執行，使用異步背景關系管理器之前，可以先看一看一般的背景關系管理器，類比過來，

• __aenter__(self)
  在語意上類似于 __enter__()，僅有的區別是它必須回傳一個 可等待物件，
• __aexit__(self, exc_type, exc_value, traceback)
  在語意上類似于 __exit__()，僅有的區別是它必須回傳一個 可等待物件，

一個簡單的例子：

import asyncio


class AContext:
    def __init__(self):
        print("init running")

    async def __aenter__(self):
        print("aenter running")

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("aexit running")


async def main():
    async with AContext() as ac:
        print("AContext", ac)   # AContext None	# __aenter__ 沒有return東西


if __name__ == '__main__':
    print("start")
    asyncio.run(main())

使用標準庫

除了上述方法可以實作異步背景關系管理器外，還可以使用contextlib.asynccontextmanager裝飾器+yield實作，yield前面的代碼對應__aenter__，其后的代碼對應__aexit__，

import contextlib
import asyncio


# 加上裝飾器
@contextlib.asynccontextmanager
async def foo():
    try:
        # 進行初始化
        yield "回傳你要操作的物件"
    finally:
        # 處理釋放資源等操作
        pass


async def main():
    async with foo() as f:
        print(f)


if __name__ == "__main__":
    asyncio.run(main())

異步迭代器

異步迭代器使用的時async for語法，和常規的for運算式一樣， async for也有一個可選的else 分句，異步迭代器有以下幾個要求：

1. 必須實作__aiter__方法，該方法回傳一個異步迭代器物件，
2. 異步迭代器物件必須實作__anext__方法，該方法回傳一個awaitable型別的值，
3. 為了停止迭代，__anext__必須拋出一個StopAsyncIteration例外，
   跟迭代器協議非常類似，
   例子：

class AsyncIteratorWrapper:
    def __init__(self, obj):
        self._it = iter(obj)

    def __aiter__(self):
        return self

    async def __anext__(self):
        try:
            value = https://www.cnblogs.com/lczmx/archive/2021/02/02/next(self._it)
        except StopIteration:
            raise StopAsyncIteration
        return value

async for letter in AsyncIteratorWrapper("abc"):
    print(letter)

綜合例子

非阻塞爬蟲

from contextlib import asynccontextmanager
from concurrent.futures import ThreadPoolExecutor
from functools import partial
import requests
import asyncio
import time


class ASyncRequests:
    """非阻塞的requests"""

    def __init__(self, count):
        """
        :param count: 執行緒池大小
        """
        assert isinstance(count, int)
        self.pool = ThreadPoolExecutor(count)  # 創建執行緒池
        self.loop = asyncio.get_running_loop()  # 獲得事件回圈
        if not self.loop:
            raise RuntimeError("event loop為None")

        # self.all_future 用于裝run_in_executor回傳的future物件
        #                  python3.9版本用to_thread

        self.all_future = []

    def get(self, url, callback, **kwargs):
        self.requests(url, "get", callback, **kwargs)

    def post(self, url, callback, **kwargs):
        self.requests(url, "post", callback, **kwargs)

    def requests(self, url, method, callback, **kwargs):
        """
        寫了兩種方法，第一種適合版本python，另一種適合最新版的python
        """

        # python3.5及以上版本可用
        # run_in_executor:
        #   把阻塞的函式放入執行緒池中執行
        #   回傳一個asyncio.Future物件
        #   傳值時，利用 偏函式 傳關鍵字引數

        result = self.loop.run_in_executor(self.pool, partial(requests.request,
                                                              url=url, method=method, **kwargs))
        result.add_done_callback(callback)  # 回呼
        self.all_future.append(result)

        # python3.9及以上版本可用
        # to_thread:
        #   使用其它的執行緒，執行阻塞函式
        #   回傳一個協程
        #   除第一個引數外的所有引數都會給func

        # coro = asyncio.to_thread(requests.request, url=url, method=method, **kwargs)
        # task = asyncio.create_task(coro)
        # task.add_done_callback(callback)            # 回呼
        # self.all_future.append(task)


def process_get(future):
    """回呼函式，處理回傳值"""

    print(type(future))  # <class '_asyncio.Future'>

    # future.result()的回傳值就是Response物件
    print(future.result().status_code)  # 200


@asynccontextmanager
async def request_session(count=5):
    try:
        async_req = ASyncRequests(count)
        yield async_req
    finally:
        await asyncio.gather(*async_req.all_future)


async def main():
    async with request_session(10) as session:
        session.get("https://www.baidu.com/", callback=process_get)
        session.get("https://www.sogo.com/", callback=process_get)
        session.get("https://cn.bing.com/", callback=process_get)
        session.get("https://www.cnblogs.com/lczmx", callback=process_get)


def block_req():
    requests.get("https://www.baidu.com/")
    requests.get("https://www.sogo.com/")
    requests.get("https://cn.bing.com/")
    requests.get("https://www.cnblogs.com/lczmx")


if __name__ == '__main__':
    start = time.time()
    asyncio.run(main())
    end = time.time()
    print("使用協程加執行緒：總用時: %f sec" % (end - start))

    block_req()
    print("直接訪問：總用時: %f sec" % (time.time() - end))
    # 訪問數越多，協程的優勢就越大
    # 訪問同一個網站過多可能會報錯 Connection aborted.', ConnectionResetError(10054, '遠程主機強迫關閉了一個現有的連接，

推薦閱讀：

Python黑魔法 --- 異步IO（ asyncio） 協程
asyncio之異步背景關系管理器
Python Async/Await入門指南
協程與任務
事件回圈

我的github
我的博客
我的筆記

標籤：其他

上一篇：二叉樹、平衡二叉樹、紅黑樹、B樹、B+樹與B*樹

下一篇：免費開源的代碼審計工具 Gosec 入門使用