Python多執行緒捕獲子執行緒的異常,並退出主程序。
阿新 • • 發佈:2021-12-24
自己在專案的開發中,一般能避免在單個程序中使用多執行緒就儘量把每個執行緒包裝成獨立的程序執行,通過socket或者一些中介軟體比如redis進行通訊,工作,協調。
但有時候必須涉及到多執行緒操作,而且碰到的情況中,多個執行緒必須協調全部正常工作才能執行邏輯,但子執行緒有著自己的棧區,報錯了並不影響其它的執行緒,導致整個程序無法退出。
我當時想到的有兩種思路,一種是多個執行緒間進行通訊或者一個全域性變數的標記,當報錯的時候,就修改這個標記,所有的子執行緒定時去查詢這個標記,但感覺這個思路的拓展性太差,而且每個子執行緒需要主動定期查詢或者通訊,太麻煩了。
後面一種就是我準備上程式碼的思路, 將所有的子執行緒設計成守護執行緒,主執行緒迴圈查詢子執行緒的狀態值,當發現任意的子執行緒狀態異常,獲取該子執行緒的異常物件,並上浮,退出主執行緒,導致所有的子執行緒退出。
import threading, traceback import time class ExcThread(threading.Thread): def __init__(self, call_obj, *args, **kwargs): super(ExcThread, self).__init__(*args, **kwargs) self.callable_obj = call_obj # 自己設定的退出狀態值 self.exit_code = 0 self.exception = None self.exc_traceback = '' # 主動設定為守護執行緒,必須條件 self.setDaemon(True) def run(self): try: self._run() except Exception as e: self.exit_code = 1 # 儲存異常物件儲存在例項物件中 self.exception = e self.exc_traceback = traceback.format_exc() def _run(self): try: self.callable_obj(*self._args, **self._kwargs) except Exception as e: raise e def t_func(name, age=18): while 1: print(name, age) time.sleep(3) if age == 1: raise Exception('hee') # 生成一份子執行緒列表物件,用於主執行緒輪詢檢查使用 def start_child_thread(): thread_task_list = [] for i in range(3): f = ExcThread(call_obj=t_func, args=('sidian',), kwargs={'age': i}) f.start() thread_task_list.append(f) return thread_task_list def check_thread(): t_list = start_child_thread() while 1: for task in t_list: if not task.is_alive(): raise task.exception time.sleep(1) if __name__ == '__main__': check_thread()