手柄在python有效阻塞操作

Question

我使用python和OpenCV从RTSP流得到的视频。我从流中获得单帧并将其保存到文件系统。

我写了处理框架获取和保存StreamingWorker。此外，还有具有所有流对象StreamPool。我认为，作为StreamingWorker总是运行，应该只有每个核心一个，以便采取尽可能多地。然后StreamPool将提供VideoCapture对象可用StreamingWorker。

问题是，大多数的脚本运行，阻塞时间：

import os
import time
import threading
import cv2 as cv

class StreamingWorker(object):

    def __init__(self, stream_pool):
        self.stream_pool = stream_pool
        self.start_loop()

    def start_loop(self):
        while True:
            try:
                # getting a stream from the read_strategy
                stream_object = self.stream_pool.next()

                # getting an image from the stream
                _, frame = stream_object['stream'].read()

                # saving image to file system
                cv.imwrite(os.path.join('result', stream_object['feed'], '{}.jpg'.format(time.time())))

            except ValueError as e:
                print('[error] {}'.format(e))

class StreamPool(object):

    def __init__(self, streams):
        self.streams = [{'feed': stream, 'stream': cv.VideoCapture(stream)} for stream in streams]
        self.current_stream = 0
        self.lock = threading.RLock()

    def next(self):
        self.lock.acquire()
        if(self.current_stream + 1 >= len(self.streams)):
            self.current_stream = 0
        else:
            self.current_stream += 1
        result = self.streams[self.current_stream]
        self.lock.release()
        return result

def get_cores():
    # This function returns the number of available cores
    import multiprocessing
    return multiprocessing.cpu_count()


def start(stream_pool):
    StreamingWorker(stream_pool)

def divide_list(input_list, amount):
    # This function divides the whole list into list of lists
    result = [[] for _ in range(amount)]
    for i in range(len(input_list)):
        result[i % len(result)].append(input_list[i])
    return result

if __name__ == '__main__':

    stream_list = ['rtsp://some/stream1', 'rtsp://some/stream2', 'rtsp://some/stream3']

    num_cores = get_cores()
    divided_streams = divide_list(stream_list, num_cores)
    for streams in divided_streams:
        stream_pool = StreamPool(streams)
        thread = threading.Thread(target=start, args=(stream_pool))
        thread.start()

当我想到这一点，我并没有考虑到大部分的操作会阻塞操作，如：

# Getting a frame blocks
_, frame = stream_object['stream'].read()

# Writing to the file system blocks
cv.imwrite(os.path.join('result', stream_object['feed'], '{}.jpg'.format(time.time())))

与花费太多时间阻塞问题是，大部分的处理能力被浪费了。我想用期货与ThreadPoolExecutor的，但我似乎无法达到我的使用处理核心可能的最大数量的目标。也许我没有设置enaugh线程。

是否有处理阻塞操作，以使核处理能力的最佳使用一个标准的方式？我没事具有语言无关的答案。

Answer 1

我结束了使用使用ThreadPoolExecutor功能add_done_callback(fn)。

class StreamingWorker(object):

    def __init__(self, stream_pool):
        self.stream_pool = stream_pool
        self.thread_pool = ThreadPoolExecutor(10)
        self.start_loop()

    def start_loop(self):
        def done(fn):
            print('[info] future done')

        def save_image(stream):
            # getting an image from the stream
            _, frame = stream['stream'].read()

            # saving image to file system
            cv.imwrite(os.path.join('result', stream['feed'], '{}.jpg'.format(time.time())))

        while True:
            try:
                # getting a stream from the read_strategy
                stream_object = self.stream_pool.next()

                # Scheduling the process to the thread pool
                self.thread_pool.submit(save_image, (stream_object)).add_done_callback(done)
            except ValueError as e:
                print('[error] {}'.format(e))

我其实没有什么都想做未来结束后，但如果我用result()那么while True将停止，这也击败对子级使用线程池的所有目的。

附注：我有打电话threading.Rlock()时添加一个self.stream_pool.next()因为很明显的OpenCV不能处理从多个线程调用。