使用Python实现一个优雅的异步定时器

更新时间：2025年04月20日 11:40:19 作者：有人说风轻的像雨

在 Python 中实现定时器功能是一个常见需求,尤其是在需要周期性执行任务的场景下,本文给大家介绍了基于 asyncio 和 threading 模块,可扩展的异步定时器实现,需要的朋友可以参考下

需求背景

定时器的核心功能是能够周期性地触发回调函数，同时需要支持启动、停止以及状态检查等操作。在多线程或异步编程场景中，希望定时器能够：

支持异步操作，避免阻塞主线程；
单例化事件循环，节省资源；
优雅地管理定时器的生命周期；
提供简单的接口，易于使用。

为此，设计了一个 Timer 类，结合 asyncio 和 threading，实现了一个高效的定时器。

代码

完整代码

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"事件循环异常: {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，因为守护线程会在主线程退出时自动结束

    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"定时器循环异常: {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}秒执行一次")

    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"回调1触发: {time.strftime('%H:%M:%S')}")

def callback2():
    print(f"回调2触发: {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. 单例事件循环的实现

为了避免每个定时器都创建一个独立的事件循环，在 Timer 类中使用了类变量和类方法来管理全局唯一的事件循环：

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 事件循环。
_thread：将事件循环运行在一个独立的守护线程中，避免阻塞主线程。
_lock：线程锁，确保在多线程环境中创建事件循环时的线程安全。
_ensure_loop：在需要时创建或重用事件循环，确保只有一个全局循环。

守护线程（daemon=True）的设计使得程序退出时无需显式关闭线程，简化了资源清理。

2. 事件循环的运行与关闭

事件循环的运行逻辑封装在 _run_loop 中：

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

run_forever：让事件循环持续运行，直到被外部停止。
异常处理：捕获可能的错误并打印，便于调试。
finally：确保循环关闭时资源被正确释放。

关闭逻辑则由 _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)

当所有定时器都停止时（_running_timers == 0），事件循环会被安全停止。

3. 定时器核心逻辑

每个 Timer 实例负责管理一个独立的定时任务：

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：异步协程，每隔 interval 秒执行一次回调函数 callback。
run_in_executor：将回调函数运行在默认的线程池中，避免阻塞事件循环。

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：启动定时器，确保事件循环可用，并记录运行中的定时器数量。
stop：通过设置 _stop_event 并取消任务来停止定时器。

5. 使用示例

以下是一个简单的使用示例：

def callback1():
    print(f"回调1触发: {time.strftime('%H:%M:%S')}")

def callback2():
    print(f"回调2触发: {time.strftime('%H:%M:%S')}")

timer1 = Timer()
timer2 = Timer()

timer1.start(2, callback1)  # 每2秒触发一次
timer2.start(3, callback2)  # 每3秒触发一次

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

输出可能如下：

回调1触发: 14:30:02
回调2触发: 14:30:03
回调1触发: 14:30:04
回调1触发: 14:30:06
回调2触发: 14:30:06
...

设计亮点

异步与多线程结合：通过 asyncio 和 threading，实现了非阻塞的定时器，适合高并发场景。
资源高效利用：全局唯一的事件循环避免了重复创建的开销。
优雅的生命周期管理：守护线程和自动关闭机制简化了资源清理。
线程安全：使用锁机制确保多线程环境下的稳定性。

适用场景

周期性任务调度，如数据刷新、状态检查。
后台服务中的定时监控。
游戏或实时应用中的计时器需求。

总结

这个异步定时器实现结合了 Python 的异步编程和多线程特性，提供了一个轻量、灵活的解决方案。无论是简单的脚本还是复杂的后台服务，它都能胜任。如果你需要一个可靠的定时器，不妨试试这个实现，或者根据需求进一步优化它！

以上就是使用Python实现一个优雅的异步定时器的详细内容，更多关于Python异步定时器的资料请关注脚本之家其它相关文章！

您可能感兴趣的文章:

Python面向对象程序设计之继承与多继承用法分析
这篇文章主要介绍了Python面向对象程序设计之继承与多继承用法,结合实例形式分析了Python继承与多继承的简单定义与使用方法,需要的朋友可以参考下
2018-07-07
OPCUA-Python实例
这篇文章主要介绍了OPCUA-Python实例,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
2024-02-02
django 删除数据库表后重新同步的方法
今天小编就为大家分享一篇django 删除数据库表后重新同步的方法，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2018-05-05
python实现txt文件格式转换为arff格式
这篇文章主要为大家详细介绍了python实现txt文件格式转换为arff格式的方法，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
2018-05-05
Python中的字符串查找操作方法总结
这里我们来整理一下Python中的字符串查找操作方法总结,除了基本的find()方法外,还会讲解到朴素匹配算法和KMP算法的使用:
2016-06-06
Python中最大最小赋值小技巧(分享)
下面小编就为大家分享一篇Python中最大最小赋值小技巧，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2017-12-12
Python数据处理篇之Sympy系列(五)---解方程
这篇文章主要介绍了Python数据处理篇之Sympy系列(五)---解方程,非常不错，具有一定的参考借鉴价值，需要的朋友可以参考下
2019-10-10
Python实现滑块拼图验证码详解
验证码作为一种自然人的机器人的判别工具，被广泛的用于各种防止程序做自动化的场景中。传统的字符型验证安全性已经名存实亡的情况下，各种新型的验证码如雨后春笋般涌现，今天给大家分享一篇Python实现滑块验证码
2022-05-05
python使用Flask操作mysql实现登录功能
这篇文章主要介绍了python使用Flask操作mysql实现登录功能,代码简单易懂，非常不错，具有一定的参考借鉴价值，需要的朋友可以参考下
2018-05-05
五个Python迷你版小程序附代码
在使用Python的过程中，我最喜欢的就是Python的各种第三方库，能够完成很多操作。下面就给大家介绍5个通过 Python 构建的实战项目，来实践 Python 编程能力。欢迎收藏学习，喜欢点赞支持
2021-11-11