如何监控reqwest客户端上传进度并实现上传进度条

你不需要给reqwest增加额外的监控方法，只需要对传入的字节流做一层包装，在每次读取文件分片准备上传时更新进度条即可，逻辑和下载时从响应流取分片计数是一致的。

• 首先添加futures依赖到Cargo.toml，用于处理流映射：

[dependencies]
# 原有依赖保留，新增以下依赖
futures = "0.3"

• 然后修改上传函数，对原有的文件读取流做一层包装，每次读取分片时更新进度，同时给上传Body指定Content-Length确保服务端能正确识别文件大小：

use futures::StreamExt;
use std::cmp::min;

pub async fn upload_file(client: &Client, url: &str, path: &str) -> Result<(), String> {
    // 打开文件并获取总大小，仅打开一次避免资源浪费
    let file = tokio::fs::File::open(path)
        .await
        .map_err(|e| format!("打开文件失败: {}", e))?;
    let total_size = file
        .metadata()
        .await
        .map_err(|e| format!("获取文件元数据失败: {}", e))?
        .len();

    // Indicatif进度条初始化，和原有逻辑一致
    let pb = ProgressBar::new(total_size);
    pb.set_style(ProgressStyle::default_bar()
        .template("{msg}\n{spinner:.green} [{elapsed_precise}] [{wide_bar:.cyan/blue}] {bytes}/{total_bytes} ({bytes_per_sec}, {eta})")
        .map_err(|e| format!("进度条样式初始化失败: {}", e))?
        .progress_chars("#>-"));
    pb.set_message(format!("上传中 {}", url));

    // 构造文件读取流
    let stream = FramedRead::new(file, BytesCodec::new());

    // 包装流，每次读取分片时更新进度条
    let mut uploaded = 0u64;
    let pb_clone = pb.clone();
    let monitored_stream = stream.map(move |res| match res {
        Ok(bytes) => {
            let chunk_len = bytes.len() as u64;
            uploaded = uploaded.saturating_add(chunk_len);
            pb_clone.set_position(min(uploaded, total_size));
            Ok(bytes)
        }
        Err(e) => Err(e),
    });

    // 构造带Content-Length的请求体，避免服务端按chunked编码接收
    let mut body = Body::wrap_stream(monitored_stream);
    body = body.with_content_length(total_size);

    // 发送上传请求
    client
        .post(url)
        .body(body)
        .send()
        .await
        .map_err(|e| format!("上传请求发送失败: {}", e))?;

    pb.finish_with_message(format!("上传完成 {} 到 {}", path, url));
    Ok(())
}

reqwest的Body::wrap_stream方法会按顺序消费流中的每个字节分片，每成功消费一个分片就代表该部分数据已经被读取到上传队列中，因此在流的映射回调中计数，即可准确跟踪上传进度，和下载时从响应流读取分片更新进度的逻辑完全对齐。

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