基于asyncio的音频流发送问题：阻塞原因及优化方案咨询

1. 阻塞原因的判断 #

你的判断方向是对的，但本质原因更明确：
stream.read(CHUNK)是同步阻塞调用。在asyncio的事件循环线程中执行阻塞函数时，会彻底卡住事件循环，导致其他异步任务（比如处理speechData事件、socket.io心跳）完全无法运行，直到这个阻塞调用结束。
await asyncio.sleep(0)能临时解决问题，是因为它强制事件循环暂停当前协程，去处理其他待执行任务，但这只是个临时 workaround——stream.read()本身还是会阻塞事件循环，只是每次阻塞后给了其他任务运行的窗口。

2. 更规范的实现：用线程池异步处理阻塞IO #

比起依赖asyncio.sleep(0)，在asyncio中处理阻塞调用的标准方式是通过asyncio.run_in_executor()将阻塞操作卸载到单独线程（或进程），这样能让事件循环始终保持空闲，处理其他异步操作。

以下是优化后的代码示例：

import asyncio
import pyaudio
import socketio

sio = socketio.AsyncClient()

@sio.event
async def speechData(data):
    print("收到消息: ", data)

async def main():
    audio = pyaudio.PyAudio()
    FORMAT = pyaudio.paInt16
    CHANNELS = 1
    RATE = 16000
    CHUNK = int(RATE / 10)

    await sio.connect("http://localhost:9000")
    await sio.emit(
        "startGoogleCloudStream",
        {
            "audio": {
                "encoding": "LINEAR16",
                "sampleRateHertz": 16000,
                "languageCode": "pt-BR",
            },
            "interimResults": True,
        },
    )

    stream = audio.open(
        format=FORMAT,
        channels=CHANNELS,
        rate=RATE,
        input=True,
        frames_per_buffer=CHUNK,
    )

    # 获取当前事件循环，用内置线程池执行阻塞的read操作
    loop = asyncio.get_running_loop()
    try:
        while True:
            # 将阻塞的stream.read放到后台线程执行，不阻塞事件循环
            data = await loop.run_in_executor(None, stream.read, CHUNK)
            await sio.emit("binaryAudioData", data)
    finally:
        # 确保资源被正确释放
        stream.stop_stream()
        stream.close()
        audio.terminate()
        await sio.disconnect()

if __name__ == "__main__":
    asyncio.run(main())

关键改进点： #

• loop.run_in_executor(None, stream.read, CHUNK)：把阻塞的音频读取操作放到后台线程执行，事件循环在等待音频数据时可以继续处理其他任务（比如speechData事件）
• 新增finally块，保证音频流、PyAudio实例和socket.io连接能被优雅关闭
• 移除了await asyncio.sleep(0)这种临时 workaround

是否需要手动创建线程？ #

不需要手动创建线程。asyncio.run_in_executor()默认会使用内置的线程池来处理阻塞操作，这种方式和asyncio事件循环无缝集成，避免了手动管理线程的繁琐。如果需要更精细的线程控制（比如指定线程数量），可以自定义concurrent.futures.ThreadPoolExecutor并传入run_in_executor，但绝大多数场景下默认线程池足够用。

内容的提问来源于stack exchange，提问作者p.magalhaes