Python子线程退出及线程退出控制的代码

晚来风

发布时间：2026-10-13 | 6人浏览过

原创

本文介绍Python中子线程退出的几种实现方式。首先通过信号处理解决Ctrl+C中断时子线程未退出的问题，随后探讨使用共享标志位的安全退出机制。最后详解利用ctypes模块强制抛出异常来终止线程的方法，包括相关代码实现及注意事项，帮助开发者掌握Python多线程退出控制技巧。

Python子线程退出问题示例

接下来通过具体代码示例，详细介绍Python中子线程退出的相关问题。


def thread_func():
  while True:
      #do something
      #do something
      #do something
t=threading.Thread(target = thread_func)
t.start()
# main thread do something
# main thread do something
# main thread do something

程序正常运行时无异常，但在按下Ctrl+C进行中断时会出现问题：主线程退出后，子线程仍在继续运行。

为此，在主线程中增加了信号处理逻辑，当接收到SIGINT信号时，修改子线程的循环条件以触发退出。


loop = True
def thread_func():
  while loop:
      #do something
      #do something
      #do something
t=threading.Thread(target = thread_func)
t.start()
# ctrl+c时，改变loop为False
def handler(signum, frame):
  global loop
  loop = False
  t.join()
  exit(0)
signal(SIGINT, handler)
# main thread do something
# main thread do something
# main thread do something

这样即可通过Ctrl+C正常退出，但这里存在一个疑问：主线程退出后，整个进程是否会自动终止？

Python线程退出控制机制

知识点扩展：Python线程退出控制机制

利用ctypes模块控制线程退出

Python标准库中的threading模块并未设计直接退出线程的机制，主要原因在于非正常退出可能导致不可预知的后果。

例如：

线程可能正持有某个必须正确释放的关键资源，如锁。

若强行终止线程，其创建的子线程也将被一并杀死。

管理自定义线程的最佳实践是引入请求退出标志，使每个线程能按一定时间间隔检查该标志，判断是否需要退出。

参考以下代码示例：


import threading
class StoppableThread(threading.Thread):
  """Thread class with a stop() method. The thread itself has to check
  regularly for the stopped() condition."""

  def __init__(self):
    super(StoppableThread, self).__init__()
    self._stop_event = threading.Event()

  def stop(self):
    self._stop_event.set()

  def stopped(self):
    return self._stop_event.is_set()

该代码中，线程应定期检查停止标志；退出时可调用stop()函数，并使用join()函数等待线程完全退出。

然而，有时确实需要强制终止线程，例如在封装外部库时，若该库进行长时间阻塞调用，可能需要中断它。

Python线程可以通过抛出异常来结束运行：

参数分别为线程ID号和退出标识。


def _async_raise(tid, exctype):
  '''Raises an exception in the threads with id tid'''
  if not inspect.isclass(exctype):
    raise TypeError("Only types can be raised (not instances)")
  res = ctypes.pythonapi.PyThreadState_SetAsyncExc(tid,
                         ctypes.py_object(exctype))
  if res == 0:
    raise ValueError("invalid thread id")
  elif res != 1:
    # "if it returns a number greater than one, you're in trouble,
    # and you should call it again with exc=NULL to revert the effect"
    ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, 0)
    raise SystemError("PyThreadState_SetAsyncExc failed")

如果线程在Python解释器外部运行，它将无法捕获中断，即抛出异常后无法对线程进行中断。

简化后的上述代码可应用于实际场景，例如当检测到线程运行时间超过可接受范围时进行中断。


def _async_raise(tid, exctype):
  """raises the exception, performs cleanup if needed"""
  tid = ctypes.c_long(tid)
  if not inspect.isclass(exctype):
    exctype = type(exctype)
  res = ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, ctypes.py_object(exctype))
  if res == 0:
    raise ValueError("invalid thread id")
  elif res != 1:
    # """if it returns a number greater than one, you're in trouble,
    # and you should call it again with exc=NULL to revert the effect"""
    ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, None)
    raise SystemError("PyThreadState_SetAsyncExc failed")
def stop_thread(thread):
  _async_raise(thread.ident, SystemExit)

总结

以上介绍了Python子线程退出及线程退出控制的相关代码，希望对大家有所帮助。如有疑问请留言，小编会及时回复。感谢大家对本站的支持！若本文对您有帮助，欢迎转载，请注明出处，谢谢！