如何用Rust的Tokio读取持续进程的stdout/err并传递参数

我最近在学习Rust，想实现异步运行外部命令，核心需求是：

• 实时读取子进程的stdout和stderr输出
• 按需向子进程发送输入消息

我不依赖Tokio，有更合适的实现方案也欢迎推荐。

目前我用Tokio的process::Command创建子进程，将输出转为流读取，但面对Python这类持续运行的进程时，无法获取stdout/stderr的输出。测试代码如下：

use std::process::Stdio;
use tokio_stream::StreamExt;
use tokio_util::codec::{FramedRead, LinesCodec, FramedWrite};

let mut process = tokio::process::Command::new("sh")
    .arg("-c")
    .arg("python3")
    .stdout(Stdio::piped())
    .stderr(Stdio::piped())
    .stdin(Stdio::piped())
    .spawn()
    .unwrap();

let stdout = FramedRead::new(process.stdout.take().unwrap(), LinesCodec::new())
    .map(|data| data.expect("fail on out!"));

let stderr = FramedRead::new(process.stderr.take().unwrap(), LinesCodec::new())
    .map(|data| data.expect("fail on err!"));

let mut stream = stdout.chain(stderr);

while let Some(msg) = stream.next().await {
    println!("{:?}", msg);
}

我尝试等待进程输出，但stdout和stderr始终为空。不过如果不重定向stderr/out/in，能正常获取Python的初始交互提示。

我的Cargo.toml依赖配置：

tokio = {version="1.26", features = ["rt", "macros", "rt-multi-thread", "fs", "process", "io-std", "io-util"]}
tokio-util = { version="0.7.7", features = ["codec", "io"]}
tokio-stream = "0.1.12"

问题原因与解决办法 #

1. 核心问题：Python的输出缓冲 #

Python在标准输出被重定向时（比如管道），会启用块缓冲模式，不会实时打印输出，只有当缓冲区满或进程退出时才会输出内容。这是你拿不到实时输出的主要原因。

解决方式：

• 启动Python时添加-u参数，强制使用无缓冲输出
• 或者设置环境变量PYTHONUNBUFFERED=1

2. 流处理优化：避免单流阻塞 #

你用stdout.chain(stderr)会先读完stdout再处理stderr，可能导致其中一个流的输出被阻塞。改用tokio::select!可以同时监听两个流，确保任何一个有输出都能及时处理。

3. 修改后的完整代码 #

use std::process::Stdio;
use tokio::select;
use tokio_stream::StreamExt;
use tokio_util::codec::{FramedRead, FramedWrite, LinesCodec};

#[tokio::main]
async fn main() {
    // 启动带无缓冲参数的Python进程
    let mut process = tokio::process::Command::new("python3")
        .arg("-u")
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .stdin(Stdio::piped())
        .spawn()
        .expect("Failed to spawn Python process");

    // 包装stdout和stderr为按行读取的流
    let mut stdout = FramedRead::new(process.stdout.take().unwrap(), LinesCodec::new());
    let mut stderr = FramedRead::new(process.stderr.take().unwrap(), LinesCodec::new());
    // 包装stdin为按行写入的流（用于给Python发消息）
    let mut stdin = FramedWrite::new(process.stdin.take().unwrap(), LinesCodec::new());

    // 异步同时处理输出和进程状态
    loop {
        select! {
            // 处理stdout输出
            Some(line) = stdout.next() => {
                match line {
                    Ok(content) => println!("STDOUT: {}", content),
                    Err(e) => eprintln!("Failed to read stdout: {}", e),
                }
            }
            // 处理stderr输出
            Some(line) = stderr.next() => {
                match line {
                    Ok(content) => println!("STDERR: {}", content),
                    Err(e) => eprintln!("Failed to read stderr: {}", e),
                }
            }
            // 监听进程退出状态
            status = process.wait() => {
                println!("Process exited with status: {:?}", status.unwrap());
                break;
            }
        }
    }

    // 示例：向Python发送命令（可以放到循环里按需调用）
    // stdin.send("print('Hello from Rust')".to_string()).await.unwrap();
}

4. 备选方案（不依赖Tokio） #

如果不想用Tokio，可以考虑：

• async-std：其process模块API设计与Tokio类似，同样支持异步子进程操作
• 同步+线程：用std::process::Command创建子进程，分别启动线程处理stdout/stderr的读取，实现伪异步效果

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