源碼解讀Python中Event事件的使用

更新時間：2023年12月22日 09:14:30 作者：古明地覺的編程教室

事件（Event）主要負責多任務之間的同步,這篇文章主要來和大家詳細介紹一下它的原理以及簡單使用,文中的示例代碼講解詳細,感興趣的小伙伴可以了解下

協(xié)程 Event

協(xié)程 Event 由 asyncio 模塊提供。

import asyncio
from asyncio import Event

async def task(event: Event):
    # 如果 event 內部的標志位是 False，會陷入阻塞
    print(f"陷入阻塞，因為標志位 = {event.is_set()}")
    await event.wait()
    print(f"解除阻塞，因為標志位 = {event.is_set()}")

async def main():
    event = Event()
    # 任務開始執(zhí)行
    asyncio.create_task(task(event))
    await asyncio.sleep(3)
    # task 內部的 event.wait() 會陷入阻塞
    # 3 秒將標志設置為 True
    print("將 event 內部的標志位設置為 True")
    event.set()

asyncio.run(main())
"""
陷入阻塞，因為標志位 = False
將 event 內部的標志位設置為 True
解除阻塞，因為標志位 = True
"""

非常簡單，當調用 event.wait() 時，如果標志是 True，那么相當于綠燈，直接通過；如果標志是 False，那么相當于紅燈，需要等待。

默認情況下是紅燈，通過 event.set() 可以設置為綠燈，也可以通過 event.clear() 重置為紅燈。調用 is_set() 方法可以判斷當前是紅燈還是綠燈，True 為綠燈，False 為紅燈。

然后再來看看它的源碼實現：

當標志位是 False，協(xié)程調用 wait 方法會陷入阻塞，那么阻塞要如何實現呢？沒錯，還是要通過 Future 對象。所以 Future 對象和 asyncio 的實現緊密相關，協(xié)程里面的阻塞等待都是基于 Future 實現的。

而 _waiters 負責保存協(xié)程內部創(chuàng)建的 Future 對象，_value 則表示標志位。至于 _loop 則用于指定事件循環(huán)，這個參數已經廢棄了。

is_set() 方法用于查看標志位，clear() 方法用于將標志位設置為 False，比較簡單。

然后是 wait() 方法，如果調用時發(fā)現標志位是 True，那么說明是綠燈，直接通過。否則說明是紅燈，于是創(chuàng)建一個 Future 對象，并添加到 _waiters 中，然后 await 它，從而陷入阻塞。

最后是 set() 方法，如果標志位是 False，將其設置為 True。然后將 _waiters 里面的 future 依次彈出，設置結果集，讓 await fut 的協(xié)程解除阻塞。

整個過程沒有任何難度，非常簡單。

線程 Event

線程 Event 也很簡單，它是由 threading 模塊提供的。

import time
import threading
from threading import Event

def task():
    # 如果 event 內部的標志位是 False，會陷入阻塞
    print(f"陷入阻塞，因為標志位 = {event.is_set()}")
    event.wait()
    print(f"解除阻塞，因為標志位 = {event.is_set()}")

def main():
    time.sleep(3)
    print("將 event 內部的標志位設置為 True")
    event.set()

event = Event()
t1 = threading.Thread(target=task)
t2 = threading.Thread(target=main)
t1.start()
t2.start()
"""
陷入阻塞，因為標志位 = False
將 event 內部的標志位設置為 True
解除阻塞，因為標志位 = True
"""

用法和協(xié)程 Event 幾乎沒什么區(qū)別，然后看一下它的內部實現。

class Event:

    def __init__(self):
        # 條件對象，所以事件對象其實是基于條件對象的一個封裝
        self._cond = Condition(Lock())
        # 標志位，初始為 False
        self._flag = False

    def is_set(self):
        # 標志位是否被設置
        return self._flag

    isSet = is_set

    def set(self):
        # 修改共享變量時需要加鎖保護
        with self._cond:
            # 設置標志位，并喚醒所有阻塞線程
            self._flag = True
            self._cond.notify_all()

    def clear(self):
        # 將標志位設置為 False
        with self._cond:
            self._flag = False

    def wait(self, timeout=None):
        # 阻塞等待，但支持超時時間
        with self._cond:
            signaled = self._flag
            if not signaled:
                signaled = self._cond.wait(timeout)
            return signaled

非常簡單，Event 內部是基于 Condition 實現的，關于條件變量，我們后續(xù)再詳細介紹。