如何在多线程中安全地使用 PyAudio

1. 背景介绍

在多线程环境下使用 PyAudio 可能会导致段错误（Segmentation Fault）或其他不可预期的行为。这是因为 PyAudio 在多线程环境下可能会出现资源冲突或线程安全问题。

PyAudio 是一个用于音频输入输出的 Python 库，它依赖于 PortAudio 库。在多线程环境下，如果多个线程同时创建和销毁 PyAudio 实例，可能会导致资源冲突，从而引发段错误。

在多线程环境下，尽量避免在每个线程中创建和销毁 PyAudio 实例。相反，应该在主线程中创建一个共享的 PyAudio 实例，并在子线程中使用它。

import pyaudio
import threading# 在主线程中创建共享的 PyAudio 实例
p = pyaudio.PyAudio()def audio_thread_function():# 在子线程中使用共享的 PyAudio 实例stream = p.open(format=pyaudio.paInt16,channels=1,rate=44100,input=True,frames_per_buffer=1024)# 音频处理逻辑stream.stop_stream()stream.close()# 创建并启动多个音频线程
threads = []
for _ in range(4):thread = threading.Thread(target=audio_thread_function)thread.start()threads.append(thread)# 等待所有线程完成
for thread in threads:thread.join()# 在主线程中释放 PyAudio 资源
p.terminate()

2. 实际问题和解决方案

2.1 问题代码

import time
import pyaudio
import numpy as np
import librosa
import soundfile as sf
import threading
import os# 查找设备索引
def find_device_index(device_name, is_input=True):p = pyaudio.PyAudio()device_count = p.get_device_count()for i in range(device_count):device_info = p.get_device_info_by_index(i)if device_name in device_info['name']:if is_input and device_info['maxInputChannels'] > 0:p.terminate()return ielif not is_input and device_info['maxOutputChannels'] > 0:p.terminate()return ip.terminate()raise ValueError(f"Device '{device_name}' not found or not a {'input' if is_input else 'output'} device.")# 播放音频文件
def play_audio(file_path, device_index, start_event):start_event.wait()  # 等待事件被设置audio_data, sr = librosa.load(file_path, sr=None)# 创建 PyAudio 实例p = pyaudio.PyAudio()# 打开流stream = p.open(format=pyaudio.paFloat32,channels=1,rate=sr,output=True,output_device_index=device_index)# 播放音频stream.write(audio_data.astype(np.float32).tobytes())# 关闭流stream.stop_stream()stream.close()p.terminate()# 录音
def record_audio(device_index, output_file, start_event, stop_event):start_event.wait()  # 等待事件被设置p = pyaudio.PyAudio()# 打开输入流stream = p.open(format=pyaudio.paFloat32,channels=1,rate=44100,input=True,input_device_index=device_index,frames_per_buffer=1024)print(f"Recording to {output_file}...")frames = []while not stop_event.is_set():  # 检查停止事件data = stream.read(1024)frames.append(data)print("Recording finished.")# 关闭流stream.stop_stream()stream.close()p.terminate()# 保存录音audio_data = b''.join(frames)audio_array = np.frombuffer(audio_data, dtype=np.float32)sf.write(output_file, audio_array, 44100)# 主程序
if __name__ == "__main__":# 根据设备名称查找设备索引CUBE_4NANO_DYNA_INDEX = find_device_index("Cube 4Nano Dyna", is_input=False)SOUNDMATRIX_A10_OUTPUT_INDEX = find_device_index("SoundMatrix A10", is_input=False)SOUNDMATRIX_A10_INPUT_INDEX = find_device_index("SoundMatrix A10", is_input=True)# 获取音频文件列表audio_file1_folder = './Soundplay'audio_file2_folder = './A10play'audio_file1_list = [f for f in os.listdir(audio_file1_folder) if f.endswith('.wav')]audio_file2_list = [f for f in os.listdir(audio_file2_folder) if f.endswith('.wav')]# 创建事件对象start_event = threading.Event()stop_event = threading.Event()# 遍历每一对组合for audio_file1 in audio_file1_list:for audio_file2 in audio_file2_list:file_path1 = os.path.join(audio_file1_folder, audio_file1)file_path2 = os.path.join(audio_file2_folder, audio_file2)print(file_path1, file_path2)# 创建线程play_thread1 = threading.Thread(target=play_audio, args=(file_path1, CUBE_4NANO_DYNA_INDEX, start_event))play_thread2 = threading.Thread(target=play_audio,args=(file_path2, SOUNDMATRIX_A10_OUTPUT_INDEX, start_event))output_file_name = f"{os.path.splitext(audio_file1)[0]}_{os.path.splitext(audio_file2)[0]}_soundmatrix.wav"output_file_path = os.path.join('./A10rec', output_file_name)record_thread = threading.Thread(target=record_audio, args=(SOUNDMATRIX_A10_INPUT_INDEX, output_file_path, start_event, stop_event))# 启动录音和播放线程record_thread.start()  # 启动录音线程play_thread1.start()  # 播放第一个音频play_thread2.start()  # 播放第二个音频# 设置事件，开始播放和录音start_event.set()# 等待播放线程完成play_thread1.join()play_thread2.join()# 设置停止事件，结束录音stop_event.set()record_thread.join()# 重置事件以便下次使用start_event.clear()stop_event.clear()print("All tasks completed.")

报错：

./Soundplay/cafeteria_SNR0_副本3.wav ./A10play/SER0.wav
Recording to ./A10rec/cafeteria_SNR0_副本3_SER0_soundmatrix.wav...
Recording finished.
./Soundplay/cafeteria_SNR0_副本2.wav ./A10play/SER0.wavProcess finished with exit code 139 (interrupted by signal 11:SIGSEGV)

2.2 解决方案：共享 PyAudio 实例

import time
import pyaudio
import numpy as np
import librosa
import soundfile as sf
import threading
import os# 查找设备索引
def find_device_index(device_name, is_input=True, p=None):if p is None:p = pyaudio.PyAudio()need_terminate = Trueelse:need_terminate = Falsedevice_count = p.get_device_count()for i in range(device_count):device_info = p.get_device_info_by_index(i)if device_name in device_info['name']:if is_input and device_info['maxInputChannels'] > 0:if need_terminate:p.terminate()return ielif not is_input and device_info['maxOutputChannels'] > 0:if need_terminate:p.terminate()return iif need_terminate:p.terminate()raise ValueError(f"Device '{device_name}' not found or not a {'input' if is_input else 'output'} device.")# 播放音频文件
def play_audio(file_path, device_index, start_event, p):start_event.wait()  # 等待事件被设置audio_data, sr = librosa.load(file_path, sr=None)# 打开流stream = p.open(format=pyaudio.paFloat32,channels=1,rate=sr,output=True,output_device_index=device_index)# 播放音频stream.write(audio_data.astype(np.float32).tobytes())# 关闭流stream.stop_stream()stream.close()# 录音
def record_audio(device_index, output_file, start_event, stop_event, p):start_event.wait()  # 等待事件被设置buffer_size = 2048# 打开输入流stream = p.open(format=pyaudio.paFloat32,channels=1,rate=44100,input=True,input_device_index=device_index,frames_per_buffer=buffer_size)print(f"Recording to {output_file}...")frames = []while not stop_event.is_set():  # 检查停止事件data = stream.read(buffer_size)frames.append(data)print("Recording finished.")# 关闭流stream.stop_stream()stream.close()# 保存录音audio_data = b''.join(frames)audio_array = np.frombuffer(audio_data, dtype=np.float32)sf.write(output_file, audio_array, 44100)# 主程序
if __name__ == "__main__":# 创建共享的 PyAudio 实例p = pyaudio.PyAudio()# 根据设备名称查找设备索引CUBE_4NANO_DYNA_INDEX = find_device_index("Cube 4Nano Dyna", is_input=False, p=p)SOUNDMATRIX_A10_OUTPUT_INDEX = find_device_index("SoundMatrix A10", is_input=False, p=p)SOUNDMATRIX_A10_INPUT_INDEX = find_device_index("SoundMatrix A10", is_input=True, p=p)# 获取音频文件列表audio_file1_folder = './Soundplay'audio_file2_folder = './A10play'audio_file1_list = [f for f in os.listdir(audio_file1_folder) if f.endswith('.wav')]audio_file2_list = [f for f in os.listdir(audio_file2_folder) if f.endswith('.wav')]# 创建事件对象start_event = threading.Event()stop_event = threading.Event()# 遍历每一对组合for audio_file1 in audio_file1_list:for audio_file2 in audio_file2_list:file_path1 = os.path.join(audio_file1_folder, audio_file1)file_path2 = os.path.join(audio_file2_folder, audio_file2)print(file_path1, file_path2)# 创建线程play_thread1 = threading.Thread(target=play_audio, args=(file_path1, CUBE_4NANO_DYNA_INDEX, start_event, p))play_thread2 = threading.Thread(target=play_audio,args=(file_path2, SOUNDMATRIX_A10_OUTPUT_INDEX, start_event, p))output_file_name = f"{os.path.splitext(audio_file1)[0]}_{os.path.splitext(audio_file2)[0]}_soundmatrix.wav"output_file_path = os.path.join('./A10rec', output_file_name)record_thread = threading.Thread(target=record_audio, args=(SOUNDMATRIX_A10_INPUT_INDEX, output_file_path, start_event, stop_event, p))# 启动录音和播放线程record_thread.start()  # 启动录音线程play_thread1.start()  # 播放第一个音频play_thread2.start()  # 播放第二个音频# 设置事件，开始播放和录音start_event.set()# 等待播放线程完成play_thread1.join()play_thread2.join()# 设置停止事件，结束录音stop_event.set()record_thread.join()# 重置事件以便下次使用start_event.clear()stop_event.clear()time.sleep(3)# 释放 PyAudio 资源p.terminate()