Python?Asyncio?庫之同步原語常用函數(shù)詳解

更新時(shí)間：2023年03月01日 14:08:38 作者：so1n

這篇文章主要為大家介紹了Python?Asyncio?庫之同步原語常用函數(shù)詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

前記

Asyncio的同步原語可以簡(jiǎn)化我們編寫資源競(jìng)爭(zhēng)的代碼和規(guī)避資源競(jìng)爭(zhēng)導(dǎo)致的Bug的出現(xiàn)。但是由于協(xié)程的特性，在大部分業(yè)務(wù)代碼中并不需要去考慮資源競(jìng)爭(zhēng)的出現(xiàn)，導(dǎo)致Asyncio同步原語被使用的頻率比較低，但是如果想基于Asyncio編寫框架則需要學(xué)習(xí)同步原語的使用。

0.基礎(chǔ)

同步原語都是適用于某些條件下對(duì)某個(gè)資源的爭(zhēng)奪，在代碼中大部分的資源都是屬于一個(gè)代碼塊，而Python對(duì)于代碼塊的管理的最佳實(shí)踐是使用with語法，with語法實(shí)際上是調(diào)用了一個(gè)類中的__enter__和__exit__方法，比如下面的代碼：

class Demo(object):
    def __enter__(self):
        return 
    def __exit__(self, exc_type, exc_val, exc_tb):
        return 
with Demo():
    pass

代碼中的Demo類實(shí)現(xiàn)了__enter__和__exit__方法后，就可以被with語法調(diào)用，其中__enter__方法是進(jìn)入代碼塊執(zhí)行的邏輯，__enxi__方法是用于退出代碼塊(包括異常退出)的邏輯。這兩個(gè)方法符合同步原語中對(duì)資源的爭(zhēng)奪和釋放，但是__enter__和__exit__兩個(gè)方法都是不支持await調(diào)用的，為了解決這個(gè)問題，Python引入了async with語法。

async with語法和with語法類似，我們只要編寫一個(gè)擁有__aenter__和__aexit__方法的類，那么這個(gè)類就支持asyncio with語法了，如下:

import asyncio
class Demo(object):
    async def __aenter__(self):
        return
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        return
async def main():
    async with Demo():
        pass
asyncio.run(main())

其中，類中的__aenter__方法是進(jìn)入代碼塊時(shí)執(zhí)行的方法，__aexit__是退出代碼塊時(shí)執(zhí)行的方法。

有了async with語法的加持，asyncio的同步原語使用起來會(huì)比較方便，所以asyncio中對(duì)資源爭(zhēng)奪的同步原語都會(huì)繼承于_ContextManagerMixin類：

class _ContextManagerMixin:
    async def __aenter__(self):
        await self.acquire()
        # We have no use for the "as ..."  clause in the with
        # statement for locks.
        return None
    async def __aexit__(self, exc_type, exc, tb):
        self.release()

并實(shí)現(xiàn)了acquire和release方法，供__aenter__和__aexit__方法調(diào)用，同時(shí)我們?cè)谑褂猛皆Z的時(shí)候盡量用到async with語法防止忘記釋放資源的占用。

1.Lock

由于協(xié)程的特性，在編寫協(xié)程代碼時(shí)基本上可以不考慮到鎖的情況，但在一些情況下我們還是需要用到鎖，并通過鎖來維護(hù)并發(fā)時(shí)的數(shù)據(jù)安全性，如下例子：

import asyncio
share_data = {}
async def sub(i):
    # 賦上相同的key和value
    share_data[i] = i
    await asyncio.sleep(0)
    print(i, share_data[i] == i)
async def sub_add(i):
    # 賦上的value值是原來的+1
    share_data[i] = i + 1
    await asyncio.sleep(0)
    print(i, share_data[i] == i + 1)
async def main():
    # 創(chuàng)建并發(fā)任務(wù)
    task_list = []
    for i in range(10):
        task_list.append(sub(i))
        task_list.append(sub_add(i))
    # 并發(fā)執(zhí)行
    await asyncio.gather(*task_list)
if __name__ == "__main__":
    asyncio.run(main())

在這個(gè)例子中程序會(huì)并發(fā)的執(zhí)行sub和sub_add函數(shù)，他們是由不同的asyncio.Task驅(qū)動(dòng)的，這意味著會(huì)出現(xiàn)這樣一個(gè)場(chǎng)景。當(dāng)負(fù)責(zé)執(zhí)行sub(1)函數(shù)的asyncio.Task在執(zhí)行完share_data[i]=i后就執(zhí)行await asyncio.sleep(0)從而主動(dòng)讓出控制權(quán)并交還給事件循環(huán)，等待事件循環(huán)的下一次調(diào)度。不過事件循環(huán)不會(huì)空下來，而是馬上安排下一個(gè)asyncio.Task執(zhí)行，此時(shí)會(huì)先執(zhí)行到sub_add(1)函數(shù)的share_data[i] = i + 1，并同樣的在執(zhí)行到await asyncio.sleep(0)的時(shí)候把控制權(quán)交會(huì)給事件循環(huán)。這時(shí)候控制權(quán)會(huì)由事件循環(huán)轉(zhuǎn)移給原先執(zhí)行sub(1)函數(shù)的asyncio.Task，獲取到控制權(quán)l(xiāng)后sub(1)函數(shù)的邏輯會(huì)繼續(xù)走，但由于share_data[i]的數(shù)據(jù)已經(jīng)被share_data[i] = i + 1修改了，導(dǎo)致最后執(zhí)行print時(shí)，share_data[i]的數(shù)據(jù)已經(jīng)變?yōu)榕K數(shù)據(jù)，而不是原本想要的數(shù)據(jù)了。

為了解決這個(gè)問題，我們可以使用asyncio.Lock來解決資源的沖突，如下:

import asyncio
share_data = {}
# 存放對(duì)應(yīng)資源的鎖
lock_dict = {}
async def sub(i):
    async with lock_dict[i]:  # <-- 通過async with語句來控制鎖的粒度
        share_data[i] = i
        await asyncio.sleep(0)
        print(i, share_data[i] == i)
async def sub_add(i):
    async with lock_dict[i]:
        share_data[i] = i + 1
        await asyncio.sleep(0)
        print(i, share_data[i] == i + 1)
async def main():
    task_list = []
    for i in range(10):
        lock_dict[i] = asyncio.Lock()
        task_list.append(sub(i))
        task_list.append(sub_add(i))
    await asyncio.gather(*task_list)
if __name__ == "__main__":
    asyncio.run(main())

從例子可以看到asyncio.Lock的使用方法跟多線程的Lock差不多，通過async with語法來獲取和釋放鎖，它的原理也很簡(jiǎn)單，主要做了如下幾件事：

1.確保某一協(xié)程獲取鎖后的執(zhí)行期間，別的協(xié)程在獲取鎖時(shí)需要一直等待，直到執(zhí)行完成并釋放鎖。
2.當(dāng)有協(xié)程持有鎖的時(shí)候，其他協(xié)程必須等待，直到持有鎖的協(xié)程釋放了鎖。
2.確保所有協(xié)程能夠按照獲取的順序獲取到鎖。

這意味著需要有一個(gè)數(shù)據(jù)結(jié)構(gòu)來維護(hù)當(dāng)前持有鎖的協(xié)程的和下一個(gè)獲取鎖協(xié)程的關(guān)系，同時(shí)也需要一個(gè)隊(duì)列來維護(hù)多個(gè)獲取鎖的協(xié)程的喚醒順序。

asyncio.Lock跟其它asyncio功能的用法一樣，使用asyncio.Future來同步協(xié)程之間鎖的狀態(tài)，使用deque維護(hù)協(xié)程間的喚醒順序，源碼如下:

class Lockl(_ContextManagerMixin, mixins._LoopBoundMixin):
    def __init__(self):
        self._waiters = None
        self._locked = False
    def locked(self):
        return self._locked
    async def acquire(self):
        if (not self._locked and (self._waiters is None or all(w.cancelled() for w in self._waiters))):
            # 目前沒有其他協(xié)程持有鎖，當(dāng)前協(xié)程可以運(yùn)行
            self._locked = True
            return True
        if self._waiters is None:
            self._waiters = collections.deque()
        # 創(chuàng)建屬于自己的容器，并推送到`_waiters`這個(gè)雙端隊(duì)列中
        fut = self._get_loop().create_future()
        self._waiters.append(fut)
        try:
            try:
                await fut
            finally:
                # 如果執(zhí)行完畢，需要把自己移除，防止被`wake_up_first`調(diào)用
                self._waiters.remove(fut)
        except exceptions.CancelledError:
            # 如果是等待的過程中被取消了，需要喚醒下一個(gè)調(diào)用`acquire`
            if not self._locked:
                self._wake_up_first()
            raise
        # 持有鎖
        self._locked = True
        return True
    def release(self):
        if self._locked:
            # 釋放鎖
            self._locked = False
            self._wake_up_first()
        else:
            raise RuntimeError('Lock is not acquired.')
    def _wake_up_first(self):
        if not self._waiters:
            return
        # 獲取還處于鎖狀態(tài)協(xié)程對(duì)應(yīng)的容器
        try:
            # 獲取下一個(gè)等待獲取鎖的waiter
            fut = next(iter(self._waiters))
        except StopIteration:
            return
        # 設(shè)置容器為True,這樣對(duì)應(yīng)協(xié)程就可以繼續(xù)運(yùn)行了。
        if not fut.done():
            fut.set_result(True)

通過源碼可以知道，鎖主要提供了獲取和釋放的功能，對(duì)于獲取鎖需要區(qū)分兩種情況：

1:當(dāng)有協(xié)程想要獲取鎖時(shí)會(huì)先判斷鎖是否被持有，如果當(dāng)前鎖沒有被持有就直接返回，使協(xié)程能夠正常運(yùn)行。
2:如果協(xié)程獲取鎖時(shí)，鎖發(fā)現(xiàn)自己已經(jīng)被其他協(xié)程持有則創(chuàng)建一個(gè)屬于當(dāng)前協(xié)程的asyncio.Future，用來同步狀態(tài)，并添加到deque中。

而對(duì)于釋放鎖就比較簡(jiǎn)單，只要獲取deque中的第一個(gè)asyncio.Future，并通過fut.set_result(True)進(jìn)行標(biāo)記，使asyncio.Future從peding狀態(tài)變?yōu)?code>done狀態(tài)，這樣一來，持有該asyncio.Future的協(xié)程就能繼續(xù)運(yùn)行，從而持有鎖。

不過需要注意源碼中acquire方法中對(duì)CancelledError異常進(jìn)行捕獲，再喚醒下一個(gè)鎖，這是為了解決acquire方法執(zhí)行異常導(dǎo)致鎖一直被卡住的場(chǎng)景，通常情況下這能解決大部分的問題，但是如果遇到錯(cuò)誤的封裝時(shí)，我們需要親自處理異常，并執(zhí)行鎖的喚醒。比如在通過繼承asyncio.Lock編寫一個(gè)超時(shí)鎖時(shí)，最簡(jiǎn)單的實(shí)現(xiàn)代碼如下:

import asyncio
class TimeoutLock(asyncio.Lock):
    def __init__(self, timeout, *, loop=None):
        self.timeout = timeout
        super().__init__(loop=loop)
    async def acquire(self) -> bool:
        return await asyncio.wait_for(super().acquire(), self.timeout)

這份代碼非常簡(jiǎn)單，他只需要在__init__方法傳入timeout參數(shù)，并在acuiqre方法中通過wait_for來實(shí)現(xiàn)鎖超時(shí)即可，現(xiàn)在假設(shè)wait_for方法是一個(gè)無法傳遞協(xié)程cancel的方法，且編寫的acquire沒有進(jìn)行捕獲異常再釋放鎖的操作，當(dāng)異常發(fā)生的時(shí)候會(huì)導(dǎo)致鎖一直被卡住。為了解決這個(gè)問題，只需要對(duì)TimeoutLock的acquire方法添加異常捕獲，并在捕獲到異常時(shí)釋放鎖即可，代碼如下:

class TimeoutLock(asyncio.Lock):
    def __init__(self, timeout, *, loop=None):
        self.timeout = timeout
        super().__init__(loop=loop)
    async def acquire(self) -> bool:
        try:
            return await asyncio.wait_for(super().acquire(), self.timeout)
        except Exception:
            self._wake_up_first()
            raise

2.Event

asyncio.Event也是一個(gè)簡(jiǎn)單的同步原語，但它跟asyncio.Lock不一樣，asyncio.Lock是確保每個(gè)資源只能被一個(gè)協(xié)程操作，而asyncio.Event是確保某個(gè)資源何時(shí)可以被協(xié)程操作，可以認(rèn)為asyncio.Lock鎖的是資源，asyncio.Event鎖的是協(xié)程，所以asyncio.Event并不需要acquire來鎖資源，release釋放資源，所以也用不到async with語法。

asyncio.Event的簡(jiǎn)單使用示例如下：

import asyncio
async def sub(event: asyncio.Event) -> None:
    await event.wait()
    print("I'm Done")
async def main() -> None:
    event = asyncio.Event()
    for _ in range(10):
        asyncio.create_task(sub(event))
    await asyncio.sleep(1)
    event.set()
asyncio.run(main())

在這個(gè)例子中會(huì)先創(chuàng)建10個(gè)asyncio.Task來執(zhí)行sub函數(shù)，但是所有sub函數(shù)都會(huì)在event.wait處等待，直到main函數(shù)中調(diào)用event.set后，所有的sub函數(shù)的event.wait會(huì)放行，使sub函數(shù)能繼續(xù)執(zhí)行。

可以看到asyncio.Event功能比較簡(jiǎn)單，它的源碼實(shí)現(xiàn)也很簡(jiǎn)單，源碼如下：

class Event(mixins._LoopBoundMixin):
    def __init__(self):
        self._waiters = collections.deque()
        self._value = False
    def is_set(self):
        return self._value
    def set(self):
        if not self._value:
            # 確保每次只能set一次
            self._value = True
            # 設(shè)置每個(gè)協(xié)程存放的容器為True，這樣對(duì)應(yīng)的協(xié)程就可以運(yùn)行了
            for fut in self._waiters:
                if not fut.done():
                    fut.set_result(True)
    def clear(self):
        # 清理上一次的set
        self._value = False
    async def wait(self):
        if self._value:
            # 如果設(shè)置了，就不需要等待了
            return True
        # 否則需要?jiǎng)?chuàng)建一個(gè)容器，并需要等待容器完成
        fut = self._get_loop().create_future()
        self._waiters.append(fut)
        try:
            await fut
            return True
        finally:
            self._waiters.remove(fut)

通過源碼可以看到wait方法主要是創(chuàng)建了一個(gè)asyncio.Future，并把它加入到deque隊(duì)列后就一直等待著，而set方法被調(diào)用時(shí)會(huì)遍歷整個(gè)deque隊(duì)列，并把處于peding狀態(tài)的asyncio.Future設(shè)置為done，這時(shí)其他在調(diào)用event.wait方法的協(xié)程就會(huì)得到放行。

通過源碼也可以看出，asyncio.Event并沒有繼承于_ContextManagerMixin，這是因?yàn)樗i的是協(xié)程，而不是資源。

asyncio.Event的使用頻率比asyncio.Lock多許多，不過通常都會(huì)讓asyncio.Event和其他數(shù)據(jù)結(jié)構(gòu)進(jìn)行封裝再使用，比如實(shí)現(xiàn)一個(gè)服務(wù)器的優(yōu)雅關(guān)閉功能，這個(gè)功能會(huì)確保服務(wù)器在等待n秒后或者所有連接都關(guān)閉后才關(guān)閉服務(wù)器，這個(gè)功能就可以使用set與asyncio.Event結(jié)合，如下:

import asyncio
class SetEvent(asyncio.Event):
    def __init__(self, *, loop=None):
        self._set = set()
        super().__init__(loop=loop)
    def add(self, value):
        self._set.add(value)
        self.clear()
    def remove(self, value):
        self._set.remove(value)
        if not self._set:
            self.set()

這個(gè)SetEvent結(jié)合了set和SetEvent的功能，當(dāng)set有數(shù)據(jù)的時(shí)候，會(huì)通過clear方法使SetEvent變?yōu)榈却隣顟B(tài)，而set沒數(shù)據(jù)的時(shí)候，會(huì)通過set方法使SetEvent變?yōu)闊o需等待的狀態(tài)，所有調(diào)用wait的協(xié)程都可以放行，通過這種結(jié)合，SetEvent擁有了等待資源為空的功能。接下來就可以用于服務(wù)器的優(yōu)雅退出功能：

async def mock_conn_io() -> None:
    await asyncio.sleep(1)
def conn_handle(set_event: SetEvent):
    task: asyncio.Task = asyncio.create_task(mock_conn_io())
    set_event.add(task)
    task.add_done_callback(lambda t: set_event.remove(t))
async def main():
    set_event: SetEvent = SetEvent()
    for _ in range(10):
        conn_handle(set_event)
	# 假設(shè)這里收到了退出信號(hào)
    await asyncio.wait(set_event.wait(), timeout=9)
asyncio.run(main())

在這個(gè)演示功能中，mock_conn_io用于模擬服務(wù)器的連接正在處理中，而conn_handle用于創(chuàng)建服務(wù)器連接，main則是先創(chuàng)建10個(gè)連接，并模擬在收到退出信號(hào)后等待資源為空或者超時(shí)才退出服務(wù)。

這只是簡(jiǎn)單的演示，實(shí)際上的優(yōu)雅關(guān)閉功能要考慮的東西不僅僅是這些。

4.Condition

condition只做簡(jiǎn)單介紹

asyncio.Condition是同步原語中使用最少的一種，因?yàn)樗褂们闆r很奇怪，而且大部分場(chǎng)景可以被其他寫法代替，比如下面這個(gè)例子:

import asyncio
async def task(condition, work_list):
    await asyncio.sleep(1)
    work_list.append(33)
    print('Task sending notification...')
    async with condition:
        condition.notify()
async def main():
    condition = asyncio.Condition()
    work_list = list()
    print('Main waiting for data...')
    async with condition:
        _ = asyncio.create_task(task(condition, work_list))
        await condition.wait()
    print(f'Got data: {work_list}')
asyncio.run(main())
# &gt;&gt;&gt; Main waiting for data...
# &gt;&gt;&gt; Task sending notification...
# &gt;&gt;&gt; Got data: [33]

在這個(gè)例子中可以看到，notify和wait方法只能在async with condition中可以使用，如果沒有在async with condition中使用則會(huì)報(bào)錯(cuò)，同時(shí)這個(gè)示例代碼有點(diǎn)復(fù)雜，沒辦法一看就知道執(zhí)行邏輯是什么，其實(shí)這個(gè)邏輯可以轉(zhuǎn)變成一個(gè)更簡(jiǎn)單的寫法：

import asyncio
async def task(work_list):
    await asyncio.sleep(1)
    work_list.append(33)
    print('Task sending notification...')
    return
async def main():
    work_list = list()
    print('Main waiting for data...')
    _task = asyncio.create_task(task(work_list))
    await _task
    print(f'Got data: {work_list}')
asyncio.run(main())
# &gt;&gt;&gt; Main waiting for data...
# &gt;&gt;&gt; Task sending notification...
# &gt;&gt;&gt; Got data: [33]

通過這個(gè)代碼可以看到這個(gè)寫法更簡(jiǎn)單一點(diǎn)，而且更有邏輯性，而condition的寫法卻更有點(diǎn)Go協(xié)程寫法/或者回調(diào)函數(shù)寫法的感覺。所以建議在認(rèn)為自己的代碼可能會(huì)用到asyncio.Conditon時(shí)需要先考慮到是否需要asyncio.Codition？是否有別的方案代替，如果沒有才考慮去使用asyncio.Conditonk。

5.Semaphore

asyncio.Semaphore--信號(hào)量是同步原語中被使用最頻繁的，大多數(shù)都是用在限流場(chǎng)景中，比如用在爬蟲中和客戶端網(wǎng)關(guān)中限制請(qǐng)求頻率。

asyncio.Semaphore可以認(rèn)為是一個(gè)延緩觸發(fā)的asyncio.Lock，asyncio.Semaphore內(nèi)部會(huì)維護(hù)一個(gè)計(jì)數(shù)器，無論何時(shí)進(jìn)行獲取或釋放，它都會(huì)遞增或者遞減(但不會(huì)超過邊界值)，當(dāng)計(jì)數(shù)器歸零時(shí)，就會(huì)進(jìn)入到鎖的邏輯，但是這個(gè)鎖邏輯會(huì)在計(jì)數(shù)器大于0的時(shí)候釋放j，它的用法如下：`

import asyncio
async def main():
    semaphore = asyncio.Semaphore(10):
    async with semaphore:
        pass
asyncio.run(main())

示例中代碼通過async with來指明一個(gè)代碼塊（代碼用pass代替），這個(gè)代碼塊是被asyncio.Semaphore管理的，每次協(xié)程在進(jìn)入代碼塊時(shí)，asyncio.Semaphore的內(nèi)部計(jì)數(shù)器就會(huì)遞減一，而離開代碼塊則asyncio.Semaphore的內(nèi)部計(jì)數(shù)器會(huì)遞增一。

當(dāng)有一個(gè)協(xié)程進(jìn)入代碼塊時(shí)asyncio.Semaphore發(fā)現(xiàn)計(jì)數(shù)器已經(jīng)為0了，則會(huì)使當(dāng)前協(xié)程進(jìn)入等待狀態(tài)，直到某個(gè)協(xié)程離開這個(gè)代碼塊時(shí)，計(jì)數(shù)器會(huì)遞增一，并喚醒等待的協(xié)程，使其能夠進(jìn)入代碼塊中繼續(xù)執(zhí)行。

asyncio.Semaphore的源碼如下，需要注意的是由于asyncio.Semaphore是一個(gè)延緩的asyncio.Lock，所以當(dāng)調(diào)用一次release后可能會(huì)導(dǎo)致被喚醒的協(xié)程和剛進(jìn)入代碼塊的協(xié)程起沖突，所以在acquire方法中要通過一個(gè)while循環(huán)來解決這個(gè)問題：`

class Semaphore(_ContextManagerMixin, mixins._LoopBoundMixin):
    def __init__(self, value=1):
        if value < 0:
            raise ValueError("Semaphore initial value must be >= 0")
        self._value = value
        self._waiters = collections.deque()
        self._wakeup_scheduled = False
    def _wake_up_next(self):
        while self._waiters:
            # 按照放置順序依次彈出容器 
            waiter = self._waiters.popleft()
            if not waiter.done():
                # 設(shè)置容器狀態(tài)，使對(duì)應(yīng)的協(xié)程可以繼續(xù)執(zhí)行
                waiter.set_result(None)
                # 設(shè)置標(biāo)記 
                self._wakeup_scheduled = True
                return
    def locked(self):
        return self._value == 0
    async def acquire(self):
        # 如果`self._wakeup_scheduled`為True或者value小于0
        while self._wakeup_scheduled or self._value <= 0:
            # 創(chuàng)建容器并等待執(zhí)行完成
            fut = self._get_loop().create_future()
            self._waiters.append(fut)
            try:
                await fut
                self._wakeup_scheduled = False
            except exceptions.CancelledError:
                # 如果被取消了，也要喚醒下一個(gè)協(xié)程
                self._wake_up_next()
                raise
        self._value -= 1
        return True
    def release(self):
        # 釋放資源占用，喚醒下一個(gè)協(xié)程。
        self._value += 1
        self._wake_up_next()

針對(duì)asyncio.Semaphore進(jìn)行修改可以實(shí)現(xiàn)很多功能，比如基于信號(hào)量可以實(shí)現(xiàn)一個(gè)簡(jiǎn)單的協(xié)程池，這個(gè)協(xié)程池可以限制創(chuàng)建協(xié)程的量，當(dāng)協(xié)程池滿的時(shí)候就無法繼續(xù)創(chuàng)建協(xié)程，只有協(xié)程中的協(xié)程執(zhí)行完畢后才能繼續(xù)創(chuàng)建(當(dāng)然無法控制在協(xié)程中創(chuàng)建新的協(xié)程)，代碼如下：

import asyncio
import time
from typing import Coroutine
class Pool(object):
    def __init__(self, max_concurrency: int):
        self._semaphore: asyncio.Semaphore = asyncio.Semaphore(max_concurrency)
    async def create_task(self, coro: Coroutine) -> asyncio.Task:
        await  self._semaphore.acquire()
        task: asyncio.Task = asyncio.create_task(coro)
        task.add_done_callback(lambda t: self._semaphore.release())
        return task
async def demo(cnt: int) -> None:
    print(f"{int(time.time())} create {cnt} task...")
    await  asyncio.sleep(cnt)
async def main() -> None:
    pool: Pool = Pool(3)
    for i in range(10):
        await pool.create_task(demo(i))
asyncio.run(main())
# >>> 1677517996 create 0 task...
# >>> 1677517996 create 1 task...
# >>> 1677517996 create 2 task...
# >>> 1677517996 create 3 task...
# >>> 1677517997 create 4 task...
# >>> 1677517998 create 5 task...
# >>> 1677517999 create 6 task...
# >>> 1677518001 create 7 task...
# >>> 1677518003 create 8 task...
# >>> 1677518005 create 9 task...

以上就是Python Asyncio 庫之同步原語常用函數(shù)詳解的詳細(xì)內(nèi)容，更多關(guān)于Python Asyncio 庫同步原語的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: