在我们使用OpenCV计算机开源视觉库去处理视屏流(rtsp, rtmp…)时，我们发现如果去实例化一个不存在的RTSP视屏流地址，会发生超时的问题，而且这个实例化线程是原子阻塞的，无法强制退出，这在某些场景下是不能被接受的，所以我们必须采取一定的方案去限定cv2.VideoCapture(rtsp)实例化的时间，经过多番探索，我发现用自定义多线程和装饰器可以解决这个问题，代码如下：

When we use the OpenCV computer open source vision library to process video streams (rtsp, rtmp…), we found that if we instantiate a non-existent RTSP video stream address, a timeout problem will occur, and This instantiation thread is atomically blocked and cannot be forced to exit. This is unacceptable in some scenarios, so we must adopt a certain plan to limit the instantiation time of cv2.VideoCapture(rtsp) After many explorations, I found that using custom multi-threading and decorators can solve this problem, the code is as follows:

# opencv_demo.py
# OpenCV-Python timeout for opening a non-existent RTSP video stream
import cv2
import base64
import time
import threading
TIME_LIMITED: int = 1


class MyThread(threading.Thread):
    def __init__(self, target, args=()):
        super(MyThread, self).__init__()
        self.func = target
        self.args = args

    def run(self):
        self.result = self.func(*self.args)

    def get_result(self):
        try:
            return self.result
        except Exception:
            return None


# Decorator to limit the actual request time or function execution time
def limit_decor(limit_time):
    """
    :param limit_time: Set the maximum allowable execution time, unit: second
    :return: Untimed returns the value of the decorated function; timed out returns None
    """
    def functions(func):
        def run(*params):
            thre_func = MyThread(target=func, args=params)
            # The thread method terminates when the main thread terminates (exceeds its length)
            thre_func.setDaemon(True)
            thre_func.start()
            # Count the number of segmental slumbers
            sleep_num = int(limit_time // 1)
            sleep_nums = round(limit_time % 1, 1)
            # Sleep briefly several times and try to get the return value
            for i in range(sleep_num):
                time.sleep(1)
                infor = thre_func.get_result()
                if infor:
                    return infor
            time.sleep(sleep_nums)
            # Final return value (whether or not the thread has terminated)
            if thre_func.get_result():
                return thre_func.get_result()
            else:
                return (False, None)  # Timeout returns can be customized

        return run

    return functions


@limit_decor(TIME_LIMITED)
def video_capture_open(rtsp):
    capture = cv2.VideoCapture(rtsp)
    return (True, capture)

def frame_get(rtsp):
    try:
        cap_status, cap = video_capture_open(rtsp)
        if not cap_status:
            print(cap_status, cap)
            return cap
        while True:
            ret, image_frame = cap.read()
            cv2.imshow("res", image_frame)
            cv2.waitKey(3)
            if not ret:
                continue
            image = cv2.imencode('.png', image_frame)[1]
            image_base64_data = str(base64.b64encode(image))[2:-1]
            return image_base64_data
    except Exception as err:
        print(err)
        pass


def main() -> None:

    RTSP = 'rtsp://192.168.3.88/av0_0' # No existing RTSP video stream address
    result = frame_get(RTSP)
    if not result:
        print("Failed to open RTSP stream")
    else:
        print(result[:10])

    return None


if __name__ == "__main__":
    main()

测试结果

python opencv_demo.py
False None
Failed to open RTSP stream