Rust中tokio::BufReader.next_line().await永久阻塞问题求助

核心原因分析 #

你遇到的tokio::BufReader永久阻塞问题，本质是wget的标准错误输出缓冲策略在非终端环境下的变化：

• 单独运行时，wget的stderr连接终端，默认使用行缓冲，输出会逐行立即刷新到流中；
• 但在futures stream的并行异步任务中，wget的stderr未连接终端，会切换为块缓冲，只有缓冲区填满或进程退出时才会输出数据，导致lines.next_line().await一直等待缓冲输出，陷入永久阻塞。
  另外，并行任务中若未同步等待子进程退出，也可能加剧流读取的挂起问题。

具体修复方案 #

1. 强制wget使用行缓冲输出

调用wget时，借助stdbuf命令强制stderr使用行缓冲，确保输出实时刷新：

use tokio::process::Command;

let mut cmd = Command::new("stdbuf");
cmd.arg("-e").arg("L") // 强制stderr采用行缓冲
    .arg("wget")
    .arg("--progress=bar:force") // 强制wget输出进度条格式
    .arg("目标下载URL")
    .arg("-O")
    .arg("本地输出路径");

若系统无stdbuf（如Windows），可直接给wget添加--progress=dot:giga参数，部分版本的wget会在非终端环境下强制行输出进度信息。

2. 同步处理stderr读取与进程退出

在download函数中，用tokio::join!同时启动stderr读取任务和进程等待任务，避免因进程未退出导致流读取阻塞：

use tokio::io::{BufReader, AsyncBufReadExt};
use tokio::process::Command;
use futures::future;

async fn download(url: &str, output_path: &str) -> Result<(), Box<dyn std::error::Error>> {
    let mut cmd = Command::new("stdbuf");
    cmd.arg("-e").arg("L")
        .arg("wget")
        .arg("--progress=bar:force")
        .arg(url)
        .arg("-O")
        .arg(output_path);

    let mut child = cmd.spawn()?;
    let stderr = child.stderr.take().expect("Failed to capture stderr");
    let mut reader = BufReader::new(stderr).lines();

    // 同时执行stderr读取和进程状态等待
    let (read_result, status) = future::join(
        async {
            while let Some(line) = reader.next_line().await? {
                // 这里插入你的indicatif进度条更新逻辑
                // 例如: progress_bar.set_position(提取到的进度值)
                println!("wget输出: {}", line);
            }
            Ok::<(), Box<dyn std::error::Error>>(())
        },
        child.wait()
    ).await;

    read_result?;
    if !status.success() {
        return Err("wget下载失败".into());
    }

    Ok(())
}

3. 限制并行任务数量

若并行数过高（如你代码中的0..1000），会导致系统资源耗尽，间接影响子进程的输出缓冲。建议用buffer_unordered限制并行任务数：

use futures::stream::{self, StreamExt};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let download_tasks = vec![("url1", "path1"), ("url2", "path2"), /* 更多任务 */];
    stream::iter(download_tasks)
        .map(|(url, path)| download(url, path))
        .buffer_unordered(10) // 限制并行数为10，避免资源过载
        .collect::<Vec<_>>()
        .await;

    Ok(())
}

验证建议 #

• 先单独测试修改后的download函数，确认stderr输出能被正常读取；
• 再逐步增加并行任务数，观察是否仍出现阻塞；
• 若问题依旧，可在stderr读取逻辑中添加超时处理，避免永久阻塞，同时排查共享资源锁是否存在隐性阻塞。

内容的提问来源于stack exchange，提问作者Hit and Run