如何用PyAudio实现麦克风到扬声器的实时音频流并处理缓冲区？

问题描述 #

需要用PyAudio实现麦克风到扬声器的实时音频流传输，同时能读取、修改和处理传输过程中的采样缓冲区。已知：

• 回调模式下，输出流的回调函数需要“拉取”数据，输入流的回调函数需要“推送”数据
• 回调函数运行在独立线程中，官方文档明确：> Do not call Stream.read() or Stream.write() if using non-blocking operation.
• 尝试过用循环缓冲区传递数据，但出现连续触发三次麦克风回调后才触发三次扬声器回调的问题，无法正常工作。

假设麦克风和扬声器时钟同步，求标准实现方式。

解决方案 #

核心是用线程安全的环形缓冲区作为输入流与输出流的中间数据通道，解决跨线程的数据传递问题，同时在数据流转过程中插入处理逻辑。

1. 实现线程安全的环形缓冲区 #

环形缓冲区适合实时音频场景，能高效处理连续的数据流，同时通过锁机制保证线程安全：

import threading

class RingBuffer:
    def __init__(self, capacity):
        self.capacity = capacity
        self.buffer = bytearray(capacity)
        self.read_ptr = 0
        self.write_ptr = 0
        self.count = 0
        self.lock = threading.Lock()

    def write(self, data):
        with self.lock:
            # 计算可写入的最大长度
            available = self.capacity - self.count
            if available == 0:
                return 0  # 缓冲区满，丢弃数据
            write_len = min(len(data), available)
            
            # 分两段写入（处理环形绕回）
            if self.write_ptr + write_len <= self.capacity:
                self.buffer[self.write_ptr:self.write_ptr+write_len] = data[:write_len]
                self.write_ptr += write_len
            else:
                first_part = self.capacity - self.write_ptr
                self.buffer[self.write_ptr:] = data[:first_part]
                self.buffer[:write_len - first_part] = data[first_part:write_len]
                self.write_ptr = write_len - first_part
            
            self.count += write_len
            return write_len

    def read(self, length):
        with self.lock:
            if self.count == 0:
                return b''  # 缓冲区空，返回空数据
            read_len = min(length, self.count)
            result = bytearray(read_len)
            
            # 分两段读取
            if self.read_ptr + read_len <= self.capacity:
                result[:read_len] = self.buffer[self.read_ptr:self.read_ptr+read_len]
                self.read_ptr += read_len
            else:
                first_part = self.capacity - self.read_ptr
                result[:first_part] = self.buffer[self.read_ptr:]
                result[first_part:] = self.buffer[:read_len - first_part]
                self.read_ptr = read_len - first_part
            
            self.count -= read_len
            return bytes(result)

2. 配置PyAudio回调函数 #

输入回调负责采集麦克风数据、处理数据后写入环形缓冲区；输出回调负责从缓冲区读取数据、处理后输出到扬声器：

import pyaudio

# 音频参数（麦克风和扬声器需保持一致）
FORMAT = pyaudio.paInt16
CHANNELS = 1
RATE = 44100
CHUNK = 1024  # 回调缓冲区大小

# 初始化环形缓冲区（容量设为CHUNK的4倍，应对微小的时钟抖动）
ring_buffer = RingBuffer(CHUNK * 4)

def input_callback(in_data, frame_count, time_info, status):
    # 处理采集到的音频数据（示例：对数据做简单放大）
    processed_data = bytearray(in_data)
    for i in range(0, len(processed_data), 2):
        # 转换为16位整数，放大1.5倍，再转回字节
        sample = int.from_bytes(processed_data[i:i+2], byteorder='little', signed=True)
        sample = min(max(int(sample * 1.5), -32768), 32767)
        processed_data[i:i+2] = sample.to_bytes(2, byteorder='little', signed=True)
    
    # 写入环形缓冲区
    ring_buffer.write(processed_data)
    return (None, pyaudio.paContinue)

def output_callback(in_data, frame_count, time_info, status):
    # 从环形缓冲区读取数据
    data = ring_buffer.read(CHUNK * 2)  # paInt16每个采样占2字节
    
    # 如果缓冲区空，返回静音数据
    if len(data) < CHUNK * 2:
        data += b'\x00' * (CHUNK * 2 - len(data))
    
    # 可选：在这里再次处理输出数据
    # processed_out = process_output_data(data)
    
    return (data, pyaudio.paContinue)

def main():
    p = pyaudio.PyAudio()

    # 打开输入流
    input_stream = p.open(format=FORMAT,
                          channels=CHANNELS,
                          rate=RATE,
                          input=True,
                          frames_per_buffer=CHUNK,
                          stream_callback=input_callback)

    # 打开输出流
    output_stream = p.open(format=FORMAT,
                           channels=CHANNELS,
                           rate=RATE,
                           output=True,
                           frames_per_buffer=CHUNK,
                           stream_callback=output_callback)

    # 启动流
    input_stream.start_stream()
    output_stream.start_stream()

    # 保持运行直到用户按下Enter
    print("实时音频流已启动，按Enter停止...")
    input()

    # 停止并关闭流
    input_stream.stop_stream()
    input_stream.close()
    output_stream.stop_stream()
    output_stream.close()
    p.terminate()

if __name__ == "__main__":
    main()

关键说明 #

• 线程安全：环形缓冲区通过threading.Lock()保证读写操作的原子性，避免多线程环境下的数据竞争。
• 缓冲区容量：设置为回调缓冲区大小的2-4倍，可吸收麦克风和扬声器回调触发频率的微小差异（即使时钟同步，驱动调度也可能有抖动）。
• 数据处理：可以在输入回调（采集后）或输出回调（输出前）插入自定义处理逻辑，比如滤波、特效、音量调节等。
• 空缓冲区处理：输出回调如果读取到空数据，返回静音字节，避免扬声器出现爆音。

解决之前的问题 #

你之前遇到的“连续三次麦克风回调后才触发扬声器回调”，本质是没有线程安全的同步机制，且缓冲区大小设置不合理。环形缓冲区的锁机制会保证数据写入后能被及时读取，适当的容量也能平衡两端的回调频率差异。

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