基于FFmpeg与Python OpenCV实现多路RTMP/RTSP推流的问题

通过Python OpenCV采集多路视频流，使用FFmpeg推送到RTMP/RTSP服务器时，控制台显示两路流均已推送，但接收端仅第二路能正常播放画面，第一路仅能收到元数据，无实际画面输出。需要实现多路流在接收端独立正常播放。

原代码 #

## st.txt = 'rtsp://ip:port/source/1' & 'rtsp://ip:port/source/2'
from multiprocessing.pool import ThreadPool
import cv2
import random
import subprocess

def f(x):
    name = (x.split('/'))[-1]
    y=30
    if len(x)==1:
        x = int(x)

    cam = cv2.VideoCapture(x)
    width = cam.get(cv2.CAP_PROP_FRAME_WIDTH)
    height = cam.get(cv2.CAP_PROP_FRAME_HEIGHT)
    fps = cam.get(cv2.CAP_PROP_FPS)
    def streamer():
        command = ['ffmpeg',
                            '-r', str(fps ),
                            '-y',
                            '-f', 'rawvideo',

                            '-pix_fmt', 'bgr24',
                            '-s', "{}x{}".format(int(width),int(height)),
                            '-i', '-',
                            '-c:v', 'libx264',
                            '-pix_fmt', 'yuv420p',
                            '-preset', 'fast',
                            '-bufsize','7000k',
                            '-flvflags', 'no_duration_filesize', 
                            '-g','180',
                            '-f', 'flv',
                            'rtmp://ip:port/live/'+ str(name)]                        
        return command                        
    p_stream = subprocess.Popen(streamer() , stdin=subprocess.PIPE)


    while cam.isOpened():
                ret, frame = cam.read()
                if not ret:
                    break
                cv2.imshow(str(name),frame)
                p_stream.stdin.write(frame.tobytes()) 
                key = cv2.waitKey(1)
                
                if key == ('w'):
                    cam.release()
                    cv2.destroyAllWindows()
                    break

def ls():
    with open(r'st.txt') as f:
        lns = [line.rstrip('\n') for line in f]  
         return lns


if __name__ == '__main__':
    
    pool = ThreadPool()
    results = pool.map(f, ls())

解决方案 #

核心问题与修复点 #

1. 线程池资源竞争
   ThreadPool的线程共享同一进程资源，易导致FFmpeg子进程的标准输入流出现数据混乱或阻塞。改用multiprocessing.Process创建独立进程处理每一路流，完全隔离资源。

2. FPS参数异常
   cv2.CAP_PROP_FPS可能读取到0或不准确值，导致FFmpeg的-r参数无效，第一路流帧率异常无法渲染。添加帧率校验，若读取值为0则设置默认帧率（如25）。

3. 缺少时间戳生成
   未添加时间戳参数，部分播放器无法正确解析流数据。在FFmpeg命令中加入-fflags +genpts强制生成时间戳。

4. 标准输入缓冲未刷新
   写入帧数据后未刷新缓冲区，数据堆积在内存中未发送给FFmpeg，导致第一路流无实际画面数据。写入后调用p_stream.stdin.flush()。

5. OpenCV窗口阻塞
   多线程/进程中cv2.imshow可能引发阻塞，影响流推送逻辑。若不需要预览可直接移除，如需保留则确保每个进程独立处理窗口事件。

修改后的代码 #

## st.txt = 'rtsp://ip:port/source/1' & 'rtsp://ip:port/source/2'
import cv2
import subprocess
from multiprocessing import Process

def stream_process(x):
    name = (x.split('/'))[-1]
    # 处理本地摄像头的情况
    if len(x) == 1:
        x = int(x)

    cam = cv2.VideoCapture(x)
    # 校验摄像头是否成功打开
    if not cam.isOpened():
        print(f"无法打开流: {x}")
        return

    width = int(cam.get(cv2.CAP_PROP_FRAME_WIDTH))
    height = int(cam.get(cv2.CAP_PROP_FRAME_HEIGHT))
    fps = cam.get(cv2.CAP_PROP_FPS)
    # 修复帧率异常问题，设置默认值
    if fps <= 0:
        fps = 25

    # 构建FFmpeg命令，添加时间戳参数
    command = [
        'ffmpeg',
        '-r', str(fps),
        '-y',
        '-f', 'rawvideo',
        '-pix_fmt', 'bgr24',
        '-s', f"{width}x{height}",
        '-i', '-',
        '-c:v', 'libx264',
        '-pix_fmt', 'yuv420p',
        '-preset', 'fast',
        '-bufsize', '7000k',
        '-flvflags', 'no_duration_filesize',
        '-g', '180',
        '-fflags', '+genpts',  # 添加时间戳生成
        '-f', 'flv',
        f'rtmp://ip:port/live/{name}'
    ]

    p_stream = subprocess.Popen(command, stdin=subprocess.PIPE)

    try:
        while cam.isOpened():
            ret, frame = cam.read()
            if not ret:
                break
            # 可选：移除预览窗口避免阻塞
            # cv2.imshow(str(name), frame)
            # cv2.waitKey(1)
            
            # 写入帧并刷新缓冲区
            p_stream.stdin.write(frame.tobytes())
            p_stream.stdin.flush()

    except Exception as e:
        print(f"流 {name} 出现异常: {e}")
    finally:
        cam.release()
        p_stream.stdin.close()
        p_stream.terminate()
        # cv2.destroyWindow(str(name))

def get_stream_list():
    with open(r'st.txt') as f:
        return [line.rstrip('\n') for line in f]

if __name__ == '__main__':
    stream_list = get_stream_list()
    processes = []
    # 为每个流创建独立进程
    for stream_url in stream_list:
        p = Process(target=stream_process, args=(stream_url,))
        p.start()
        processes.append(p)
    
    # 等待所有进程结束
    for p in processes:
        p.join()

内容的提问来源于stack exchange，提问作者Sia