Python中安全终止死循环线程的解决方案

Cora

发布时间：2026-10-14| 5人浏览过

原创

本文探讨在Python中如何安全停止运行中的死循环线程。针对join()方法在死循环场景下失效的问题，提出通过引入计时线程和状态标志位来控制线程执行时长的方案。文章详细解析了从创建线程到修改执行逻辑的完整代码实现，帮助开发者解决线程无法正常退出的难题。

多线程基础：串行与并行

进程执行任务必须依赖线程。线程是进程中最小的执行单位，且一个进程至少包含一个线程。

理解多线程需要区分串行与并行两个概念，这有助于深入掌握多线程机制。

串行是指单条线程按顺序执行多个任务。以下载文件为例，串行模式下必须等待文件A下载完成后，才能开始下载文件B，两者在时间上不会重叠。

并行指开启多条线程同时下载多个文件，这是严格意义上的同一时刻发生，时间上存在重叠。

问题描述：join()方法的局限性

问题描述：

在近期项目开发中，遇到一个死循环线程，需要在特定时刻直接停止该线程。尝试使用基本的join()方法无效。查阅百度及国外官方文档后发现，出于保护进程的考虑，直接强行停止死循环线程较为困难。

解决方案：通过状态标志控制线程

整体解决方案：

创建线程函数


class myThread(threading.Thread):
  def __init__(self, threadID, name, counter):
    threading.Thread.__init__(self)
    self.threadID = threadID
    self.name = name
    self.counter = counter

  def run(self):
    print("开启线程： " + self.name)
    # 获取锁，用于线程同步
    threadLock.acquire()
    # 执行要开启的线程

    # 释放锁，开启下一个线程
    threadLock.release()

死循环函数实现

死循环函数


def Dead_circle():
  while True:
    print("死循环")

增加结束条件

增加结束的条件


def run_go():
  num = 0
  while True:
    time.sleep(1)
    num += 1
    if num == 10:
      return False # 假设我执行10s后结束

增加新方法


def run_go():
  num = 0
  while True:
    time.sleep(1)
    num += 1
    if num == 10:
      return False

修改执行方法


def Dead_circle():
  thread1 = myThread(1, "Thread-1", 1)
  thread1.start()
  while thread1.counter:
    time.sleep(1)
    print("死循环")
  thread1.join()

修改线程类


class myThread(threading.Thread):
  def __init__(self, threadID, name, counter):
    threading.Thread.__init__(self)
    self.threadID = threadID
    self.name = name
    self.counter = counter
    self.State = True

  def run(self):
    print("开启线程： " + self.name)
    # 获取锁，用于线程同步
    threadLock.acquire()
    # 执行要开启的线程
    if not run_go():
      self.counter = False
    # 释放锁，开启下一个线程
    threadLock.release()

整体代码示例

整体代码


import threading
import time
from tkinter import *

threadLock = threading.Lock()


class myThread(threading.Thread):
  def __init__(self, threadID, name, counter):
    threading.Thread.__init__(self)
    self.threadID = threadID
    self.name = name
    self.counter = counter
    self.State = True

  def run(self):
    print("开启线程： " + self.name)
    # 获取锁，用于线程同步
    threadLock.acquire()
    # 执行要开启的线程
    if not run_go():
      self.counter = False
    # 释放锁，开启下一个线程
    threadLock.release()


def run_go():
  num = 0
  while True:
    time.sleep(1)
    num += 1
    if num == 10:
      return False


def Dead_circle():
  thread1 = myThread(1, "Thread-1", 1)
  thread1.start()
  while thread1.counter:
    time.sleep(1)
    print("死循环")
  thread1.join()


if __name__ == '__main__':
  Dead_circle()