要点：

参考： process读取多进程

参考：实时视频传入深度学习目标检测模型进行检测 - 知乎

一 实时读取视频流

解决实时读取延迟卡顿问题，实时读取多个网络摄像头。

读取视频：

import cv2
import time
import multiprocessing as mp

# 导入必要的模块
def image_put(q, name, pwd, ip, channel=1):
    """
    图像获取函数，将摄像头的图像放入队列中
    :param q: 队列，用于存储摄像头图像
    :param name: 摄像头用户名
    :param pwd: 摄像头密码
    :param ip: 摄像头IP地址
    :param channel: 摄像头通道号，默认为1
    """
    # 尝试使用 HIKVISION 协议连接摄像头
    cap = cv2.VideoCapture("rtsp://%s:%s@%s//Streaming/Channels/%d" % (name, pwd, ip, channel))
    if cap.isOpened():
        print('HIKVISION')
    else:
        # 如果连接失败，则尝试使用 DaHua 协议连接摄像头
        cap = cv2.VideoCapture("rtsp://%s:%s@%s/cam/realmonitor?channel=%d&subtype=0" % (name, pwd, ip, channel))
        print('DaHua')

    while True:
        q.put(cap.read()[1])  # 从摄像头读取图像并放入队列
        q.get() if q.qsize() > 1 else time.sleep(0.01)  # 控制队列大小和图像获取的间隔

def image_get(q, window_name):
    """
    图像显示函数，从队列中获取图像并显示
    :param q: 队列，用于存储摄像头图像
    :param window_name: 窗口名称，用于显示图像
    """
    cv2.namedWindow(window_name, flags=cv2.WINDOW_FREERATIO)  # 创建一个命名窗口
    while True:
        frame = q.get()  # 从队列中获取图像
        cv2.imshow(window_name, frame)  # 显示图像
        cv2.waitKey(1)  # 等待按键响应，每隔1毫秒刷新窗口

def run_multi_camera():
    """
    多摄像头并行处理函数，用于设置摄像头的参数并启动多个图像获取和显示进程
    """
    user_name, user_pwd = "admin", "admin123456"  # 设置摄像头的用户名和密码
    
    camera_ip_l = [
        "172.20.114.26",  # 摄像头IP地址列表，可以包含多个IP地址
        "[fe80::3aaf:29ff:fed3:d260]",  # 如果是IPv6地址，需要加上中括号
        # 添加你的摄像头IP地址到这里
    ]

    mp.set_start_method(method='spawn')  # 设置多进程的启动方式为spawn
    queues = [mp.Queue(maxsize=4) for _ in camera_ip_l]  # 创建多个队列，用于存储不同摄像头的图像

    processes = []
    for queue, camera_ip in zip(queues, camera_ip_l):
        # 创建图像获取和显示进程，并传递相应的参数
        processes.append(mp.Process(target=image_put, args=(queue, user_name, user_pwd, camera_ip)))  # 创建图像获取进程
        processes.append(mp.Process(target=image_get, args=(queue, camera_ip)))  # 创建图像显示进程

    for process in processes:
        process.daemon = True  # 设置进程为守护进程，即主进程退出时自动结束子进程
        process.start()  # 启动进程

    for process in processes:
        process.join()  # 等待所有进程结束

if __name__ == '__main__':
    run_multi_camera()  # 执行主函数

在 run_multi_camera 函数中，首先设置摄像头的用户名和密码，并创建一个存储不同摄像头图像的队列列表。然后，使用 mp.set_start_method 函数设置多进程的启动方式为 spawn 。接下来，通过循环遍历摄像头IP地址列表，为每个摄像头创建一个图像获取进程和一个图像显示进程，并将队列和相应的参数传递给它们。然后，将所有进程设置为守护进程，并依次启动它们。最后，使用 process.join() 等待所有进程结束。

1.1 解决实时读取延迟卡顿

关键代码如下，我使用Python自带的多线程队列：

import multiprocessing as mp
...
img_queues = [mp.Queue(maxsize=2) for _ in camera_ip_l]  # 创建多个队列，每个摄像头对应一个队列
...
q.put(frame) if is_opened else None  # 如果摄像头打开成功，则将帧放入队列中，线程A负责将图片放入队列
q.get() if q.qsize() > 1 else time.sleep(0.01)  # 如果队列中的帧数量大于1，则从队列中移除旧的帧，线程A负责移除队列中的旧图
...

这段代码使用了 multiprocessing.Queue 创建了多个队列，每个摄像头对应一个队列。这些队列用于在不同的进程之间传递帧数据。

在第一个注释中，img_queues是一个列表，通过列表推导式创建了多个队列，其中maxsize=2表示每个队列的最大容量为2，即最多可以存放2帧数据。

扫描二维码关注公众号，回复： 15167835 查看本文章

在第二个注释中，根据条件判断是否将帧放入队列中。is_opened表示摄像头是否打开成功，如果打开成功，则将当前帧frame放入对应的队列q中。

在第三个注释中，根据队列的大小判断是否需要从队列中移除旧的帧。q.qsize()返回队列中的帧数量，如果大于1，则调用q.get()从队列中移除旧的帧；否则，通过time.sleep(0.01)暂停0.01秒，等待新的帧放入队列。这样可以保持队列中始终只有最新的帧数据，避免队列过大导致内存占用过高。

这些操作是在多个线程或进程之间并行执行的，通过队列进行线程间通信，实现了帧数据的生产和消费，保证了实时的视频显示。

二 详细解读

2.1 入门OpenCV读取方式

def run_opencv_camera():
    video_stream_path = 0  # local camera (e.g. the front camera of laptop)
    cap = cv2.VideoCapture(video_stream_path)

    while cap.isOpened():
        is_opened, frame = cap.read()
        cv2.imshow('frame', frame)
        cv2.waitKey(1)
    cap.release()

当我们需要读取网络摄像头的时候，我们可以对 cap = cv2.VideoCapture(括号里面的东西进行修改)，填写上我们想要读取的视频流，它可以是：

• 数字0，代表计算机的默认摄像头（例如上面提及的笔记本前置摄像头）
• video.avi 视频文件的路径，支持其他格式的视频文件
• rtsp路径（不同品牌的路径一般是不同的，如下面举出的海康与大华，详细情况查看 附录的「关于rtsp协议」）

user, pwd, ip, channel = "admin", "admin123456", "172.20.114.26", 1

video_stream_path = 0  # local camera (e.g. the front camera of laptop)
video_stream_path = 'video.avi'  # the path of video file
video_stream_path = "rtsp://%s:%s@%s/h265/ch%s/main/av_stream" % (user, pwd, ip, channel)  # HIKIVISION old version 2015
video_stream_path = "rtsp://%s:%s@%s//Streaming/Channels/%d" % (user, pwd, ip, channel)  # HIKIVISION new version 2017
video_stream_path = "rtsp://%s:%s@%s/cam/realmonitor?channel=%d&subtype=0" % (user, pwd, ip, channel)  # dahua

cap = cv2.VideoCapture(video_stream_path)

直接使用参考官网写出来的简单版Demo有延迟卡顿问题，如果读取速度低于视频流的输出速度，窗口显示的图片是好几秒钟前的内容。一段时间过后，缓存区将会爆满，程序报错，我可以使用rtsp读取摄像头：

2.2 使用rtsp读取摄像头

def run_opencv_camera():
    video_stream_path = 0  # local camera (e.g. the front camera of laptop)
    cap = cv2.VideoCapture(video_stream_path)

    while cap.isOpened():
        is_opened, frame = cap.read()
        cv2.imshow('frame', frame)
        cv2.waitKey(1000)  # wait for 1000ms(1s) HERE!!!!!!!!!!!!!!
    cap.release()

2.3 使用多线程队列，解决延迟卡顿问题

import cv2
import time
import multiprocessing as mp

def image_put(q, user, pwd, ip, channel=1):
    # 使用rtsp协议连接摄像头
    cap = cv2.VideoCapture("rtsp://%s:%s@%s//Streaming/Channels/%d" % (user, pwd, ip, channel))
    if cap.isOpened():
        print('HIKVISION')
    else:
        # 如果连接失败，尝试使用不同的URL格式连接摄像头
        cap = cv2.VideoCapture("rtsp://%s:%s@%s/cam/realmonitor?channel=%d&subtype=0" % (user, pwd, ip, channel))
        print('DaHua')

    while True:
        # 读取视频帧并将其放入队列中
        q.put(cap.read()[1])
        # 如果队列中的帧数超过阈值，移除最早的帧
        q.get() if q.qsize() > 1 else time.sleep(0.01)

def image_get(q, window_name):
    # 创建窗口来显示视频帧
    cv2.namedWindow(window_name, flags=cv2.WINDOW_FREERATIO)
    while True:
        # 从队列中获取视频帧并显示在窗口中
        frame = q.get()
        cv2.imshow(window_name, frame)
        cv2.waitKey(1)

def run_single_camera():
    user_name, user_pwd, camera_ip = "admin", "admin123456", "172.20.114.26"

    mp.set_start_method(method='spawn')  # 初始化多进程方法
    queue = mp.Queue(maxsize=2)  # 创建一个队列，用于存储视频帧

    processes = [
        mp.Process(target=image_put, args=(queue, user_name, user_pwd, camera_ip)),
        mp.Process(target=image_get, args=(queue, camera_ip))
    ]

    # 启动并等待子进程完成
    [process.start() for process in processes]
    [process.join() for process in processes]

def run_multi_camera():
    user_name, user_pwd = "admin", "admin123456"
    camera_ip_l = [
        "172.20.114.26",  # ipv4
        "[fe80::3aaf:29ff:fed3:d260]",  # ipv6
    ]

    mp.set_start_method(method='spawn')  # 初始化多进程方法
    queues = [mp.Queue(maxsize=4) for _ in camera_ip_l]  # 创建多个队列，用于存储视频帧

    processes = []
    for queue, camera_ip in zip(queues, camera_ip_l):
        processes.append(mp.Process(target=image_put, args=(queue, user_name, user_pwd, camera_ip)))
        processes.append(mp.Process(target=image_get, args=(queue, camera_ip)))

    # 启动并等待子进程完成
    for process in processes:
        process.daemon = True
        process.start()
    for process in processes:
        process.join()

if __name__ == '__main__':
    # 运行单个摄像头示例
    # run_single_camera()
    
    # 运行多个摄像头示例
    run_multi_camera()
    pass

使用Python3自带的多线程模块，创建一个队列，线程A从通过rtsp协议从视频流中读取出每一帧，并放入队列中，线程B从队列中将图片取出，处理后进行显示。线程A如果发现队列里有两张图片（证明线程B的读取速度跟不上线程A），那么线程A主动将队列里面的旧图片删掉，换上新图片。通过多线程的方法：

• 线程A的读取速度始终不收线程B的影响，防止网络摄像头的缓存区爆满
• 线程A更新了队列中的图片，使线程B始终读取到最新的画面，降低了延迟

import multiprocessing as mp
...
img_queues = [mp.Queue(maxsize=2) for _ in camera_ip_l]  # queue
...
q.put(frame) if is_opened else None  # 线程A不仅将图片放入队列
q.get() if q.qsize() > 1 else time.sleep(0.01) # 线程A还负责移除队列中的旧图
...

三 常见问题

3.1 关于rtsp协议

通过rtsp协议读取视频流，下面依次是我在网络上查到的海康与大华 rtsp 读取路径。经过测试，我手头的海康摄像头支持前面两种读取方式（新旧两种）。大华摄像头用第三种读取方式。

video_stream_path = "rtsp://%s:%s@%s/h264/ch%s/main/av_stream" % (user, pwd, ip, channel)  # HIKIVISION old version 2015
video_stream_path = "rtsp://%s:%s@%s//Streaming/Channels/%d" % (user, pwd, ip, channel)  # HIKIVISION new version 2017
video_stream_path = "rtsp://%s:%s@%s/cam/realmonitor?channel=%d&subtype=0" % (user, pwd, ip, channel)  # dahua

1 海康、大华IpCamera RTSP地址和格式

2 海康摄像机、NVR、流媒体服务器、回放取流RTSP地址规则说明

3.2 如果你需要收集多个摄像头拍摄的画面

（也就是将5个子进程拍摄到的图片收集 (collect) 到一个进程中去），那么我们需要这样子处理：

run_multi_camera_in_a_window() 修改自 run_multi_camera()
image_collect() 修改自 image_get()

def image_collect(queue_list, camera_ip_l):
    import numpy as np  # 实际使用的时候记得放在外面

    """show in single opencv-imshow window"""
    window_name = "%s_and_so_no" % camera_ip_l[0]
    cv2.namedWindow(window_name, flags=cv2.WINDOW_FREERATIO)
    while True:
        imgs = [q.get() for q in queue_list]
        imgs = np.concatenate(imgs, axis=1)
        cv2.imshow(window_name, imgs)
        cv2.waitKey(1)

    # """show in multiple opencv-imshow windows""" 
    # [cv2.namedWindow(window_name, flags=cv2.WINDOW_FREERATIO)
    #  for window_name in camera_ip_l]
    # while True:
    #     for window_name, q in zip(camera_ip_l, queue_list):
    #         cv2.imshow(window_name, q.get())
    #         cv2.waitKey(1)


def run_multi_camera_in_a_window():
    user_name, user_pwd = "admin", "admin123456"
    camera_ip_l = [
        "172.20.114.196",  # ipv4
        "[fe80::3aaf:29ff:fed3:d260]",  # ipv6
        # 我在这里分别用ipv4 与ipv6 打开了两个摄像头，只有一个摄像头的话就填写一个IP
    ]

    mp.set_start_method(method='spawn')  # init
    queues = [mp.Queue(maxsize=4) for _ in camera_ip_l]

    processes = [mp.Process(target=image_collect, args=(queues, camera_ip_l))]
    for queue, camera_ip in zip(queues, camera_ip_l):
        processes.append(mp.Process(target=image_put, args=(queue, user_name, user_pwd, camera_ip)))

    for process in processes:
        process.daemon = True  # setattr(process, 'deamon', True)
        process.start()
    for process in processes:
        process.join()

3.3 视频读取已经结束， 而程序没有自动退出

cap = cv2.VideoCapture(video_stream_path)
is_opened = cap.isOpened()

while is_opened:  # 如果视频读取完毕，那么 is_opened 为False，循环自动跳出
    is_opened, frame = cap.read()
    cv2.imshow('frame', frame)
    cv2.waitKey(1)
cap.release()