如何优雅终止ffmpeg进程及配套的ffprobe进度监控进程？

之前仅运行ffmpeg进程时可实现优雅终止，新增ffprobe进程跟踪处理进度后，取消进程时抛出异常。不想用try-except块规避，求正确实现方式。

P.S. 代码已简化为可执行示例，见谅。

示例代码 #

import subprocess as sp
import shlex
import json
import time
import threading

def start_ffmpeg_thread(audio_part, video_part, path):

    global ffmpeg_process
    if (ffmpeg_process is None) or ffmpeg_process.poll():
        data = sp.run(shlex.split(f'ffprobe -v error -select_streams v:0 -count_packets -show_entries stream=nb_read_packets -of csv=p=0 -of json "{video_part}"'), stdout=sp.PIPE).stdout
        dict = json.loads(data)
        tot_n_frames = float(dict['streams'][0]['nb_read_packets'])

        ffmpeg_process = sp.Popen(shlex.split(f'ffmpeg -y -loglevel error -i "{video_part}" -i "{audio_part}" -progress pipe:1 "{path}"'), stdout=sp.PIPE, stdin=sp.PIPE)

        q = [0]

        ffmpeg_progress_reader_thread = threading.Thread(target=ffmpeg_progress_reader, args=(ffmpeg_process, q))
        ffmpeg_progress_reader_thread.start()

        while True:
            if ffmpeg_process.poll() is not None:
                break

            n_frame = q[0]
            progress_percent = (n_frame/tot_n_frames)*100
            print(f"Progress: [%] {progress_percent}", end="\r")
        ffmpeg_progress_reader_thread.join()

def ffmpeg_progress_reader(procs, q):

    while True:
        if procs.poll() is not None:
            break

        progress_text = procs.stdout.readline()
        progress_text = progress_text.decode("utf-8")
        if progress_text.startswith("frame="):
            frame = int(progress_text.partition('=')[-1])
            q[0] = frame

def cancel_ffmpeg():

    time.sleep(10)
    global ffmpeg_process
    if (ffmpeg_process is not None) and (ffmpeg_process.poll() is None):
            ffmpeg_process.stdin.write('q'.encode("GBK"))
            ffmpeg_process.communicate()
            ffmpeg_process.wait()
            ffmpeg_process = None


ffmpeg_process = None

threading.Thread(target=cancel_ffmpeg).start()
start_ffmpeg_thread(<<AUDIO_FILE_FULL_PATH>>, <<VIDEO_FILE_FULL_PATH>>, <<OUTPUT_FULL_PATH>>)

抛出的异常 #

Exception in thread Thread-2 (ffmpeg_progress_reader):
Traceback (most recent call last):
  File "D:\Python311\Lib\threading.py", line 1038, in _bootstrap_inner
    self.run()9.354796147248976
  File "D:\Python311\Lib\threading.py", line 975, in run
    self._target(*self._args, **self._kwargs)
  File "d:\Python Projects\main.py", line 82, in ffmpeg_progress_reader
    progress_text = procs.stdout.readline()
                    ^^^^^^^^^^^^^^^^^^^^^^^
ValueError: PyMemoryView_FromBuffer(): info->buf must not be NULL
Traceback (most recent call last):
  File "d:\Python Projects\main.py", line 101, in <module>
    start_ffmpeg_thread("aud.mp3", "vid.mp4", "output.mp4")
  File "d:\Python Projects\main.py", line 69, in start_ffmpeg_thread
    if ffmpeg_process.poll() is not None:
       ^^^^^^^^^^^^^^^^^^^
AttributeError: 'NoneType' object has no attribute 'poll'

异常核心原因是多线程竞争全局变量ffmpeg_process，以及communicate()调用会直接关闭进程的stdin/stdout，导致进度读取线程报错。以下是针对性修正：

1. 用线程锁保护全局变量访问 #

全局变量ffmpeg_process被多线程同时读写，必须加锁避免竞态条件：

# 新增全局锁
process_lock = threading.Lock()

所有访问ffmpeg_process的代码块都要包裹在with process_lock:中，比如：

• 启动ffmpeg前的判断：

with process_lock:
    if (ffmpeg_process is None) or ffmpeg_process.poll():
        # 初始化ffmpeg进程逻辑

• 取消进程时的操作：

with process_lock:
    if (ffmpeg_process is not None) and (ffmpeg_process.poll() is None):
        # 终止进程逻辑

• 主循环中的进程状态检查：

while True:
    with process_lock:
        if ffmpeg_process is None or ffmpeg_process.poll() is not None:
            break

2. 替换communicate()，避免直接关闭流 #

communicate()会一次性读取所有输出并关闭stdin/stdout，导致进度线程的readline()报错。改用stdin.flush()+wait()组合：

# 替换cancel_ffmpeg中的代码
ffmpeg_process.stdin.write('q'.encode("utf-8"))  # 用utf-8兼容性更好
ffmpeg_process.stdin.flush()  # 确保命令发送到ffmpeg进程
ffmpeg_process.wait()         # 等待进程正常退出
with process_lock:
    ffmpeg_process = None

3. 给进度读取线程加终止标记 #

避免进度线程在进程退出后仍尝试读取流，新增线程安全的终止事件：

# 启动进度线程时传入终止事件
stop_event = threading.Event()
progress_thread = threading.Thread(target=ffmpeg_progress_reader, args=(ffmpeg_process, q, stop_event))
progress_thread.start()

# 主循环结束时触发终止事件
with process_lock:
    if ffmpeg_process is None or ffmpeg_process.poll() is not None:
        stop_event.set()
        break

# 修改进度读取函数
def ffmpeg_progress_reader(procs, q, stop_event):
    while not stop_event.is_set():
        with process_lock:
            if procs is None or procs.poll() is not None:
                break
        try:
            progress_text = procs.stdout.readline()
            if not progress_text:
                break
            progress_text = progress_text.decode("utf-8").strip()
            if progress_text.startswith("frame="):
                frame = int(progress_text.split('=')[-1])
                q[0] = frame
        except Exception:
            break

完整修正后的代码 #

import subprocess as sp
import shlex
import json
import time
import threading

# 全局变量与锁
ffmpeg_process = None
process_lock = threading.Lock()

def start_ffmpeg_thread(audio_part, video_part, path):
    global ffmpeg_process
    with process_lock:
        if (ffmpeg_process is None) or ffmpeg_process.poll():
            # 获取总帧数
            data = sp.run(shlex.split(f'ffprobe -v error -select_streams v:0 -count_packets -show_entries stream=nb_read_packets -of json "{video_part}"'), stdout=sp.PIPE).stdout
            stream_info = json.loads(data)
            tot_n_frames = float(stream_info['streams'][0]['nb_read_packets'])

            # 启动ffmpeg进程
            ffmpeg_process = sp.Popen(
                shlex.split(f'ffmpeg -y -loglevel error -i "{video_part}" -i "{audio_part}" -progress pipe:1 "{path}"'),
                stdout=sp.PIPE, stdin=sp.PIPE, stderr=sp.PIPE
            )

            q = [0]
            stop_event = threading.Event()
            # 启动进度读取线程
            progress_thread = threading.Thread(target=ffmpeg_progress_reader, args=(ffmpeg_process, q, stop_event))
            progress_thread.start()

            # 进度循环
            while True:
                with process_lock:
                    if ffmpeg_process is None or ffmpeg_process.poll() is not None:
                        stop_event.set()
                        break
                # 更新进度
                progress_percent = (q[0]/tot_n_frames)*100
                print(f"Progress: [{progress_percent:.2f}%]", end="\r")
                time.sleep(0.1)  # 降低循环频率，减少CPU占用

            progress_thread.join()

def ffmpeg_progress_reader(procs, q, stop_event):
    while not stop_event.is_set():
        with process_lock:
            if procs is None or procs.poll() is not None:
                break
        try:
            progress_text = procs.stdout.readline()
            if not progress_text:
                break
            progress_text = progress_text.decode("utf-8").strip()
            if progress_text.startswith("frame="):
                frame = int(progress_text.split('=')[-1])
                q[0] = frame
        except Exception:
            break

def cancel_ffmpeg():
    time.sleep(10)
    global ffmpeg_process
    with process_lock:
        if (ffmpeg_process is not None) and (ffmpeg_process.poll() is None):
            # 发送终止命令
            ffmpeg_process.stdin.write('q'.encode("utf-8"))
            ffmpeg_process.stdin.flush()
            # 等待进程退出
            ffmpeg_process.wait()
            # 置空全局变量
            ffmpeg_process = None

# 启动取消线程与主进程
threading.Thread(target=cancel_ffmpeg, daemon=True).start()
start_ffmpeg_thread("aud.mp3", "vid.mp4", "output.mp4")

关键说明 #

• 线程锁：解决多线程同时读写全局变量的竞态问题，避免AttributeError。
• 替换communicate()：保留stdin/stdout直到进程正常退出，防止进度线程读取时的ValueError。
• 终止事件：让进度线程可以优雅退出，无需依赖进程状态的滞后判断。
• 降低循环频率：主进度循环加time.sleep(0.1)，减少不必要的CPU消耗。

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