真实的国产乱ⅩXXX66竹夫人,五月香六月婷婷激情综合,亚洲日本VA一区二区三区,亚洲精品一区二区三区麻豆

成都創(chuàng)新互聯(lián)網(wǎng)站制作重慶分公司

從頭造輪子：python3 asyncio之 sleep （4）

前言

書接上文：，本文造第四個輪子，也是asyncio包里面非常常用，并且非常復(fù)雜的一個函數(shù)sleep

彌勒網(wǎng)站制作公司哪家好，找創(chuàng)新互聯(lián)建站！從網(wǎng)頁設(shè)計、網(wǎng)站建設(shè)、微信開發(fā)、APP開發(fā)、成都響應(yīng)式網(wǎng)站建設(shè)等網(wǎng)站項目制作，到程序開發(fā)，運營維護。創(chuàng)新互聯(lián)建站2013年開創(chuàng)至今到現(xiàn)在10年的時間，我們擁有了豐富的建站經(jīng)驗和運維經(jīng)驗，來保證我們的工作的順利進行。專注于網(wǎng)站建設(shè)就選創(chuàng)新互聯(lián)建站。

一、知識準(zhǔn)備

● time.sleep直接讓當(dāng)前線程睡覺，但是這種方式顯然是不能接受的，如果當(dāng)前線程睡覺，那我們所有的協(xié)程任務(wù)都會被卡主，并發(fā)也就無從談起了
● 理解socket.socketpair()創(chuàng)建的套接字對象
● 理解selectors的應(yīng)用
● 理解最小堆以及heapq的應(yīng)用
● 理解對象比較
● 這一小結(jié)的基礎(chǔ)知識很多，希望大家優(yōu)先了解上述的知識再開始閱讀，否則很容易不知所云


二、環(huán)境準(zhǔn)備

組件 版本
python 3.7.7

三、sleep的實現(xiàn)

先來看下官方sleep的使用方法：

|># more main.py
import asyncio

async def hello():
    print('enter hello ...')
    await asyncio.sleep(5)
    print('hello sleep end...')
    return 'return hello...'

async def world():
    print('enter world ...')
    await asyncio.sleep(3)
    print('world sleep end...')
    return 'return world...'

async def helloworld():
    print('enter helloworld')
    ret = await asyncio.gather(hello(), world())
    print('exit helloworld')
    return ret


if __name__ == "__main__":
    ret = asyncio.run(helloworld())
    print(ret)
    
|># time python3 main.py
enter helloworld
enter hello ...
enter world ...
world sleep end...
hello sleep end...
exit helloworld
['return hello...', 'return world...']

real	0m5.256s
user	0m0.077s
sys	0m0.020s

來看下造的輪子的使用方式：

? more main.py
async def hello():
    print('enter hello ...')
    await wilsonasyncio.sleep(5)
    print('hello sleep end...')
    return 'return hello...'

async def world():
    print('enter world ...')
    await wilsonasyncio.sleep(3)
    print('world sleep end...')
    return 'return world...'

async def helloworld():
    print('enter helloworld')
    ret = await wilsonasyncio.gather(hello(), world())
    print('exit helloworld')
    return ret

    
if __name__ == "__main__":
    ret = wilsonasyncio.run(helloworld())
    print(ret)

    
? time python3 main.py
enter helloworld
enter hello ...
enter world ...
world sleep end...
hello sleep end...
exit helloworld
['return hello...', 'return world...']
python3 main.py  0.06s user 0.04s system 1% cpu 5.406 total

都是用了5s左右，自己造的輪子也很好的運行了，下面我們來看下輪子的代碼

四、代碼解析

輪子代碼

1）代碼組成

|># tree
.
├── eventloops.py 
├── futures.py
├── main.py
├── tasks.py
├── wilsonasyncio.py
文件 作用
eventloops.py 事件循環(huán)
futures.py futures對象
tasks.py tasks對象
wilsonasyncio.py 可調(diào)用方法集合
main.py 入口

2）代碼概覽：

eventloops.py

類/函數(shù) 方法 對象 作用 描述
Eventloop 事件循環(huán)，一個線程只有運行一個
__init__ 初始化兩個重要對象 self._ready 與 self._stopping
self._ready 所有的待執(zhí)行任務(wù)都是從這個隊列取出來，非常重要
self._scheduled 待調(diào)度的任務(wù)隊列，這個隊列的任務(wù)會在合適的時機進入self._ready中 新增
self._stopping 事件循環(huán)完成的標(biāo)志
current_time 從線程啟動到現(xiàn)在經(jīng)歷的秒數(shù) 新增
call_later 調(diào)用該方法會經(jīng)歷一個延時delay之后，再將任務(wù)添加到待執(zhí)行隊列 新增
call_at 調(diào)用該方法會在指定的時間將任務(wù)添加到待執(zhí)行隊列 新增
call_soon 調(diào)用該方法會立即將任務(wù)添加到待執(zhí)行隊列
run_once 被run_forever調(diào)用，從self._ready隊列里面取出任務(wù)執(zhí)行 重新改造，添加了大量邏輯
run_forever 死循環(huán)，若self._stopping則退出循環(huán)
run_until_complete 非常重要的函數(shù)，任務(wù)的起點和終點（后面詳細(xì)介紹）
create_task 將傳入的函數(shù)封裝成task對象，這個操作會將task.__step添加到__ready隊列
create_future 返回Future對象 新增
Handle 所有的任務(wù)進入待執(zhí)行隊列（Eventloop.call_soon）之前都會封裝成Handle對象
__init__ 初始化兩個重要對象 self._callback 與 self._args
self._callback 待執(zhí)行函數(shù)主體
self._args 待執(zhí)行函數(shù)參數(shù)
_run 待執(zhí)行函數(shù)執(zhí)行
TimerHandle 帶時間戳的對象 新增
__init__ 初始化重要對象 self._when 新增
self._when 調(diào)度的時間 新增
__lt__``__gt__``__eq__ 一系列魔術(shù)方法，實現(xiàn)對象比較 新增
get_event_loop 獲取當(dāng)前線程的事件循環(huán)
fake_socket 創(chuàng)建一對套接字對象，并且將一種一條套接字注冊到多路復(fù)用對象sel，返回sel 新增
_complete_eventloop 將事件循環(huán)的_stopping標(biāo)志置位True
run 入口函數(shù)
gather 可以同時執(zhí)行多個任務(wù)的入口函數(shù)
_GatheringFuture 將每一個任務(wù)組成列表，封裝成一個新的類
sleep 入口函數(shù) 新增

tasks.py

類/函數(shù) 方法 對象 作用 描述
Task 繼承自Future，主要用于整個協(xié)程運行的周期
__init__ 初始化對象 self._coro ，并且call_soon將self.__step加入self._ready隊列
self._coro 用戶定義的函數(shù)主體
__step Task類的核心函數(shù)
__wakeup 喚醒任務(wù)
ensure_future 如果對象是一個Future對象，就返回，否則就會調(diào)用create_task返回，并且加入到_ready隊列

futures.py

類/函數(shù) 方法 對象 作用 描述
Future 主要負(fù)責(zé)與用戶函數(shù)進行交互
__init__ 初始化兩個重要對象 self._loop 與 self._callbacks
self._loop 事件循環(huán)
self._callbacks 回調(diào)隊列，任務(wù)暫存隊列，等待時機成熟（狀態(tài)不是PENDING），就會進入_ready隊列
add_done_callback 添加任務(wù)回調(diào)函數(shù)，狀態(tài)_PENDING，就虎進入_callbacks隊列，否則進入_ready隊列
set_result 獲取任務(wù)執(zhí)行結(jié)果并存儲至_result，將狀態(tài)置位_FINISH，調(diào)用__schedule_callbacks
__schedule_callbacks 將回調(diào)函數(shù)放入_ready，等待執(zhí)行
result 獲取返回值
__await__ 使用await就會進入這個方法
__iter__ 使用yield from就會進入這個方法
set_result_unless_cancelled 其實就是Future.set_result，只不過調(diào)用場景與調(diào)用方式不一樣 新增

新加了很多的函數(shù)，后面我們邊走流程，邊講解他們的用途

3）執(zhí)行過程

3.1）入口函數(shù)

main.py

    
if __name__ == "__main__":
    ret = wilsonasyncio.run(helloworld())
    print(ret)

3.2）事件循環(huán)啟動，同gather，不再贅述

3.3）第一次循環(huán)run_forever --> run_once，同gather，不再贅述

3.3.1）gather完成，回到helloworld()，同gather，不再贅述

3.4）第二次循環(huán)run_forever --> run_once，從這里開始，不一樣的地方來了

  • 從隊列中取出數(shù)據(jù)，此時_ready隊列有兩個任務(wù)，hello() world()，在gather的for循環(huán)時添加的
    def run_once(self):
        timeout = 0
        if not self._ready and self._scheduled:
            heapq.heapify(self._scheduled)
            when = self._scheduled[0]._when
            timeout = min(max(0, when - self.current_time()), 60)
        self._selector.select(timeout)

        end_time = self.current_time()
        while self._scheduled:
            handle = self._scheduled[0]
            if handle._when >= end_time:
                break
            handle = heapq.heappop(self._scheduled)
            self._ready.append(handle)
        

        ntodo = len(self._ready)
        for _ in range(ntodo):
            handle = self._ready.popleft()
            handle._run()
  • run_once做了改動，由于if not self._ready and self._scheduled不滿足（self._ready有內(nèi)容）
  • timeout=0，self._selector.select(timeout)不會等待，直接跳過
  • while self._scheduled不滿足，不會進入while循環(huán)
  • 直接進入handle._run()階段

上述的函數(shù)都沒有執(zhí)行，所以沒有分析，在后面執(zhí)行的時候會詳細(xì)分析作用

tasks.py

    def __step(self, exc=None):
        coro = self._coro
        try:
            if exc is None:
                result = coro.send(None)
            else:
                result = coro.throw(exc)
        except StopIteration as exc:
            super().set_result(exc.value)
        else:
            blocking = getattr(result, '_asyncio_future_blocking', None)
            if blocking:
                result._asyncio_future_blocking = False
                result.add_done_callback(self.__wakeup, result)
        finally:
            self = None
  • 經(jīng)過coro.send(None)回到各自的函數(shù)本體中
async def hello():
    print('enter hello ...')
    await wilsonasyncio.sleep(5)
    return 'return hello...'

async def world():
    print('enter world ...')
    await wilsonasyncio.sleep(3)
    return 'return world...'
  • 由于hello() world()使用await調(diào)用了sleep，我們開看看sleep的源碼：
async def sleep(delay, result=None, *, loop=None):
    if loop is None:
        loop = get_event_loop()
    future = loop.create_future()
    loop.call_later(delay, set_result_unless_cancelled, future, result)
    return await future
  • future = loop.create_future()創(chuàng)建了一個Future對象，隨后調(diào)用call_later
    def call_later(self, delay, callback, *args):
        timer = self.call_at(self.current_time() + delay, callback, *args)
        return timer

    def call_at(self, when, callback, *args):
        timer = TimerHandle(when, callback, *args)
        heapq.heappush(self._scheduled, timer)
        return timer

  • call_later call_at，主要的邏輯：
    ? ? ? ? a) self.current_time()獲取當(dāng)前的時間線，再加上傳入的delay（hello(5)``world(3)）計算出延時
    ? ? ? ? b) timer = TimerHandle(when, callback, *args)將延時、callback（即set_result_unless_cancelled）再加上參數(shù)封裝成TimerHandle對象
    ? ? ? ? c) heapq.heappush(self._scheduled, timer)將對象推入self._scheduled隊列等待合適的時間調(diào)度到self._ready
    ? ? ? ? d) 第一點需要注意的是，self._scheduled是一個最小堆
    ? ? ? ? e) 第二點需要注意的是，TimerHandle實現(xiàn)了__lt__ __gt__ __eq__，所以會通過self._when進行對象比較。在本例中，hello()的延時是current_time+5，world()的延時是current_time+3，所以world()會優(yōu)先調(diào)度

  • 返回到sleep

async def sleep(delay, result=None, *, loop=None):
    if loop is None:
        loop = get_event_loop()
    future = loop.create_future()
    loop.call_later(delay, set_result_unless_cancelled, future, result)
    return await future
  • return await future在gather小節(jié)中描述過，一旦調(diào)用了await，就會來到__await__
    def __await__(self):
        if self._state == _PENDING:
            self._asyncio_future_blocking = True
            yield self
        return self.result()
  • yield self self就是之前創(chuàng)建的future對象，并且返回到當(dāng)初send的地方，對的，就是Task.__step里面
    def __step(self, exc=None):
        coro = self._coro
        try:
            if exc is None:
                result = coro.send(None)
            else:
                result = coro.throw(exc)
        except StopIteration as exc:
            super().set_result(exc.value)
        else:
            blocking = getattr(result, '_asyncio_future_blocking', None)
            if blocking:
                result._asyncio_future_blocking = False
                result.add_done_callback(self.__wakeup, result)
        finally:
            self = None

  • 由于他們?nèi)?code>__await__溜達了一圈，所以_asyncio_future_blocking=True，所以hello() world()的回調(diào)函數(shù)是self.__wakeup

  • 這一次循環(huán)結(jié)束了，hello() world()并沒有結(jié)束，而是分分掛起，等待他們的子任務(wù)await wilsonasyncio.sleep執(zhí)行結(jié)束完之后才會喚醒

3.5）第三次循環(huán)run_forever --> run_once

eventloops:

    def run_once(self):
        timeout = 0
        if not self._ready and self._scheduled:
            heapq.heapify(self._scheduled)
            when = self._scheduled[0]._when
            timeout = min(max(0, when - self.current_time()), 60)
        self._selector.select(timeout)

        end_time = self.current_time()
        while self._scheduled:
            handle = self._scheduled[0]
            if handle._when >= end_time:
                break
            handle = heapq.heappop(self._scheduled)
            self._ready.append(handle)
        

        ntodo = len(self._ready)
        for _ in range(ntodo):
            handle = self._ready.popleft()
            handle._run()
  • run_once函數(shù)進行了改造
  • self._ready為空，并且self._scheduled隊列不為空，里面的內(nèi)容是hello() world()2個函數(shù)中調(diào)用await wilsonasyncio.sleep()時添加進self._scheduled，分別是兩個TimerHandle對象，并且是按照最小堆排序
  • 由于是按照最小堆排序，直接取出來肯定就是時間最小的，timeout = min(max(0, when - self.current_time()), 60)，獲取timeout
  • self._selector.select(timeout)核心代碼，主要的邏輯就會等待timeout的時間，是整個線程等待，所有的協(xié)程任務(wù)都會卡主，但是當(dāng)前隊列里面所有的任務(wù)都是等待，并且當(dāng)前的timeout是等待時間最小的。在本例中，timeout就是world()的等待時間：3s

self._selector.select(timeout) 這里要詳細(xì)描述一下了

  • 核心邏輯就是將一堆等待中的協(xié)程任務(wù)中，選一個等待時間最小的出來，然后線程sleep
  • 這時的sleep可以用粗暴的time.sleep，但是我們選擇了self._selector.select(timeout)，主要_selector在后面的應(yīng)用中發(fā)揮了巨大的作用，只是在這里看不出來
  • self._selector是一個多路復(fù)用對象，就是我們熟悉的select() poll() epoll()，我們將一條socket注冊到對象上面，然后調(diào)用self._selector.select(timeout)，整個線程就會在timeout期間，監(jiān)聽socket是否有新的數(shù)據(jù)，一旦socket有新的數(shù)據(jù)，就會立即執(zhí)行回調(diào)函數(shù)
  • 那socket對象是怎么來的，event_loop初始化的時候，會通過socket.socketpair()創(chuàng)建一對socket，并且把它注冊進self._selector，所以self._selector.select(timeout)會監(jiān)聽socket是否有讀寫事件
  • sel.register(_ssock.fileno(), selectors.EVENT_READ, None)當(dāng)然，我們的回調(diào)函數(shù)是None，沒有注冊任何回調(diào)，在本例中，我們只需要它的等待功能
  • 肯定會有人問了，那如果這時候出現(xiàn)了一個需要立即執(zhí)行的任務(wù)，怎么辦？這是后面解答的問題，在后面的輪子中肯定會解答這個問題，而在本例中，不可能會出現(xiàn)這種情況（如果是立即執(zhí)行的任務(wù)，那么一定出現(xiàn)在self._ready隊列里面，在第一次循環(huán)就已經(jīng)執(zhí)行完成了）

簡單解釋了一下，我們喝一口水，繼續(xù)。。。

  • end_time = self.current_time()記錄一下當(dāng)前的時間
        while self._scheduled:
            handle = self._scheduled[0]
            if handle._when >= end_time:
                break
            handle = heapq.heappop(self._scheduled)
            self._ready.append(handle)
  • 將_when大于當(dāng)前時間的協(xié)程任務(wù)取出來，放入self._ready隊列，等待下一次循環(huán)調(diào)度
  • _when小于當(dāng)前事件的協(xié)程任務(wù)留在self._scheduled，他們還需要繼續(xù)睡覺
  • 全部搞完，繼續(xù)往下走
        ntodo = len(self._ready)
        for _ in range(ntodo):
            handle = self._ready.popleft()
            handle._run()
  • 熟悉的流程，handle是world()的sleep，執(zhí)行其回調(diào)函數(shù)set_result_unless_cancelled，就是set_result
  • 將__wakeup加入到self._ready隊列，world()等待被喚醒

3.6）第四次循環(huán)run_forever --> run_once

if not self._ready and self._scheduled:
  • timeout=0, self._ready有協(xié)程任務(wù)，不會等待
  • __wakeup --> __step --> send(None) --> 回到 world()
async def world():
    print('enter world ...')
    await wilsonasyncio.sleep(3)
    print('world sleep end...')
    return 'return world...'
  • world()協(xié)程完成，將回調(diào)_done_callback加入到self._ready隊列，同gather

3.7）第五次循環(huán)run_forever --> run_once

  • gather._done_callback()，由于條件不滿足，不會添加回調(diào)到self._ready，同gather

3.8）第六次循環(huán)run_forever --> run_once

        if not self._ready and self._scheduled:
            heapq.heapify(self._scheduled)
            when = self._scheduled[0]._when
            timeout = min(max(0, when - self.current_time()), 60)

  • timeout = min(max(0, when - self.current_time()), 60)，由于剛才等了3s，hello()的delay是5s，所以這里timeout=2（這之間還有很多代碼的執(zhí)行時間，不過執(zhí)行速度非?？欤?/p>

  • self._selector.select(timeout)等待2s之后，重復(fù)了world()的流程

  • handle是hello()的sleep，執(zhí)行其回調(diào)函數(shù)set_result_unless_cancelled，就是set_result

  • 將__wakeup加入到self._ready隊列，hello()等待被喚醒

  • results.append(res)將子任務(wù)的結(jié)果取出來，放進父任務(wù)的results里面

  • 子任務(wù)執(zhí)行完成，終于到了喚醒父任務(wù)的時候了task.__wakeup

    def __wakeup(self, future):
        try:
            future.result()
        except Exception as exc:
            raise exc
        else:
            self.__step()
        self = None

3.8）第七次循環(huán)run_forever --> run_once

  • __wakeup --> __step --> send(None) --> 回到 hello()
async def hello():
    print('enter hello ...')
    await wilsonasyncio.sleep(5)
    print('hello sleep end...')
    return 'return hello...'
  • hello()協(xié)程完成，將回調(diào)_done_callback加入到self._ready隊列，同gather

3.9）第八次循環(huán)run_forever --> run_once

  • gather._done_callback()，由于條件滿足，將回調(diào)到self._ready，helloworld等待被喚醒。同上一小節(jié)gather

3.10）第九次循環(huán)run_forever --> run_once

  • 循環(huán)結(jié)束
  • 回到run

3.11）回到主函數(shù)，獲取返回值

if __name__ == "__main__":
    ret = wilsonasyncio.run(helloworld())
    print(ret)

3.12）執(zhí)行結(jié)果

? python3 main.py
enter helloworld
enter hello ...
enter world ...
world sleep end...
hello sleep end...
exit helloworld
['return hello...', 'return world...']

五、流程總結(jié)

六、小結(jié)

● 無法總結(jié)。。。。有問題私信、留言
● 本文中的代碼，參考了python 3.7.7中asyncio的源代碼，裁剪而來
● 本文中代碼：代碼


至此，本文結(jié)束
在下才疏學(xué)淺，有撒湯漏水的，請各位不吝賜教...
更多文章，請關(guān)注我：wilson.chai


網(wǎng)頁名稱：從頭造輪子：python3 asyncio之 sleep （4）
URL網(wǎng)址：http://weahome.cn/article/dsojodi.html

其他資訊

在線咨詢

微信咨詢

電話咨詢

028-86922220（工作日）

18980820575（7×24）

提交需求

返回頂部