本文主要是介绍python主线程捕获子线程异常，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
目录
问题抛出：子线程出现异常，主线程仍会正常进行吗
设计实现python主线程捕获子线程异常
需求
问题
解决思路
示例代码
问题抛出：子线程出现异常，主线程仍会正常进行吗
结论：子线程出现异常，主线程仍会正常进行
例子1
测试代码
import threadingdef func(x, m_file):y = 100/(x-5)m_file.write(str(x) + "\n")if __name__ == "__main__":file = open("./test.txt", "w")th_list = []for i in range(0, 10):th = threading.Thread(target=func, args=(i, file))th.start()th_list.append(th)for th in th_list:th.join()file.close()运行完后报错提示
D:\software\python_3.10.1_32\python.exe D:\hutao\projects\clear_market_data\test.py
Exception in thread Thread-6 (func):
Traceback (most recent call last):File "D:\software\python_3.10.1_32\lib\threading.py", line 1009, in _bootstrap_innerself.run()File "D:\software\python_3.10.1_32\lib\threading.py", line 946, in runself._target(*self._args, **self._kwargs)File "D:\hutao\projects\clear_market_data\test.py", line 6, in funcy = 100/(x-5)
ZeroDivisionError: division by zeroProcess finished with exit code 0可以看到只有线程6报错了，但是其它线程和主进程是正常进行的
再次查看test.txt内容
0
1
2
3
4
6
7
8
9可以看到，内容仅仅没有线程6的结果，说明子线程异常不会影响主线程
例子2
示例代码
import threadingdef func(x):y = 100/(x-5)data_list.append(x)if __name__ == "__main__":data_list = []th_list = []for i in range(10):th = threading.Thread(target=func, args=(i,))th.start()th_list.append(th)for th in th_list:th.join()print(data_list)运行结果
D:\Python3.6.6\python.exe D:\hutao\projects\grid_trade\test.py
Exception in thread Thread-6:
Traceback (most recent call last):File "D:\Python3.6.6\lib\threading.py", line 916, in _bootstrap_innerself.run()File "D:\Python3.6.6\lib\threading.py", line 864, in runself._target(*self._args, **self._kwargs)File "D:\hutao\projects\grid_trade\test.py", line 5, in funcy = 100/(x-5)
ZeroDivisionError: division by zero[0, 1, 2, 3, 4, 6, 7, 8, 9]Process finished with exit code 0设计实现python主线程捕获子线程异常
需求
python多线程时，子线程出现异常，主线程依然正常向下执行，显然不符合工程代码，为解决子线程异常问题，提出研究内容：捕获子线程异常。
捕获子线程异常的好处是：
- 程序在在多线程处理多个任务时，能够知道每个任务是否正常执行完毕，能够知道是哪个任务出了问题。不然即使子线程出了问题，但整个程序完毕，都不知道出了问题。
- 在出了问题后，需要快速定位到是哪个地方出了问题。
问题
直接看代码：
def thread_text(i):time.sleep(i)raise Exception(u'error')def main():t = threading.Thread(target=thread_text, args=(1,))t.start()t.join()print(u'end')if __name__ =='__main__':main()这是一个简单的测试，在子线程中返回一个异常，输出如下：
endException in thread Thread-1:
Traceback (most recent call last):File "/usr/lib/python3.6/threading.py", line 916, in _bootstrap_innerself.run()File "/usr/lib/python3.6/threading.py", line 864, in runself._target(*self._args, **self._kwargs)File "/usercode/file.py", line 8, in thread_textraise Exception(u'error')
Exception: error可以看到，虽然子线程报出了异常，但main方法依然正常打印出了‘end’，这显然不是我们预期的，接下来，我们开始解决
解决思路
这里用的是queue来实现，它是python3的内置模块，创建堆栈队列，用来处理多线程通信。
queue用法参考：python内置模块之queue（队列）用法-CSDN博客
思路为：当子线程异常，则向queue添加一条数据，可以是任何数据，main方法检测queue是否为空，如果是空的，说明子线程无异常，反之则异常。然后通过检测活跃的线程数来确定子线程是否执行完毕。
更进一步的，添加的那条数据最好是携带更多的信息，比如具体是哪个线程，具体是什么异常信息，以及其它自定义的信息。比如，对于单个任务而言，可能会经过很多步骤，我们需要明确是在哪一步报了错，而且更重要的是如果该任务整体用时非常长，对于报错的任务，是不需要从头开始重新执行的。因此可以设计两个队列变量，一个用来记录错误信息，A，一个用来记录运行过程，B，比如完成一个任务需要10步，每成功运行一步就往B添加一条信息，在某一步报错时就往A添加一条信息。通过查看A和B内的数据，就可以明确是在哪一步报了错，然后检查代码bug等，以及对报错了的线程以及根据A和B中记录的信息来单独处理报错，而不是重新运行那个执行多线程任务的程序。
示例代码
检测活跃的线程数代码
threading.enumerate() # list 当前活跃的线程，包含主线程
threading.active_count() # int 当前活跃线程数，包含主线程输出如下
[<_MainThread(MainThread, started 8628338176)>]
1好了，思路及用到的模块都介绍完毕，直接上代码
import queue
import threading
import sys
import timedef thread_text(test_queue, i):try:time.sleep(i)raise Exception(u'test_error')except:error_info = sys.exc_info()# error_info = "......"test_queue.put(error_info)if __name__ == '__main__':q = queue.Queue()t = threading.Thread(target=thread_text, args=(q, 1))t.start()while True:if q.empty():if threading.active_count() == 0:breakelse:x = q.get()print(f"({x})") raise Exception(u'main error')print(u'end')输出
(<class 'Exception'>, Exception('test_error',), <traceback object at 0x00000256790AA8C8>)
Traceback (most recent call last):File "D:\BaiduSyncdisk\py_project\zs_market\codeList\test.py", line 26, in <module>raise Exception(u'main error')
Exception: main errorProcess finished with exit code 1可以看到，并没有输出‘end’, 而是返回‘main error’, 说明主线程成功捕获到子线程，这里我是在子线程里queue.put了sys模块的exc_info()，能够更方便查错改错，当然可以是自定义的其它信息，根据需求来。
end
这篇关于python主线程捕获子线程异常的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
