使用Python實現(xiàn)一個優(yōu)雅的異步定時器

更新時間：2025年04月20日 11:40:19 作者：有人說風(fēng)輕的像雨

在 Python 中實現(xiàn)定時器功能是一個常見需求,尤其是在需要周期性執(zhí)行任務(wù)的場景下,本文給大家介紹了基于 asyncio 和 threading 模塊,可擴展的異步定時器實現(xiàn),需要的朋友可以參考下

需求背景

定時器的核心功能是能夠周期性地觸發(fā)回調(diào)函數(shù)，同時需要支持啟動、停止以及狀態(tài)檢查等操作。在多線程或異步編程場景中，希望定時器能夠：

支持異步操作，避免阻塞主線程；
單例化事件循環(huán)，節(jié)省資源；
優(yōu)雅地管理定時器的生命周期；
提供簡單的接口，易于使用。

為此，設(shè)計了一個 Timer 類，結(jié)合 asyncio 和 threading，實現(xiàn)了一個高效的定時器。

代碼

完整代碼

import asyncio
import threading
import time
import sys

class Timer:
    _loop = None
    _thread = None
    _lock = threading.Lock()
    _running_timers = 0

    @classmethod
    def _ensure_loop(cls):
        with cls._lock:
            if cls._loop is None or not cls._thread or not cls._thread.is_alive():
                cls._loop = asyncio.new_event_loop()
                cls._thread = threading.Thread(
                    target=cls._run_loop,
                    args=(cls._loop,),
                    daemon=True
                )
                cls._thread.start()

    @classmethod
    def _run_loop(cls, loop):
        asyncio.set_event_loop(loop)
        try:
            loop.run_forever()
        except Exception as e:
            print(f"事件循環(huán)異常: {e}")
        finally:
            loop.close()

    @classmethod
    def _shutdown(cls):
        with cls._lock:
            if cls._running_timers == 0 and cls._loop is not None and cls._loop.is_running():
                cls._loop.call_soon_threadsafe(cls._loop.stop)
                # 不使用 join，因為守護線程會在主線程退出時自動結(jié)束

    def __init__(self):
        self.is_running = False
        self._stop_event = asyncio.Event()
        self._task = None

    async def _timer_loop(self, interval, callback):
        try:
            while not self._stop_event.is_set():
                await asyncio.sleep(interval)
                if not self._stop_event.is_set():
                    await asyncio.get_event_loop().run_in_executor(None, callback)
        except asyncio.CancelledError:
            pass  # 正常取消時忽略
        except Exception as e:
            print(f"定時器循環(huán)異常: {e}")
        finally:
            self.is_running = False
            Timer._running_timers -= 1
            Timer._shutdown()

    def start(self, interval, callback):
        if not self.is_running:
            Timer._ensure_loop()
            self.is_running = True
            self._stop_event.clear()
            self._task = asyncio.run_coroutine_threadsafe(
                self._timer_loop(interval, callback),
                Timer._loop
            )
            Timer._running_timers += 1
            # print(f"定時器已啟動，每{interval}秒執(zhí)行一次")

    def stop(self):
        if self.is_running:
            self._stop_event.set()
            if self._task:
                Timer._loop.call_soon_threadsafe(self._task.cancel)
            self.is_running = False
            # print("定時器已停止")

    def is_active(self):
        return self.is_running

# 使用示例
def callback1():
    print(f"回調(diào)1觸發(fā): {time.strftime('%H:%M:%S')}")

def callback2():
    print(f"回調(diào)2觸發(fā): {time.strftime('%H:%M:%S')}")

if __name__ == "__main__":
    timer1 = Timer()
    timer2 = Timer()

    timer1.start(2, callback1)
    timer2.start(3, callback2)

    try:
        time.sleep(100)
        timer1.stop()
        time.sleep(2)
        timer2.stop()
    except KeyboardInterrupt:
        timer1.stop()
        timer2.stop()
    finally:
        # 確保在程序退出時清理
        Timer._shutdown()

1. 單例事件循環(huán)的實現(xiàn)

為了避免每個定時器都創(chuàng)建一個獨立的事件循環(huán)，在 Timer 類中使用了類變量和類方法來管理全局唯一的事件循環(huán)：

class Timer:
    _loop = None
    _thread = None
    _lock = threading.Lock()
    _running_timers = 0

    @classmethod
    def _ensure_loop(cls):
        with cls._lock:
            if cls._loop is None or not cls._thread or not cls._thread.is_alive():
                cls._loop = asyncio.new_event_loop()
                cls._thread = threading.Thread(
                    target=cls._run_loop,
                    args=(cls._loop,),
                    daemon=True
                )
                cls._thread.start()

_loop：存儲全局的 asyncio 事件循環(huán)。
_thread：將事件循環(huán)運行在一個獨立的守護線程中，避免阻塞主線程。
_lock：線程鎖，確保在多線程環(huán)境中創(chuàng)建事件循環(huán)時的線程安全。
_ensure_loop：在需要時創(chuàng)建或重用事件循環(huán)，確保只有一個全局循環(huán)。

守護線程（daemon=True）的設(shè)計使得程序退出時無需顯式關(guān)閉線程，簡化了資源清理。

2. 事件循環(huán)的運行與關(guān)閉

事件循環(huán)的運行邏輯封裝在 _run_loop 中：

@classmethod
def _run_loop(cls, loop):
    asyncio.set_event_loop(loop)
    try:
        loop.run_forever()
    except Exception as e:
        print(f"事件循環(huán)異常: {e}")
    finally:
        loop.close()

run_forever：讓事件循環(huán)持續(xù)運行，直到被外部停止。
異常處理：捕獲可能的錯誤并打印，便于調(diào)試。
finally：確保循環(huán)關(guān)閉時資源被正確釋放。

關(guān)閉邏輯則由 _shutdown 方法控制：

@classmethod
def _shutdown(cls):
    with cls._lock:
        if cls._running_timers == 0 and cls._loop is not None and cls._loop.is_running():
            cls._loop.call_soon_threadsafe(cls._loop.stop)

當(dāng)所有定時器都停止時（_running_timers == 0），事件循環(huán)會被安全停止。

3. 定時器核心邏輯

每個 Timer 實例負責(zé)管理一個獨立的定時任務(wù)：

def __init__(self):
    self.is_running = False
    self._stop_event = asyncio.Event()
    self._task = None

async def _timer_loop(self, interval, callback):
    try:
        while not self._stop_event.is_set():
            await asyncio.sleep(interval)
            if not self._stop_event.is_set():
                await asyncio.get_event_loop().run_in_executor(None, callback)
    except asyncio.CancelledError:
        pass  # 正常取消時忽略
    finally:
        self.is_running = False
        Timer._running_timers -= 1
        Timer._shutdown()

_stop_event：一個 asyncio.Event 對象，用于控制定時器的停止。
_timer_loop：異步協(xié)程，每隔 interval 秒執(zhí)行一次回調(diào)函數(shù) callback。
run_in_executor：將回調(diào)函數(shù)運行在默認的線程池中，避免阻塞事件循環(huán)。

4. 啟動與停止

啟動和停止方法是用戶的主要接口：

def start(self, interval, callback):
    if not self.is_running:
        Timer._ensure_loop()
        self.is_running = True
        self._stop_event.clear()
        self._task = asyncio.run_coroutine_threadsafe(
            self._timer_loop(interval, callback),
            Timer._loop
        )
        Timer._running_timers += 1

def stop(self):
    if self.is_running:
        self._stop_event.set()
        if self._task:
            Timer._loop.call_soon_threadsafe(self._task.cancel)
        self.is_running = False

start：啟動定時器，確保事件循環(huán)可用，并記錄運行中的定時器數(shù)量。
stop：通過設(shè)置 _stop_event 并取消任務(wù)來停止定時器。

5. 使用示例

以下是一個簡單的使用示例：

def callback1():
    print(f"回調(diào)1觸發(fā): {time.strftime('%H:%M:%S')}")

def callback2():
    print(f"回調(diào)2觸發(fā): {time.strftime('%H:%M:%S')}")

timer1 = Timer()
timer2 = Timer()

timer1.start(2, callback1)  # 每2秒觸發(fā)一次
timer2.start(3, callback2)  # 每3秒觸發(fā)一次

time.sleep(10)  # 運行10秒
timer1.stop()
timer2.stop()

輸出可能如下：

回調(diào)1觸發(fā): 14:30:02
回調(diào)2觸發(fā): 14:30:03
回調(diào)1觸發(fā): 14:30:04
回調(diào)1觸發(fā): 14:30:06
回調(diào)2觸發(fā): 14:30:06
...

設(shè)計亮點

異步與多線程結(jié)合：通過 asyncio 和 threading，實現(xiàn)了非阻塞的定時器，適合高并發(fā)場景。
資源高效利用：全局唯一的事件循環(huán)避免了重復(fù)創(chuàng)建的開銷。
優(yōu)雅的生命周期管理：守護線程和自動關(guān)閉機制簡化了資源清理。
線程安全：使用鎖機制確保多線程環(huán)境下的穩(wěn)定性。

適用場景

周期性任務(wù)調(diào)度，如數(shù)據(jù)刷新、狀態(tài)檢查。
后臺服務(wù)中的定時監(jiān)控。
游戲或?qū)崟r應(yīng)用中的計時器需求。

總結(jié)

這個異步定時器實現(xiàn)結(jié)合了 Python 的異步編程和多線程特性，提供了一個輕量、靈活的解決方案。無論是簡單的腳本還是復(fù)雜的后臺服務(wù)，它都能勝任。如果你需要一個可靠的定時器，不妨試試這個實現(xiàn)，或者根據(jù)需求進一步優(yōu)化它！

以上就是使用Python實現(xiàn)一個優(yōu)雅的異步定時器的詳細內(nèi)容，更多關(guān)于Python異步定時器的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章:

欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

使用Python實現(xiàn)一個優(yōu)雅的異步定時器

目錄

需求背景

代碼

1. 單例事件循環(huán)的實現(xiàn)

2. 事件循環(huán)的運行與關(guān)閉

3. 定時器核心邏輯

4. 啟動與停止

5. 使用示例

設(shè)計亮點

適用場景

總結(jié)

相關(guān)文章

最新評論

大家感興趣的內(nèi)容

最近更新的內(nèi)容

常用在線小工具