標準庫 threading.Timer 僅支援**單次**延遲執行。實務上常需「每 N 秒執行一次」的週期性定時器。本文展示如何封裝 Timer 實現可啟動、停止、動態調整間隔的 PeriodicTimer 類別，並處理執行緒安全、例外捕獲、優雅關閉等細節。
import threading
import time
from typing import Callable, Optional
class PeriodicTimer:
"""週期性定時器：每 interval 秒執行一次 callback，支援動態調整與優雅停止"""
def __init__(self, interval: float, callback: Callable[[], None],
name: Optional[str] = None, daemon: bool = True):
self._interval = interval
self._callback = callback
self._name = name or f"PeriodicTimer-{id(self)}"
self._daemon = daemon
self._timer: Optional[threading.Timer] = None
self._running = False
self._lock = threading.RLock() # 可重入鎖，保護啟動/停止競態
@property
def interval(self) -> float:
with self._lock:
return self._interval
@interval.setter
def interval(self, value: float):
with self._lock:
if value <= 0:
raise ValueError("interval 必須 > 0")
self._interval = value
# 若正在運行，重新排程以套用新間隔
if self._running:
self._reschedule()
def _run(self):
"""實際執行 callback，完成後若仍在運行則排程下一次"""
try:
self._callback()
except Exception as e:
# 避免 callback 例外導致定時器靜默死亡
import logging
logging.exception(f"[{self._name}] callback 執行失敗: {e}")
finally:
with self._lock:
if self._running:
self._reschedule()
def _reschedule(self):
"""建立新 Timer 排程下一次執行"""
self._timer = threading.Timer(self._interval, self._run)
self._timer.daemon = self._daemon
self._timer.name = self._name
self._timer.start()
def start(self):
with self._lock:
if self._running:
return
self._running = True
self._reschedule()
def stop(self, timeout: Optional[float] = None):
"""停止定時器，等待當前 callback 完成 (可選 timeout)"""
with self._lock:
if not self._running:
return
self._running = False
if self._timer:
self._timer.cancel()
self._timer = None
# 在鎖外等待，避免死鎖 (callback 可能需取得同一鎖)
if timeout is not None:
# 簡易等待：實際專案建議用 threading.Event 通知
time.sleep(min(timeout, self._interval + 0.1))
def __enter__(self):
self.start()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.stop()
return False
def __repr__(self):
return f""
def heartbeat():
print(f"[{time.strftime('%H:%M:%S')}] heartbeat")
timer = PeriodicTimer(2.0, heartbeat, name="Heartbeat")
timer.start()
# 主執行緒做其他事
time.sleep(10)
timer.stop()
print("停止完成")
timer = PeriodicTimer(1.0, lambda: print("tick"))
timer.start()
time.sleep(3)
timer.interval = 0.5 # 加快頻率
time.sleep(2)
timer.interval = 5.0 # 減慢頻率
time.sleep(6)
timer.stop()
with PeriodicTimer(1.0, lambda: print("auto")) as t:
time.sleep(3)
# 離開 with 區塊自動 stop()
def worker(task_id: int, msg: str):
print(f"Task {task_id}: {msg}")
# 使用 functools.partial 或 lambda 綁定參數
from functools import partial
timer = PeriodicTimer(2.0, partial(worker, 42, "running"))
timer.start()
| 細節 | 處理方式 | 原因 |
|---|---|---|
| callback 例外 | try/except + logging | 防止例外中斷定時器迴圈，導致後續不再執行 |
| 啟動/停止競態 | RLock 保護 _running, _timer | 避免同時呼叫 start/stop 導致狀態不一致 |
| 間隔動態調整 | setter 中呼叫 _reschedule() | 立即生效，無需重啟 |
| 優雅停止 | cancel() + 等待 callback 完成 | 確保資源釋放、狀態一致 |
| daemon 執行緒 | 預設 True | 主程式結束時自動退出，不阻塞進程 |
| 命名 | Timer.name = self._name | 除錯時在 threading.enumerate() 易識別 |
上述實現為**固定延遲**：上一次結束後等待 interval 再啟動下一次。若需**固定頻率**（嚴格每 N 秒啟動，不累積誤差）：
class FixedRateTimer(PeriodicTimer):
def _reschedule(self):
# 計算下一個理論觸發時間
if not hasattr(self, '_next_time'):
self._next_time = time.monotonic() + self._interval
else:
self._next_time += self._interval
delay = max(0, self._next_time - time.monotonic())
self._timer = threading.Timer(delay, self._run)
self._timer.daemon = self._daemon
self._timer.start()
| 方案 | 優點 | 缺點 | 適用場景 |
|---|---|---|---|
| threading.Timer 封裝 | 標準庫、無依賴、輕量 | 精度受 GIL/排程影響、無持久化 | 簡單背景任務、健康檢查、輪詢 |
| sched.scheduler | 同執行緒、可優先級、可取消特定事件 | 阻塞主執行緒、需自行跑 loop | 單執行緒事件模擬、測試 |
| APScheduler | 持久化、叢集、cron、管理介面 | 第三方依賴、較重 | 生產級排程服務 |
| asyncio + create_task | 高併發、精確控制、現代語法 | 需 async/await 架構 | 新專案、高效能服務 |
Timer.cancel() **無法中斷正在執行的 callback**；若 callback 耗時長，stop() 返回後 callback 仍在跑。input() 或 time.sleep() 阻塞，daemon=True 確保進程可退出。threading.Timer 精度約 15ms；高精度需求請用 time.sleep 迴圈或 asyncio。這個 PeriodicTimer 類別輕量、無依賴、API 直觀，適合在無法引入 APScheduler 的環境中快速實現週期性背景任務。
