解决python ThreadPoolExecutor 线程池中的异常捕获问题

更新时间：2020年04月08日 12:00:07 作者：Mleeeeeee

这篇文章主要介绍了解决python ThreadPoolExecutor 线程池中的异常捕获问题，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧

问题

最近写了涉及线程池及线程的 python 脚本，运行过程中发现一个有趣的现象，线程池中的工作线程出现问题，引发了异常，但是主线程没有捕获异常，还在发现 BUG 之前一度以为线程池代码正常返回。

先说重点

这里主要想介绍 python concurrent.futuresthread.ThreadPoolExecutor 线程池中的 worker 引发异常的时候，并不会直接向上抛起异常，而是需要主线程通过调用concurrent.futures.Future.exception(timeout=None) 方法主动获取 worker 的异常。

问题重现及解决

引子

问题主要由这样一段代码引起的：

def thread_executor():
 logger.info("I am slave. I am working. I am going to sleep 3s")
 sleep(3)
 logger.info("Exit thread executor")


def main():
 thread_obj = threading.Thread(target=thread_executor)
 while True:
  logger.info("Master starts thread worker")

  try:
   # 工作线程由于某种异常而结束并退出了，想重启工作线程的工作，但又不想重复创建线程
   thread_obj.start() # 这一行会报错，同一线程不能重复启动
  except Exception as e:
   logger.error("Master start thread error", exc_info=True)
   raise e

  logger.info("Master is going to sleep 5s")
  sleep(5)

上面这段代码的功能如注释中解释的，主要要实现类似生产者消费者的功能，工作线程一直去生产资源，主线程去消费工作线程生产的资源。但是工作线程由于异常推出了，想重新启动生产工作。显然，这个代码会报错。

运行结果:

thread: MainThread [INFO] Master starts thread worker
thread: Thread-1 [INFO] I am slave. I am working. I am going to sleep 3s
thread: MainThread [INFO] Master is going to sleep 5s
thread: Thread-1 [INFO] Exit thread executor because of some exception
thread: MainThread [INFO] Master starts thread worker
thread: MainThread [ERROR] Master start thread error
Traceback (most recent call last):
File "xxx.py", line 47, in main
 thread_obj.start()
File "E:\anaconda\lib\threading.py", line 843, in start
 raise RuntimeError("threads can only be started once")
RuntimeError: threads can only be started once
Traceback (most recent call last):
File "xxx.py", line 56, in <module>
 main()
File "xxx.py", line 50, in main
 raise e
File "xxx.py", line 47, in main
 thread_obj.start()
File "E:\anaconda\lib\threading.py", line 843, in start
 raise RuntimeError("threads can only be started once")
RuntimeError: threads can only be started once

切入正题

然而脚本还有其他业务代码要运行，所以需要把上面的资源生产和消费的代码放到一个线程里完成，所以引入线程池来执行这段代码：

def thread_executor():
 while True:
  logger.info("I am slave. I am working. I am going to sleep 3s")
  sleep(3)
  logger.info("Exit thread executor because of some exception")
  break


def main():
 thread_obj = threading.Thread(target=thread_executor)
 while True:
  logger.info("Master starts thread worker")

  # 工作线程由于某种异常而结束并退出了，想重启工作线程的工作，但又不想重复创建线程
  # 没有想到这里会有异常
  thread_obj.start() # 这一行会报错，同一线程不能重复启动

  logger.info("Master is going to sleep 5s")
  sleep(5)


def thread_pool_main():
 thread_obj = ThreadPoolExecutor(max_workers=1, thread_name_prefix="WorkExecutor")
 logger.info("Master ThreadPool Executor starts thread worker")
 thread_obj.submit(main)

 while True:
  logger.info("Master ThreadPool Executor is going to sleep 5s")
  sleep(5)

if __name__ == '__main__':
 thread_pool_main()

代码运行结果如下:

INFO [thread: MainThread] Master ThreadPool Executor starts thread worker
INFO [thread: WorkExecutor_0] Master starts thread worker
INFO [thread: MainThread] Master ThreadPool Executor is going to sleep 5s
INFO [thread: Thread-1] I am slave. I am working. I am going to sleep 3s
INFO [thread: WorkExecutor_0] Master is going to sleep 5s
INFO [thread: Thread-1] Exit thread executor because of some exception
INFO [thread: MainThread] Master ThreadPool Executor is going to sleep 5s
INFO [thread: WorkExecutor_0] Master starts thread worker
INFO [thread: MainThread] Master ThreadPool Executor is going to sleep 5s
INFO [thread: MainThread] Master ThreadPool Executor is going to sleep 5s
INFO [thread: MainThread] Master ThreadPool Executor is going to sleep 5s
INFO [thread: MainThread] Master ThreadPool Executor is going to sleep 5s
INFO [thread: MainThread] Master ThreadPool Executor is going to sleep 5s
INFO [thread: MainThread] Master ThreadPool Executor is going to sleep 5s
INFO [thread: MainThread] Master ThreadPool Executor is going to sleep 5s

... ...

显然，由上面的结果，在线程池 worker 执行到 INFO [thread: WorkExecutor_0] Master starts thread worker 的时候，是会有异常产生的，但是整个代码并没有抛弃任何异常。

解决方法

发现上面的 bug 后，想在线程池 worker 出错的时候，把异常记录到日志。查阅资料，要获取线程池的异常信息，需要调用 concurrent.futures.Future.exception(timeout=None) 方法，为了记录日志，这里加了线程池执行结束的回调函数。同时，日志中记录异常信息，用了 logging.exception() 方法。

def thread_executor():
 while True:
  logger.info("I am slave. I am working. I am going to sleep 3s")
  sleep(3)
  logger.info("Exit thread executor because of some exception")
  break


def main():
 thread_obj = threading.Thread(target=thread_executor)
 while True:
  logger.info("Master starts thread worker")

  # 工作线程由于某种异常而结束并退出了，想重启工作线程的工作，但又不想重复创建线程
  # 没有想到这里会有异常
  thread_obj.start() # 这一行会报错，同一线程不能重复启动

  logger.info("Master is going to sleep 5s")
  sleep(5)


def thread_pool_callback(worker):
 logger.info("called thread pool executor callback function")
 worker_exception = worker.exception()
 if worker_exception:
  logger.exception("Worker return exception: {}".format(worker_exception))


def thread_pool_main():
 thread_obj = ThreadPoolExecutor(max_workers=1, thread_name_prefix="WorkExecutor")
 logger.info("Master ThreadPool Executor starts thread worker")
 thread_pool_exc = thread_obj.submit(main)
 thread_pool_exc.add_done_callback(thread_pool_callback)
 # logger.info("thread pool exception: {}".format(thread_pool_exc.exception()))

 while True:
  logger.info("Master ThreadPool Executor is going to sleep 5s")
  sleep(5)


if __name__ == '__main__':
 thread_pool_main()

代码运行结果:

INFO [thread: MainThread] Master ThreadPool Executor starts thread worker
INFO [thread: WorkExecutor_0] Master starts thread worker
INFO [thread: MainThread] Master ThreadPool Executor is going to sleep 5s
INFO [thread: Thread-1] I am slave. I am working. I am going to sleep 3s
INFO [thread: WorkExecutor_0] Master is going to sleep 5s
INFO [thread: Thread-1] Exit thread executor because of some exception
INFO [thread: MainThread] Master ThreadPool Executor is going to sleep 5s
INFO [thread: WorkExecutor_0] Master starts thread worker
INFO [thread: WorkExecutor_0] called thread pool executor callback function
ERROR [thread: WorkExecutor_0] Worker return exception: threads can only be started once
Traceback (most recent call last):
File "E:\anaconda\lib\concurrent\futures\thread.py", line 57, in run
 result = self.fn(*self.args, **self.kwargs)
File "xxxx.py", line 46, in main
 thread_obj.start() # 这一行会报错，同一线程不能重复启动
File "E:\anaconda\lib\threading.py", line 843, in start
 raise RuntimeError("threads can only be started once")
RuntimeError: threads can only be started once
INFO [thread: MainThread] Master ThreadPool Executor is going to sleep 5s
INFO [thread: MainThread] Master ThreadPool Executor is going to sleep 5s
INFO [thread: MainThread] Master ThreadPool Executor is going to sleep 5s
... ...

最终的写法

其实，上面写法中，想重复利用一个线程去实现生产者线程的实现方法是有问题的，在此处，一般情况下，线程执行结束后，线程资源会被会被操作系统，所以线程不能被重复调用 start() 。

一种可行的实现方式就是，用线程池替代。当然，这样做得注意上面提到的线程池执行体的异常捕获问题。

def thread_executor():
 while True:
  logger.info("I am slave. I am working. I am going to sleep 3s")
  sleep(3)
  logger.info("Exit thread executor because of some exception")
  break

def executor_callback(worker):
 logger.info("called worker callback function")
 worker_exception = worker.exception()
 if worker_exception:
  logger.exception("Worker return exception: {}".format(worker_exception))
  # raise worker_exception


def main():
 slave_thread_pool = ThreadPoolExecutor(max_workers=1, thread_name_prefix="SlaveExecutor")
 restart_flag = False
 while True:
  logger.info("Master starts thread worker")

  if not restart_flag:
   restart_flag = not restart_flag
   logger.info("Restart Slave work")
  slave_thread_pool.submit(thread_executor).add_done_callback(executor_callback)

  logger.info("Master is going to sleep 5s")
  sleep(5)

总结

这个问题主要还是因为对 Python 的 concurrent.futuresthread.ThreadPoolExecutor 不够了解导致的，接触这个包是在书本上，但是书本没完全介绍包的全部 API 及用法，所以代码产生异常情况后，DEBUG 了许久在真正找到问题所在。查阅 python docs 后才对其完整用法有所认识，所以，以后学习新的 python 包的时候还是可以查一查官方文档的。

参考资料

英文版: docs of python concurrent.futures

中文版: python docs concurrent.futures — 启动并行任务

exception(timeout=None)

返回由调用引发的异常。如果调用还没完成那么这个方法将等待 timeout 秒。如果在 timeout 秒内没有执行完成，concurrent.futures.TimeoutError 将会被触发。timeout 可以是整数或浮点数。如果 timeout 没有指定或为 None，那么等待时间就没有限制。

如果 futrue 在完成前被取消则 CancelledError 将被触发。

如果调用正常完成那么返回 None。

add_done_callback(fn)

附加可调用 fn 到期程。当期程被取消或完成运行时，将会调用 fn，而这个期程将作为它唯一的参数。

加入的可调用对象总被属于添加它们的进程中的线程按加入的顺序调用。如果可调用对象引发一个 Exception 子类，它会被记录下来并被忽略掉。如果可调用对象引发一个 BaseException 子类，这个行为没有定义。

如果期程已经完成或已取消，fn 会被立即调用。

以上这篇解决python ThreadPoolExecutor 线程池中的异常捕获问题就是小编分享给大家的全部内容了，希望能给大家一个参考，也希望大家多多支持脚本之家。

您可能感兴趣的文章:

mvc框架打造笔记之wsgi协议的优缺点以及接口实现
这篇文章主要给大家介绍了关于mvc框架打造笔记之wsgi协议的优缺点以及接口实现的相关资料，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2018-08-08
Python xlwt工具使用详解,生成excel栏位宽度可自适应内容长度
这篇文章主要介绍了Python xlwt工具使用详解,生成excel栏位宽度可自适应内容长度，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2022-02-02
python3实现raspberry pi（树莓派）4驱小车控制程序
这篇文章主要为大家详细介绍了python3实现raspberry pi（树莓派）4驱小车控制程序，文中示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
2020-02-02
python提取特定格式的数据的操作方法
这篇文章主要介绍了python提取特定格式的数据,通过本脚本,我们学习了如何使用 Pandas 读取和处理 Excel 数据,如何使用 Tkinter 创建图形用户界面,以及如何处理文件对话框和文件操作,需要的朋友可以参考下
2024-06-06
快速下载VScode并配置Python运行环境(图文教程)
本文主要介绍了快速下载VScode并配置Python运行环境，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2023-05-05
关于pandas的离散化,面元划分详解
今天小编就为大家分享一篇关于pandas的离散化,面元划分详解，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2019-11-11
Python常见报错解决方案总结(新手拯救指南)
我们再使用python难免会出现各种各样的报错,下面这篇文章主要给大家介绍了关于Python常见报错解决方案的相关资料,文中通过示例代码介绍的非常详细,需要的朋友可以参考下
2022-05-05
pytorch之pytorch hook和关于pytorch backward过程问题
这篇文章主要介绍了pytorch之pytorch hook和关于pytorch backward过程问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
2023-09-09
Pandas数据处理库画图与文件读取使用示例
这篇文章主要为大家介绍了Pandas数据处理库画图与文件读取使用示例，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪
2022-10-10
Python 运算符Inplace 与Standard
这篇文章主要介绍了Python 运算符Inplace 与Standard，nplace运算符的行为类似于普通运算符，只是它们在可变和不可变目标的情况下以不同的方式运行
2022-09-09

解决python ThreadPoolExecutor 线程池中的异常捕获问题

相关文章

最新评论

大家感兴趣的内容

最近更新的内容

常用在线小工具