Python音频流处理并发设计：如何保证录制无中断？

• Python版本：3.11
• 基于PySimpleGUI的音频流式处理应用，核心函数：
  • _start_adding：通过while True循环录制5秒音频（I/O操作），将录制帧发送至队列
  • _start_processing：基于队列迭代器的while True循环，从队列接收消息并执行CPU密集型计算，处理速度快于录制速度（5秒录制内容仅需2-3秒处理完成）
• 当前实现问题：将两个函数放在独立线程中（避免UI卡顿），但_start_processing线程会延迟录制线程的循环迭代，导致音频出现间隙

from queue import Queue
from threading import Thread
from typing import Callable


class ClosableQueue(Queue):
    SENTINEL = object()
    def close(self):
        self.put(self.SENTINEL)
    
    def __iter__(self):
        while True:
            item = self.get()
            try:
                if item is self.SENTINEL:
                    return
                yield item
            finally:
                self.task_done()

class StoppableWorker(Thread):
    def __init__(self, func: Callable, queue: ClosableQueue):
        super().__init__()
        self.func = func
        self.queue = queue
    
    def run(self):
        for item in self.queue:
            self.func(item)


def _start_adding():
    global stop_game_thread, message_queue
    chunk_time = 5
    while True:
        if stop_game_thread:
            message_queue.close()
            break
        frames = record.record_frames_for_seconds(seconds=chunk_time)
        message_queue.put(frames)


def _start_processing(message):
    frames = message
    do_cpu_operation_for_2_seconds(frames)


message_queue = ClosableQueue()
adding_thread = Thread(target=_start_adding)
processing_thread = StoppableWorker(_start_processing, message_queue)

有哪些解决方案？哪种方案能确保_start_adding持续无中断录制？使用协程实现_start_adding是否更合适？

将其中一个函数放入独立进程，让_start_adding能够无中断运行（仅受队列发送帧的耗时影响）

可行解决方案 #

1. 将CPU密集型任务移至独立进程 #

这是最直接有效的方案，Python的GIL（全局解释器锁）会导致同一进程内的线程无法真正并行执行CPU密集型任务——当_start_processing占用GIL时，_start_adding线程会被阻塞，进而影响录制连续性。把处理任务放到独立进程，能彻底绕开GIL限制。

实现要点：

• 使用multiprocessing.Queue替代线程队列（必须用进程安全的队列实现进程间通信）
• 将_start_processing放入multiprocessing.Process中运行
• 确保传递的音频帧数据可序列化（比如字节流、numpy数组等，需支持pickle序列化）

修改后的代码片段：

from multiprocessing import Process, Queue
from threading import Thread
from typing import Callable


class ClosableProcessQueue(Queue):
    SENTINEL = object()
    def close(self):
        self.put(self.SENTINEL)
    
    def __iter__(self):
        while True:
            item = self.get()
            if item is self.SENTINEL:
                return
            yield item

class StoppableWorkerProcess(Process):
    def __init__(self, func: Callable, queue: ClosableProcessQueue):
        super().__init__()
        self.func = func
        self.queue = queue
    
    def run(self):
        for item in self.queue:
            self.func(item)


# 初始化部分
message_queue = ClosableProcessQueue()
adding_thread = Thread(target=_start_adding)
processing_process = StoppableWorkerProcess(_start_processing, message_queue)

2. 提升录制线程优先级（仅作缓解） #

在Linux或Windows系统中，可以提升_start_adding线程的优先级，让系统更优先调度录制任务，减少被CPU密集型线程抢占的概率。但这种方法无法从根本上解决GIL的限制，只能临时缓解卡顿。

Windows平台示例：

import win32api
import win32process
import win32con

def set_high_priority(thread):
    handle = win32api.OpenThread(win32con.THREAD_ALL_ACCESS, True, thread.ident)
    win32process.SetThreadPriority(handle, win32con.THREAD_PRIORITY_HIGHEST)

# 启动录制线程后设置优先级
adding_thread.start()
set_high_priority(adding_thread)

3. 协程是否合适？ #

不合适。协程是单线程内的协作式调度，CPU密集型任务会阻塞整个线程，导致录制任务完全无法执行。_start_adding是I/O绑定任务，协程仅能在I/O等待时切换，但_start_processing的CPU密集操作会占用全部线程执行时间，反而让录制延迟更严重。

最优方案选择 #

将CPU密集型的_start_processing移至独立进程是最优解，它能让录制线程（I/O绑定）和处理进程（CPU绑定）真正并行执行，彻底避免GIL导致的录制线程阻塞，确保音频录制无间隙。

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