需求背景

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

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

为此，设计了一个

1. Timer

复制代码

类，结合

1. asyncio

复制代码

和

1. threading

复制代码

，实现了一个高效的定时器。

代码

完整代码

1. import asyncio
   
2. import threading
   
3. import time
   
4. import sys
   
5. 
   
6. class Timer:
   
7.     _loop = None
   
8.     _thread = None
   
9.     _lock = threading.Lock()
   
10.     _running_timers = 0
    
11. 
    
12.     @classmethod
    
13.     def _ensure_loop(cls):
    
14.         with cls._lock:
    
15.             if cls._loop is None or not cls._thread or not cls._thread.is_alive():
    
16.                 cls._loop = asyncio.new_event_loop()
    
17.                 cls._thread = threading.Thread(
    
18.                     target=cls._run_loop,
    
19.                     args=(cls._loop,),
    
20.                     daemon=True
    
21.                 )
    
22.                 cls._thread.start()
    
23. 
    
24.     @classmethod
    
25.     def _run_loop(cls, loop):
    
26.         asyncio.set_event_loop(loop)
    
27.         try:
    
28.             loop.run_forever()
    
29.         except Exception as e:
    
30.             print(f"事件循环异常: {e}")
    
31.         finally:
    
32.             loop.close()
    
33. 
    
34.     @classmethod
    
35.     def _shutdown(cls):
    
36.         with cls._lock:
    
37.             if cls._running_timers == 0 and cls._loop is not None and cls._loop.is_running():
    
38.                 cls._loop.call_soon_threadsafe(cls._loop.stop)
    
39.                 # 不使用 join，因为守护线程会在主线程退出时自动结束
    
40. 
    
41.     def __init__(self):
    
42.         self.is_running = False
    
43.         self._stop_event = asyncio.Event()
    
44.         self._task = None
    
45. 
    
46.     async def _timer_loop(self, interval, callback):
    
47.         try:
    
48.             while not self._stop_event.is_set():
    
49.                 await asyncio.sleep(interval)
    
50.                 if not self._stop_event.is_set():
    
51.                     await asyncio.get_event_loop().run_in_executor(None, callback)
    
52.         except asyncio.CancelledError:
    
53.             pass  # 正常取消时忽略
    
54.         except Exception as e:
    
55.             print(f"定时器循环异常: {e}")
    
56.         finally:
    
57.             self.is_running = False
    
58.             Timer._running_timers -= 1
    
59.             Timer._shutdown()
    
60. 
    
61.     def start(self, interval, callback):
    
62.         if not self.is_running:
    
63.             Timer._ensure_loop()
    
64.             self.is_running = True
    
65.             self._stop_event.clear()
    
66.             self._task = asyncio.run_coroutine_threadsafe(
    
67.                 self._timer_loop(interval, callback),
    
68.                 Timer._loop
    
69.             )
    
70.             Timer._running_timers += 1
    
71.             # print(f"定时器已启动，每{interval}秒执行一次")
    
72. 
    
73.     def stop(self):
    
74.         if self.is_running:
    
75.             self._stop_event.set()
    
76.             if self._task:
    
77.                 Timer._loop.call_soon_threadsafe(self._task.cancel)
    
78.             self.is_running = False
    
79.             # print("定时器已停止")
    
80. 
    
81.     def is_active(self):
    
82.         return self.is_running
    
83. 
    
84. # 使用示例
    
85. def callback1():
    
86.     print(f"回调1触发: {time.strftime('%H:%M:%S')}")
    
87. 
    
88. def callback2():
    
89.     print(f"回调2触发: {time.strftime('%H:%M:%S')}")
    
90. 
    
91. if __name__ == "__main__":
    
92.     timer1 = Timer()
    
93.     timer2 = Timer()
    
94. 
    
95.     timer1.start(2, callback1)
    
96.     timer2.start(3, callback2)
    
97. 
    
98.     try:
    
99.         time.sleep(100)
    
100.         timer1.stop()
     
101.         time.sleep(2)
     
102.         timer2.stop()
     
103.     except KeyboardInterrupt:
     
104.         timer1.stop()
     
105.         timer2.stop()
     
106.     finally:
     
107.         # 确保在程序退出时清理
     
108.         Timer._shutdown()

复制代码

1. 单例事件循环的实现

为了避免每个定时器都创建一个独立的事件循环，在

1. Timer

复制代码

类中使用了类变量和类方法来管理全局唯一的事件循环：

1. class Timer:
   
2.     _loop = None
   
3.     _thread = None
   
4.     _lock = threading.Lock()
   
5.     _running_timers = 0
   
6. 
   
7.     @classmethod
   
8.     def _ensure_loop(cls):
   
9.         with cls._lock:
   
10.             if cls._loop is None or not cls._thread or not cls._thread.is_alive():
    
11.                 cls._loop = asyncio.new_event_loop()
    
12.                 cls._thread = threading.Thread(
    
13.                     target=cls._run_loop,
    
14.                     args=(cls._loop,),
    
15.                     daemon=True
    
16.                 )
    
17.                 cls._thread.start()

复制代码

• 1. _loop

  复制代码
  ：存储全局的
  1. asyncio

  复制代码
  事件循环。
  
• 1. _thread

  复制代码
  ：将事件循环运行在一个独立的守护线程中，避免阻塞主线程。
  
• 1. _lock

  复制代码
  ：线程锁，确保在多线程环境中创建事件循环时的线程安全。
  
• 1. _ensure_loop

  复制代码
  ：在需要时创建或重用事件循环，确保只有一个全局循环。
  

守护线程（

1. daemon=True

复制代码

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

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

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

1. _run_loop

复制代码

中：

1. @classmethod
   
2. def _run_loop(cls, loop):
   
3.     asyncio.set_event_loop(loop)
   
4.     try:
   
5.         loop.run_forever()
   
6.     except Exception as e:
   
7.         print(f"事件循环异常: {e}")
   
8.     finally:
   
9.         loop.close()

复制代码

• 1. run_forever

  复制代码
  ：让事件循环持续运行，直到被外部停止。
  
• 异常处理：捕获可能的错误并打印，便于调试。
  
• 1. finally

  复制代码
  ：确保循环关闭时资源被正确释放。
  

关闭逻辑则由

1. _shutdown

复制代码

方法控制：

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

复制代码

当所有定时器都停止时（

1. _running_timers == 0

复制代码

），事件循环会被安全停止。

3. 定时器核心逻辑

每个

1. Timer

复制代码

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

1. def __init__(self):
   
2.     self.is_running = False
   
3.     self._stop_event = asyncio.Event()
   
4.     self._task = None
   
5. 
   
6. async def _timer_loop(self, interval, callback):
   
7.     try:
   
8.         while not self._stop_event.is_set():
   
9.             await asyncio.sleep(interval)
   
10.             if not self._stop_event.is_set():
    
11.                 await asyncio.get_event_loop().run_in_executor(None, callback)
    
12.     except asyncio.CancelledError:
    
13.         pass  # 正常取消时忽略
    
14.     finally:
    
15.         self.is_running = False
    
16.         Timer._running_timers -= 1
    
17.         Timer._shutdown()

复制代码

• 1. _stop_event

  复制代码
  ：一个
  1. asyncio.Event

  复制代码
  对象，用于控制定时器的停止。
  
• 1. _timer_loop

  复制代码
  ：异步协程，每隔
  1. interval

  复制代码
  秒执行一次回调函数
  1. callback

  复制代码
  。
  
• 1. run_in_executor

  复制代码
  ：将回调函数运行在默认的线程池中，避免阻塞事件循环。
  

4. 启动与停止

启动和停止方法是用户的主要接口：

1. def start(self, interval, callback):
   
2.     if not self.is_running:
   
3.         Timer._ensure_loop()
   
4.         self.is_running = True
   
5.         self._stop_event.clear()
   
6.         self._task = asyncio.run_coroutine_threadsafe(
   
7.             self._timer_loop(interval, callback),
   
8.             Timer._loop
   
9.         )
   
10.         Timer._running_timers += 1
    
11. 
    
12. def stop(self):
    
13.     if self.is_running:
    
14.         self._stop_event.set()
    
15.         if self._task:
    
16.             Timer._loop.call_soon_threadsafe(self._task.cancel)
    
17.         self.is_running = False

复制代码

• 1. start

  复制代码
  ：启动定时器，确保事件循环可用，并记录运行中的定时器数量。
  
• 1. stop

  复制代码
  ：通过设置
  1. _stop_event

  复制代码
  并取消任务来停止定时器。
  

5. 使用示例

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

1. def callback1():
   
2.     print(f"回调1触发: {time.strftime('%H:%M:%S')}")
   
3. 
   
4. def callback2():
   
5.     print(f"回调2触发: {time.strftime('%H:%M:%S')}")
   
6. 
   
7. timer1 = Timer()
   
8. timer2 = Timer()
   
9. 
   
10. timer1.start(2, callback1)  # 每2秒触发一次
    
11. timer2.start(3, callback2)  # 每3秒触发一次
    
12. 
    
13. time.sleep(10)  # 运行10秒
    
14. timer1.stop()
    
15. timer2.stop()

复制代码

输出可能如下：

1. 回调1触发: 14:30:02
   
2. 回调2触发: 14:30:03
   
3. 回调1触发: 14:30:04
   
4. 回调1触发: 14:30:06
   
5. 回调2触发: 14:30:06
   
6. ...

复制代码

设计亮点

• 异步与多线程结合：通过
  1. asyncio

  复制代码
  和
  1. threading

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

适用场景

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

总结

这个异步定时器实现结合了 Python 的异步编程和多线程特性，提供了一个轻量、灵活的解决方案。无论是简单的脚本还是复杂的后台服务，它都能胜任。如果你需要一个可靠的定时器，不妨试试这个实现，或者根据需求进一步优化它！
以上就是使用Python实现一个优雅的异步定时器的详细内容，更多关于Python异步定时器的资料请关注脚本之家其它相关文章！

来源：https://www.jb51.net/python/339974n5e.htm
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！