如何停止cv2人体检测程序的运行 避免死循环阻塞主程序执行

核心问题说明 #

你遇到的阻塞问题源于检测代码把视频读取、逻辑处理、窗口渲染全耦合在同一个无限循环里，只要调用就会占满当前线程，导致主程序后续代码无法执行。下面提供两种可直接落地的解决方案：

方案1：封装为单帧处理工具（最推荐） #

把核心检测逻辑拆成独立类/函数，每次调用仅处理单帧图像，循环逻辑完全交给主程序控制，耦合度最低、灵活度最高。

重构后的检测模块代码（pose_detector.py） #

import cv2
import mediapipe as mp

mpDraw = mp.solutions.drawing_utils
mppose = mp.solutions.pose

class PoseDetector:
    def __init__(self):
        self.pose = mppose.Pose()
    
    def process_frame(self, img):
        # 输入为BGR格式的图像帧，返回关键点列表和绘制后的图像
        imgRGB = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
        results = self.pose.process(imgRGB)
        lmlist = []
        if results.pose_landmarks:
            mpDraw.draw_landmarks(img, results.pose_landmarks, mppose.POSE_CONNECTIONS)
            h, w, c = img.shape
            for id, lm in enumerate(results.pose_landmarks.landmark):
                cx, cy = int(lm.x*w), int(lm.y*h)
                lmlist.append([id, cx, cy])
            # 绘制鼻关键点
            if lmlist:
                cv2.circle(img, (lmlist[0][1], lmlist[0][2]), 15, (255, 0, 255), cv2.FILLED)
        return lmlist, img

主程序调用示例 #

import cv2
import time
from pose_detector import PoseDetector

if __name__ == "__main__":
    detector = PoseDetector()
    cap = cv2.VideoCapture('open cv/squidgamee/3.mp4')
    cap.set(3, 400)
    cap.set(4, 800)
    ptime = 0
    
    # 主程序自主控制循环，可随时加终止条件
    while True:
        succ, img = cap.read()
        # 视频读完自动退出
        if not succ:
            break 
        lmlist, processed_img = detector.process_frame(img)
        
        # 此处可插入主程序其他业务逻辑，不会被阻塞
        if lmlist:
            print(lmlist[0])
            # 示例：满足自定义条件直接终止检测
            # if lmlist[0][1] > 300:
            #     break
        
        # 渲染逻辑也可自主控制
        ctime = time.time()
        fps = 1/(ctime-ptime)
        ptime = ctime
        cv2.putText(processed_img, str(int(fps)), (70, 50), cv2.FONT_HERSHEY_COMPLEX, 3, (255,255,0), 3)
        cv2.imshow("image", processed_img)
        key = cv2.waitKey(1)
        if key == ord('q'):
            break
    
    # 后续主程序代码可正常执行
    cap.release()
    cv2.destroyAllWindows()
    print("检测结束，执行主程序后续逻辑")

方案2：多线程运行检测逻辑 #

如果需要检测任务后台运行、不影响主程序同步执行其他逻辑，可以把整个检测循环放到子线程中运行。

调整后的检测模块代码（pose_detector.py） #

import cv2
import mediapipe as mp
import time

mpDraw = mp.solutions.drawing_utils
mppose = mp.solutions.pose
pose = mppose.Pose()

def run_pose_detection(video_path, exit_flag=None):
    cap = cv2.VideoCapture(video_path)
    cap.set(3, 400)
    cap.set(4, 800)
    ptime =0
    while True:
        # 支持主程序主动发送信号终止检测
        if exit_flag is not None and exit_flag.is_set():
            break
        succ, img = cap.read()
        if not succ:
            break
        imgRGB = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
        results = pose.process(imgRGB)
        if results.pose_landmarks:
            mpDraw.draw_landmarks(img, results.pose_landmarks, mppose.POSE_CONNECTIONS)
            lmlist= []
            for id, lm in enumerate(results.pose_landmarks.landmark):
                h, w, c = img.shape
                cx ,  cy = int(lm.x*w), int(lm.y*h)
                lmlist.append([id, cx, cy])
                cv2.circle(img, (lmlist[0][1], lmlist[0][2]), 15 , (255, 0, 255), cv2.FILLED)
            print(lmlist[0])
        ctime = time.time()
        fps = 1/(ctime-ptime)
        ptime = ctime
        cv2.putText(img, str(int(fps)), (70, 50), cv2.FONT_HERSHEY_COMPLEX, 3, (255,255,0), 3)
        cv2.imshow("image", img)
        key= cv2.waitKey(1)
        if key == ord('q'):
            break
    cap.release()
    cv2.destroyAllWindows()

主程序调用示例 #

import threading
from pose_detector import run_pose_detection

if __name__ == "__main__":
    # 创建退出信号，主程序可主动终止检测
    exit_flag = threading.Event()
    detect_thread = threading.Thread(target=run_pose_detection, args=('open cv/squidgamee/3.mp4', exit_flag))
    detect_thread.start()
    
    # 主程序其他逻辑可直接执行，不会被阻塞
    print("主程序其他逻辑正在运行...")
    # 需要主动停止检测时调用即可
    # exit_flag.set()
    # 等待检测线程结束
    detect_thread.join()
    print("检测结束，主程序继续执行")

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