Python实现自定义超时异常的三种方法：线程、Signal与返回值封装
本文详解Python中实现自定义超时异常的三种实用方案。首先介绍基于threading结合join()阻塞时间的线程超时机制，其次讲解Linux环境下利用signal.alarm()触发SIGALRM信号的方法，最后提供封装Thread类以获取函数返回值并捕获异常的完整实现。内容涵盖代码原理、执行流程及注意事项，帮助开发者灵活处理第三方接口调用或耗时操作中的超时问题。
Python自定义超时异常背景与方案概述
在近期项目实践中，调用第三方接口时频繁遭遇请求超时或参数错误引发的代码异常。针对此类超时异常，查阅Python官方文档及相关资源后，并未发现现成的、支持按用户自定义时间抛出超时异常的模块。因此，决定自行实现一个自定义超时异常处理机制。目前总结并验证了两种主要实现路径：利用signal.alarm()机制以及通过threading模块实现超时控制。
基于threading与time模块的超时控制
方法一：基于threading与time模块的超时控制
其核心原理是将待执行的功能函数封装至子线程中运行。通过设定子线程的阻塞等待时间，若超时则主线程不再等待子线程执行完毕。此时主线程退出，作为守护线程的子线程也会随之终止。
实现的关键在于程序中调用join()方法，该方法用于等待线程结束。具体而言，在子线程完成运行之前，调用join()的父线程（主线程）会被阻塞。
# coding=utf-8
import threading
import time
def myFunc():
time.sleep(4)
print("myFunc执行了")
if __name__ == '__main__':
t = threading.Thread(target=myFunc)
t.setDaemon(True)
t.start()
t.join(2)
print("it's over")执行结果分析如下：
输出显示：it's over
从结果可以看出，当主线程执行至设定的2秒时间点后结束并退出。此时子线程尚未执行完毕（耗时4秒），因此被强制终止。
# coding=utf-8
import threading
import time
def myFunc():
time.sleep(1)
print("myFunc执行了")
if __name__ == '__main__':
t = threading.Thread(target=myFunc)
t.setDaemon(True)
t.start()
t.join(2)
print("it's over")
另一种显示结果情况如下：
输出显示： myFunc执行了 it's over
此结果表明，当子线程执行耗时为1秒时，未超过主线程设定的3秒等待时间。因此，子线程正常执行完毕，主线程也顺利执行后续代码，两者均正常结束。
利用signal.alarm()实现超时（Linux环境）
方法二：利用signal.alarm()实现，需注意两点限制：一是signal信号机制仅在Linux系统环境下有效；二是该信号处理必须作用于主线程中才生效。
import signal
import time
# Define signal handler function
def myHandler(signum, frame):
exit("TimeoutError")
def test_fun():
# time.sleep(3)
int("afsdf")
a = 2 + 3
return a
if __name__ == '__main__':
try:
signal.signal(signal.SIGALRM, myHandler)
signal.alarm(2)
test = test_fun()
print(test)
signal.alarm(0)
except Exception as ret:
print("msg:", ret)执行结果分析如下：
当test_fun内部包含time.sleep(3)时，会触发抛出TimeoutError异常；若test_fun内部出现如int("afsdf")这样的类型转换错误，则会抛出ValueError("invalid literal for int() with base 10: 'afsdf'")；若test_fun函数执行时间小于设定的2秒限制，则正常返回函数计算结果。
封装Thread类获取返回值与异常处理
方法三：带有返回值的超时异常处理，可通过自定义继承自threading.Thread的类来实现对执行结果的捕捉。
import threading
import sys
import time
class Dispacher(threading.Thread):
def __init__(self, fun, args):
threading.Thread.__init__(self)
self.setDaemon(True)
self.result = None
self.error = None
self.fun = fun
self.args = args
self.start()
def run(self):
try:
self.result = self.fun(self.args)
except:
self.error = sys.exc_info()
def test_fun(i):
# time.sleep(4)
a = i*i
# b
return a
def main_fun():
c = Dispacher(test_fun, 2)
c.join(2)
if c.isAlive():
return "TimeOutError"
elif c.error:
return c.error[1]
t = c.result
return t
if __name__ == '__main__':
fun = main_fun()
print(fun)执行结果分析如下：
当test_fun执行时间超过设定的2秒限制时，将抛出TimeOutError；若test_fun执行时间小于2秒且函数逻辑正常执行（如输入参数为2，返回4），则显示计算结果4；若test_fun内部出现如引用未定义变量“b”的情况，则会抛出global name 'b' is not defined的异常。
以上即为Python中实现自定义超时异常的几种主要方法及其应用场景，希望能为相关开发工作提供参考与帮助。
本站声明
本文内容仅供参考，不代表本站观点。转载或引用请注明出处。如发现内容问题，请通过联系我们页面反馈。
