python - 图像处理中的多线程——视频python opencv

Question

我正在使用 opencv python 从实时流视频中进行对象检测。我的程序在单线程上运行，因为屏幕上显示的结果视频看起来不像视频，因为检测过程有延迟。因此，我尝试使用多线程重新实现它。我使用一个线程读取帧，另一个线程显示检测结果，大约 5 个线程同时在多个帧上运行检测算法。我写了下面的代码但是结果和单线程程序没有什么不同。我是 python 的新手。因此，我们将不胜感激。

import threading, time
import cv2
import queue


def detect_object():
    while True:
        print("get")
        frame = input_buffer.get()
        if frame is not None:
            time.sleep(1)
            detection_buffer.put(frame)
        else:
            break
    return


def show():
    while True:
        print("show")
        frame = detection_buffer.get()
        if frame is not None:
            cv2.imshow("Video", frame)
        else:
            break
        if cv2.waitKey(1) & 0xFF == ord('q'):
            break
    return


if __name__ == "__main__":

    input_buffer = queue.Queue()
    detection_buffer = queue.Queue()

    cap = cv2.VideoCapture(0)

    for i in range(5):
        t = threading.Thread(target=detect_object)
        t.start()

    t1 = threading.Thread(target=show)
    t1.start()

    while True:
        ret, frame = cap.read()
        if ret:
            input_buffer.put(frame)
            time.sleep(0.025)
        else:
            break

    print("program ended")

Answer 1

假设检测算法是 CPU 密集型的，您需要使用多处理而不是多线程，因为由于争用全局解释器锁，多个线程不会并行运行 Python 字节码。您还应该摆脱对 sleep 的所有调用。当您运行多个线程或以您的方式处理时，也不清楚是什么保证帧将以正确的顺序输出，也就是说，第二帧的处理可以在第一帧的处理之前完成，并且首先写入detection_buffer。

下面使用了 6 个进程的处理池(现在不需要隐式输入队列)。

from multiprocessing import Pool, Queue
import time
import cv2

# intialize global variables for the pool processes:
def init_pool(d_b):
    global detection_buffer
    detection_buffer = d_b


def detect_object(frame):
    time.sleep(1)
    detection_buffer.put(frame)


def show():
    while True:
        print("show")
        frame = detection_buffer.get()
        if frame is not None:
            cv2.imshow("Video", frame)
        else:
            break
        if cv2.waitKey(1) & 0xFF == ord('q'):
            break
    return


# required for Windows:
if __name__ == "__main__":

    detection_buffer = Queue()
    # 6 workers: 1 for the show task and 5 to process frames:
    pool = Pool(6, initializer=init_pool, initargs=(detection_buffer,))
    # run the "show" task:
    show_future = pool.apply_async(show)

    cap = cv2.VideoCapture(0)

    futures = []
    while True:
        ret, frame = cap.read()
        if ret:
            f = pool.apply_async(detect_object, args=(frame,))
            futures.append(f)
            time.sleep(0.025)
        else:
            break
    # wait for all the frame-putting tasks to complete:
    for f in futures:
        f.get()
    # signal the "show" task to end by placing None in the queue
    detection_buffer.put(None)
    show_future.get()
    print("program ended")