在 Python 中同时运行依赖线程

Question

我有两个线程类提取和检测。

Extract 从视频中提取帧并将其存储在文件夹中，Detect 从提取帧的文件夹中获取图像并检测对象。

但是当我运行下面的代码时，只有提取有效：

global q
q = Queue()

class extract(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)

    def run(self):
        print("T1")
        cam = cv2.VideoCapture(video_name)
        frameNum = 0
        # isCaptured = True
        frameCount = 0
        while True:
            isCapture, frame = cam.read()
            if not isCapture:
                break
            if frameCount % 5 == 0:
                frameNum = frameNum + 1
                fileName = vid + str(frameNum) + '.jpg'
                cv2.imwrite('images/extracted/' + fileName, frame)
                q.put(fileName)
            frameCount += 1
        cam.release()
        cv2.destroyAllWindows()

class detect(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)

    def run(self):
        print("T2")
        #logic to detect objects. 


if __name__ == '__main__':
    thread1 = extract()
    thread1.start()
    thread2 = detect()
    thread2.start()

这只会打印 T1 而不会打印 T2。 我想可能检测到先运行，队列是空的，所以什么也没发生，所以我在队列中添加了虚拟条目，它按照我想要的方式运行。

但它只针对虚拟条目运行，它不适用于提取函数添加到队列中的条目。 查找了其他问题，但似乎都没有解决问题，因此在此处发布此问题

Answer 1

您可能还希望将检测逻辑保持在无限循环中。

class detect(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)

    def run(self):
        while True: 
        #detect frame

如果是单帧检测。 然后考虑在检测线程中等待。

from time import sleep
class detect(threading.Thread):
        def __init__(self):
            threading.Thread.__init__(self)

        def run(self):
           sleep(120)
           # Detect logic

Answer 2

您可以使用Event() 并让您的检测线程等到设置Event() 后再执行检测，而不是等待硬编码时间

如果设置了事件，则意味着所有任务都已完成。此外，如果有任何任务尚未处理，您还必须密切关注队列。

我已经编写了一个示例代码来演示它是如何工作的，您可以根据需要修改代码。

这里extract 需要 5 秒来将任务添加到队列中，detect 每 1 秒检查一次任务。因此，如果提取的速度比任何可用的东西都慢，那么detect 将处理它。当所有任务都完成后，检测会跳出循环。

import threading
import queue
import time

global q
q = queue.Queue()

class extract(threading.Thread):
    all_tasks_done = threading.Event()
    def __init__(self):
        threading.Thread.__init__(self)

    def run(self):
        counter = 5
        while counter:
            time.sleep(5)
            counter -= 1
            q.put(1)
            print("added a task to queue")
        extract.all_tasks_done.set()



class detect(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)

    def run(self):
        while not extract.all_tasks_done.wait(1) or not q.empty():
            print(q.get())
        print("detection done")
        #logic to detect objects. 


if __name__ == '__main__':
    thread1 = extract()
    thread1.start()
    thread2 = detect()
    thread2.start()
    thread1.join()
    thread2.join()
    print("work done")