FastAPI WebSocket调用Popen读取设备stdout：终端正常但无法实时推送到客户端

问题场景 #

搭建FastAPI WebSocket服务端与HTML/JS客户端，点击客户端按钮触发服务端执行cat /dev/ttyUSB0读取实时串口设备内容。终端打印输出正常，但WebSocket在程序中断前无任何推送；尝试单次调用process.stdout.readline()并推送有效，但循环推送失效。

核心原因 #

原代码中for output in process.stdout是同步阻塞式迭代，在FastAPI的异步事件循环中，这个循环会独占线程，导致后续的await websocket.send_text()无法被及时调度执行。事件循环被阻塞后，WebSocket的I/O操作得不到处理，自然看不到实时推送。而单次调用readline()时，执行完成后会让出事件循环，所以能成功推送一次。

解决方案 #

方案1：异步Subprocess（推荐） #

使用asyncio的异步子进程模块，全程异步读取输出，不会阻塞事件循环：

import asyncio
from fastapi import WebSocket
from subprocess import shlex

async def run_command(websocket: WebSocket):
    print("running command")
    command = "cat /dev/ttyUSB0"
    args = shlex.split(command)

    # 创建异步子进程
    process = await asyncio.create_subprocess_exec(
        *args,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.STDOUT
    )

    # 异步迭代输出流
    async for output in process.stdout:
        await websocket.send_text(output.decode().strip())
    
    await process.wait()

方案2：线程+队列（兼容同步操作） #

如果必须使用同步subprocess.Popen，可以把阻塞的读取操作放到单独线程，用队列传递数据，异步从队列取数推送：

import asyncio
from fastapi import WebSocket
from subprocess import Popen, PIPE, STDOUT
import shlex
from threading import Thread
from queue import Queue, Empty

async def run_command(websocket: WebSocket):
    print("running command")
    command = "cat /dev/ttyUSB0"
    args = shlex.split(command)
    output_queue = Queue()

    # 线程函数：同步读取输出并放入队列
    def read_device_output(process, queue):
        for line in iter(process.stdout.readline, b''):
            queue.put(line)
        process.stdout.close()

    with Popen(args, stdout=PIPE, stderr=STDOUT, bufsize=1) as process:
        # 启动线程读取串口输出
        read_thread = Thread(target=read_device_output, args=(process, output_queue))
        read_thread.start()

        # 异步循环从队列取数据推送
        while read_thread.is_alive() or not output_queue.empty():
            try:
                # 非阻塞取数，超时0.1秒避免占用事件循环
                line = output_queue.get(timeout=0.1)
                await websocket.send_text(line.decode().strip())
            except Empty:
                # 让出事件循环给其他任务
                await asyncio.sleep(0.01)
        read_thread.join()

客户端优化（可选） #

如果需要把推送内容显示到页面，修改客户端onmessage函数：

ws.onmessage = function(event) {
    console.log(event.data);
    const outputList = document.getElementById('output');
    const newItem = document.createElement('li');
    newItem.textContent = event.data;
    outputList.appendChild(newItem);
};

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