Python并行编程：线程进程关闭后主进程仍运行的问题求助

问题现象 #

作为Python并行编程初学者，我遇到线程和进程关闭后代码仍在运行的问题。用以下代码检查存活线程/进程：

print(f"thread : {threading.enumerate()}")
print(f"process : {multiprocessing.active_children()}")

返回结果：

thread : [<_MainThread(MainThread, started 15848)>, <Thread(QueueFeederThread, started daemon 5336)>]
process : []

不知道如何关闭这两个线程，以下是我的代码：

import threading
import multiprocessing
import queue
import torch
import cv2
import time


class read_video(threading.Thread):
    def __init__(self, cv, lock, sync_lock, display_frame_queue, predict_frame_queue):
        super().__init__(name="SHOW_VIDEO")
        self.daemon = True
        self.cv = cv2.VideoCapture(cv)
        self.display_frame_queue = display_frame_queue
        self.lock = lock
        self.sync_lock = sync_lock
        self.predict_frame_queue = predict_frame_queue

    def read(self):

        self.sync_lock.wait()
        while True:
            if self.display_frame_queue.qsize() <= 3:
                ret, frame = self.cv.read()
                if ret:
                    frame = cv2.resize(frame, (1280, 720))
                    self.display_frame_queue.put(frame)
                    self.predict_frame_queue.put(frame)
                else:
                    break
            if self.lock.is_set():
                self.cv.release()
                break

        return

    def run(self):
        self.read()


class display_video(threading.Thread):
    def __init__(self, lock, sync_lock, display_frame_queue, predict_result_queue):
        super().__init__(name="READ_VIDEO")
        self.lock = lock
        self.daemon = True
        self.sync_lock = sync_lock
        self.display_frame_queue = display_frame_queue
        self.predict_result_queue = predict_result_queue

    def show(self):

        color = [(0, 255, 0), (255, 255, 0), (0, 255, 255), (0, 140, 255)]
        class_name = ['Car', 'Motorcycle', 'Person', 'Truck']
        self.sync_lock.wait()
        while True:
            frame = self.display_frame_queue.get()
            result = self.predict_result_queue.get()
            for i in result:
                object_type = int(i[-1])
                cv2.rectangle(frame, (int(i[0]), int(i[1])), (int(i[2]), int(i[3])), color[object_type], 1)
                cv2.putText(frame, f"{format(i[-2], '.2f')} {class_name[object_type]}", (int(i[0]), int(i[1] - 10)),
                            cv2.FONT_HERSHEY_SIMPLEX, 0.6, color[object_type], 2)
            cv2.imshow("vnaoisdf", frame)
            if cv2.waitKey(1) & 0xff == ord('c'):
                self.lock.set()
                cv2.destroyAllWindows()
                break

        return

    def run(self):
        self.show()


class Model(multiprocessing.Process):
    def __init__(self, model_path, lock, sync_lock, predict_frame_queue, predict_result_queue):
        super().__init__(name="MODEL")
        self.daemon = True
        self.model_path = model_path
        self.lock = lock
        self.sync_lock = sync_lock
        self.predict_frame_queue = predict_frame_queue
        self.predict_result_queue = predict_result_queue

    def predict(self):

        model = torch.hub.load('ultralytics/yolov5', 'custom', path=self.model_path)
        model.cuda()
        self.sync_lock.wait()
        while True:
            if not self.predict_frame_queue.empty():
                frame = self.predict_frame_queue.get()
                predict = model(frame).xyxy[0].cpu().numpy()
                self.predict_result_queue.put(predict)
            if self.lock.is_set():
                break
        self.close()

        return

    def run(self):
        self.predict()


if __name__ == "__main__":
    lock = multiprocessing.Event()
    sync_lock = multiprocessing.Barrier(3)
    display_queue = queue.Queue()
    predict_frame_queue = multiprocessing.Queue()
    predict_result_queue = multiprocessing.Queue()

    a = read_video("testingv.mp4", lock, sync_lock, display_queue, predict_frame_queue)
    b = display_video(lock, sync_lock, display_queue, predict_result_queue)
    c = Model("best200.pt", lock, sync_lock, predict_frame_queue, predict_result_queue)
    a.start()
    b.start()
    c.start()

    a.join()
    b.join()
    c.join()

    predict_frame_queue.close()
    predict_result_queue.close()
    c.terminate()

    print(f"thread : {threading.enumerate()}")
    print(f"process : {multiprocessing.active_children()}")

问题原因与解决方法 #

1. 关于QueueFeederThread #

这个线程是multiprocessing.Queue内部自动创建的守护线程，专门处理进程间队列的数据传输。它的属性是守护线程，意味着只要主线程结束，它会自动退出，不需要手动关闭。你看到它存活，是因为主线程还在执行收尾逻辑，程序最终会自动终止。

2. 主线程（_MainThread）无法正常结束的核心问题 #

主线程是程序本身，只有所有非守护线程结束后才会退出。你的代码存在几个阻塞线程退出的问题：

问题点1：队列get()阻塞卡死

display_video的show()方法中，frame = self.display_frame_queue.get()和result = self.predict_result_queue.get()是阻塞调用——如果队列空了，线程会一直卡在这，即使触发lock信号也无法执行break。

解决：给get()添加超时，或者先检查终止信号再取队列：

# 修改display_video的show方法循环部分
while True:
    # 优先检查终止信号
    if self.lock.is_set():
        cv2.destroyAllWindows()
        break
    # 非阻塞获取队列，避免卡死
    try:
        frame = self.display_frame_queue.get(timeout=0.1)
        result = self.predict_result_queue.get(timeout=0.1)
    except queue.Empty:
        continue
    # 后续绘制代码不变

问题点2：multiprocessing.Queue未完成清理

你调用了close()但没调用join_thread()，无法等待队列后台线程完成数据清理，会导致主线程被挂起。

解决：关闭队列后添加join_thread()：

predict_frame_queue.close()
predict_frame_queue.join_thread()
predict_result_queue.close()
predict_result_queue.join_thread()

问题点3：Model进程中不存在close()方法

Model的predict()方法里调用了self.close()，但这个方法没有定义，会抛出异常导致进程无法正常退出。

解决：删除self.close()，如果需要清理模型，手动添加释放代码：

# 修改Model的predict方法
def predict(self):
    model = torch.hub.load('ultralytics/yolov5', 'custom', path=self.model_path)
    model.cuda()
    self.sync_lock.wait()
    while True:
        if not self.predict_frame_queue.empty():
            frame = self.predict_frame_queue.get()
            predict = model(frame).xyxy[0].cpu().numpy()
            self.predict_result_queue.put(predict)
        if self.lock.is_set():
            # 清理模型和GPU缓存
            del model
            torch.cuda.empty_cache()
            break
    # 去掉self.close()
    return

问题点4：read_video无法及时响应终止信号

read_video的循环先检查队列大小，再读取视频，最后才检查lock——如果队列一直大于3，会卡在视频读取循环里，无法及时响应终止信号。

解决：把lock检查放到循环最开头：

# 修改read_video的read方法循环
while True:
    # 优先检查终止信号
    if self.lock.is_set():
        self.cv.release()
        break
    if self.display_frame_queue.qsize() <= 3:
        ret, frame = self.cv.read()
        if ret:
            frame = cv2.resize(frame, (1280, 720))
            self.display_frame_queue.put(frame)
            self.predict_frame_queue.put(frame)
        else:
            self.cv.release()
            break

最终效果 #

修改后，按下c触发lock.set()，所有线程/进程都会及时检测到信号并退出，队列完成清理后主线程会正常结束，QueueFeederThread也会随主线程自动退出，程序完全终止。

内容的提问来源于stack exchange，提问作者BON