Python多线程超时控制方法

推荐使用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在超时后写入标记，主线程根据结果判断。

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=)，代码清晰且跨平台兼容。若需精细控制，可结合事件标志实现协作式中断，避免强行杀线程带来的风险。