Python多进程中OpenCV GStreamer VideoWriter write挂起问题

你的核心问题是：将视频编码逻辑放到独立Process后，本地文件写入正常，但GStreamer管道会阻塞在cv2.VideoWriter.write()调用处，且管道在主进程中可正常工作。以下是针对性的调试和解决方向：

1. 调整GStreamer管道的初始化时机 #

cv2.VideoWriter初始化GStreamer管道的操作是在主进程的__init__方法中完成的，而GStreamer的上下文（如总线、线程池）可能与进程绑定，子进程复用主进程初始化的管道会导致异常。

修改方案：将VideoWriter的初始化移到子进程的run()方法中（即子进程的执行上下文内）：

class VideoEncoder(Process):
    def __init__(self):
        super(VideoEncoder, self).__init__()
        self.loop = ProcessEvent()
        self.loop.set()
        self.new_frame = ProcessEvent()
        self.new_frame.clear()
        # 先不初始化VideoWriter
        self.video_writer = None
        data = np.zeros((1232, 1640, 3), dtype=np.uint8).flatten()
        self.frame = shared_memory.SharedMemory(create=True, size=data.nbytes)

    def run(self):
        # 在子进程内初始化GStreamer管道
        self.video_writer = cv2.VideoWriter(
            "my_very_long_gstreamer_pipeline",
            fourcc=0,
            fps=30,
            frameSize=(960, 720),
            isColor=True,
        )
        while self.loop.is_set():
            if self.new_frame.wait():
                self.process_frame()
                self.new_frame.clear()
        self.video_writer.release()

2. 启用GStreamer调试日志定位阻塞点 #

GStreamer默认不输出详细日志，无法看到管道内部的阻塞原因。可以在子进程启动前设置环境变量，强制输出调试信息：

import os

# 在encoder.start()前添加
os.environ["GST_DEBUG"] = "3"  # 3级日志足够定位常见问题，可按需调到5级
encoder = VideoEncoder()
encoder.start()

运行后会输出GStreamer各元素的状态变化，比如某个下游元素是否处于阻塞状态、是否有数据积压等，能直接找到write()阻塞的根源。

3. 给GStreamer管道添加缓冲队列 #

如果阻塞是因为下游处理速度跟不上（比如网络传输延迟），可以在管道中插入queue元素增加缓冲区，避免上游被阻塞：
示例管道修改（根据你的实际管道调整）：

# 原管道：appsrc ! ... ! autovideosink
# 修改后：appsrc ! queue ! ... ! autovideosink

queue元素会创建独立的线程处理数据，隔离上下游的速度差异，防止上游的write()调用被阻塞。

4. 验证共享内存的帧同步 #

当前的事件通知机制可能存在帧数据覆盖的风险（主进程在子进程处理完前就写入新帧），虽然日志显示第一次写入正常，但后续阻塞可能和帧数据损坏有关。添加互斥锁确保读写互斥：

from multiprocessing import Lock

class VideoEncoder(Process):
    def __init__(self):
        super(VideoEncoder, self).__init__()
        self.loop = ProcessEvent()
        self.loop.set()
        self.new_frame = ProcessEvent()
        self.new_frame.clear()
        self.frame_lock = Lock()  # 添加互斥锁
        self.video_writer = None
        data = np.zeros((1232, 1640, 3), dtype=np.uint8).flatten()
        self.frame = shared_memory.SharedMemory(create=True, size=data.nbytes)

    def write(self, frame: np.ndarray):
        with self.frame_lock:  # 写时加锁
            sha = np.ndarray(frame.shape, dtype=frame.dtype, buffer=self.frame.buf)
            sha[:] = frame[:]
        self.new_frame.set()
        print("Requested new frame")

    def process_frame(self):
        with self.frame_lock:  # 读时加锁
            frame = np.ndarray((1232, 1640, 3), dtype=np.uint8, buffer=self.frame.buf)
            # 复制帧数据到本地变量，避免锁持有时间过长
            frame_copy = frame.copy()
        self.new_frame.clear()

        frame_copy = cv2.resize(frame_copy, (960, 720))
        print("Writing frame to video writer...")
        self.video_writer.write(frame_copy)
        print(" > Written frame!")

5. 测试OpenCV GStreamer后端的子进程兼容性 #

如果以上方法都无效，可能是OpenCV的GStreamer后端在子进程中存在兼容性问题。可以用最简代码测试子进程中GStreamer管道的可用性：

from multiprocessing import Process
import cv2
import numpy as np

def test_gst_subprocess():
    pipeline = "your_gstreamer_pipeline_here"
    writer = cv2.VideoWriter(pipeline, 0, 30, (960,720), True)
    if not writer.isOpened():
        print("VideoWriter failed to open!")
        return
    for i in range(10):
        frame = np.zeros((720,960,3), dtype=np.uint8)
        print(f"Writing frame {i}")
        writer.write(frame)
    writer.release()

if __name__ == "__main__":
    p = Process(target=test_gst_subprocess)
    p.start()
    p.join()

如果这个测试也阻塞，说明需要换用GStreamer原生Python绑定（如pygobject）直接操作GStreamer管道，替代cv2.VideoWriter。

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