推荐使用concurrent.futures实现线程超时控制，其future.result(timeout=)方法可安全处理超时；通过ThreadPoolExecutor提交任务并捕获TimeoutError异常，避免强制终止线程导致的资源泄漏，代码简洁且跨平台兼容。
在Python中使用多线程时，经常会遇到需要对某个任务设置超时时间的场景。比如调用外部API、执行耗时计算或等待资源响应。如果任务长时间未完成，我们希望主动中断它，避免程序卡死。虽然Python原生的threading模块不直接支持线程超时控制（无法强制终止线程），但可以通过一些技巧实现安全的超时处理。
1. 使用 threading.Timer 实现超时监控
一种简单方式是利用Timer对象，在指定时间后触发超时逻辑，配合标志位判断任务是否完成。
import threading
import time
def long_task(result):
time.sleep(5) # 模拟耗时操作
result['done'] = True
result['value'] = '任务完成'
def task_with_timeout(timeout=3):
result = {}
thread = threading.Thread(target=long_task, args=(result,))
timer = threading.Timer(timeout, lambda: result.setdefault('timeout', True))
thread.start()
timer.start()
thread.join(timeout)
timer.cancel() # 若任务提前完成，取消定时器
if 'timeout' in result:
print("任务超时")
elif result.get('done'):
print(result['value'])
else:
print("未知状态")
task_with_timeout()
这种方法通过共享字典result传递状态，Timer在超时后写入标记，主线程根据结果判断。
立即学习“Python免费学习笔记（深入）”；
2. 利用 concurrent.futures 控制线程执行时间
更推荐的方式是使用concurrent.futures.ThreadPoolExecutor，其future.result(timeout=)方法天然支持超时抛出异常。
from concurrent.futures import ThreadPoolExecutor
import time
def slow_function():
time.sleep(4)
return "任务成功"
def run_with_timeout():
with ThreadPoolExecutor(max_workers=1) as executor:
future = executor.submit(slow_function)
try:
result = future.result(timeout=3)
print(result)
except TimeoutError:
print("任务执行超时")
run_with_timeout()
这种方式简洁且安全，不需要手动管理线程生命周期，适合大多数场景。
3. 避免强行终止线程的陷阱
Python的线程不能被外部强制停止，调用thread.stop()等方法会破坏解释器状态，导致资源泄漏或数据不一致。正确的做法是：
- 任务内部定期检查退出标志
- 使用可中断的阻塞调用（如带timeout的queue.get）
- 让任务自行退出，而非暴力终止
def cancellable_task(stop_event):
for i in range(100):
if stop_event.is_set():
print("任务被取消")
return
time.sleep(0.1) # 模拟工作
print("任务完成")
主线程可在超时后调用stop_event.set()通知子线程退出。
4. 结合信号（Signal）用于单线程超时（仅限Unix）
在主线程中可以使用signal.alarm()实现函数级超时，但仅适用于单线程且不可用于线程中。
import signal
class TimeoutError(Exception): pass
def timeout_handler(signum, frame):
raise TimeoutError("函数执行超时")
def run_with_signal_timeout():
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(3)
try:
slow_function()
except TimeoutError as e:
print(e)
finally:
signal.alarm(0)
注意：此方法仅适用于CPython主线程，且Windows不支持SIGALRM。
基本上就这些。对于多数多线程超时需求，推荐优先使用concurrent.futures配合result(timeout=)，代码清晰且跨平台兼容。若需精细控制，可结合事件标志实现协作式中断，避免强行杀线程带来的风险。
