在開發 DevOps Bot 時,其實之中有存在一個問題,那就是我的bot整體是採用非同步的設計框架,但為了達到定時排程的功能,需要用到schedule 庫,這問題的根源在於 schedule 是一個優秀的同步定時任務庫,但它使用time.sleep() 會阻塞執行緒,這就跟我的程式設計需要持續運行事件循環產生了衝突,因此如果直接在非同步環境中使用 schedule,整個 Bot 都會被阻塞
而解決方案就是-建立跨執行緒事件橋樑
# 關鍵設計:非同步事件與同步執行緒的橋接
weekly_check_event = asyncio.Event()
def run_scheduler():
"""在獨立執行緒中運行的同步排程器"""
print("⏰ 排程器設定完成:每週一 01:00 UTC (09:00 UTC+8) 自動檢查")
while True:
# 同步世界:使用傳統的排程庫
schedule.run_pending()
time.sleep(60) # 阻塞式等待,但不影響主執行緒
def trigger_weekly_check():
"""從同步世界向非同步世界發送信號"""
print("🔔 排程器觸發每周檢查")
weekly_check_event.set() # 這行代碼是魔法所在!
@tasks.loop(seconds=30)
async def check_scheduled_events():
"""在非同步世界中監聽同步世界的信號"""
if weekly_check_event.is_set():
weekly_check_event.clear() # 重置信號
await execute_scheduled_check() # 執行非同步任務
這樣設計的好處在於:
架構圖可以看出當到了排程表定時間時,呼叫 weekly_check_event.set(),將事件狀態設為 True,表示觸發檢查(此為同步階段),而Bot 主執行緒會每 30 秒檢查一次 weekly_check_event 是否被觸發,當檢測到觸發後會向 GitHub API 發送非同步請求,取得 PR 資料接著根據資料生成報告並發送