Rust Tokio hyper 协程下载文件工具

根据 HTTP range requests 实现并发下载文件。

本文代码

Usage

cargo run --release <size> <uri> <file-path>

[##################################################] [任务 1 下载完成] [1.80 KiB/1.80 KiB] (0s)
[##################################################] [任务 2 下载完成] [1.80 KiB/1.80 KiB] (0s)
[##################################################] [任务 3 下载完成] [1.80 KiB/1.80 KiB] (0s)
[##################################################] [任务 4 下载完成] [1.80 KiB/1.80 KiB] (0s)
[##################################################] [合并文件完成] (0s)
耗时：698.633037ms

Cargo.toml

[package]
name = "download"
version = "0.1.0"
authors = ["Li Lei <this.lilei@gmail.com>"]
edition = "2021"[dependencies]
tokio = { version = "1.17.0", features = ["full"] }
hyper = { version = "0.14.18", features = ["full"] }
hyper-tls = "0.5.0"
anyhow = "1.0.56"
lazy_static = "1.4.0"
indicatif = "0.17.0-rc.10"[dependencies.clap]
version = "3.1.9"
default-features = false
features = ["std", "cargo", "derive"][dependencies.uuid]
version = "0.8.2"
features = ["v4"]

src/config.rs

use std::env::temp_dir;
use std::path::{Path, PathBuf};use anyhow::anyhow;
use clap::{crate_authors, crate_description, crate_name, crate_version, Arg, Command};
use hyper::Uri;
use uuid::Uuid;use crate::Result;pub struct Config {pub size: usize,pub uri: Uri,pub file_path: String,pub temp_file_dir: PathBuf,
}impl Config {pub fn get() -> Result<Self> {let matches = Command::new(crate_name!()).version(crate_version!()).author(crate_authors!()).about(crate_description!()).args(&[Arg::new("size").help("并发任务数量").required(true),Arg::new("uri").help("资源 URI").required(true),Arg::new("file-path").help("保存文件路径").required(true),]).get_matches();let size = matches.value_of_t("size")?;let uri = matches.value_of_t("uri")?;let file_path = matches.value_of_t("file-path")?;// 检查文件是否已存在if Path::new(&file_path).exists() {return Err(anyhow!("文件 `{}` 已存在", file_path));}let temp_file_dir = temp_dir().join(Uuid::new_v4().to_string());Ok(Self {size,uri,file_path,temp_file_dir,})}
}

src/http.rs

use std::time::Instant;use anyhow::anyhow;
use hyper::body::HttpBody;
use hyper::client::HttpConnector;
use hyper::header::{ACCEPT_RANGES, CONTENT_LENGTH};
use hyper::{Body, Client, Method, Request, Response};
use hyper_tls::HttpsConnector;
use indicatif::{MultiProgress, ProgressBar, ProgressStyle};
use lazy_static::lazy_static;
use tokio::fs::{create_dir, remove_dir_all, File, OpenOptions};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::spawn;
use tokio::task::JoinHandle;use crate::config::Config;
use crate::Result;lazy_static! {static ref CONFIG: Config = Config::get().unwrap();static ref PROGRESS: MultiProgress = MultiProgress::new();/// HTTPS 客户端static ref CLIENT: Client<HttpsConnector<HttpConnector>> =Client::builder().build(HttpsConnector::new());
}fn add_bar(size: u64, message: String, template: &str) -> Result<ProgressBar> {let bar = PROGRESS.add(ProgressBar::new(size));bar.set_style(ProgressStyle::default_bar().template(template)?.progress_chars("#>-"),);bar.set_message(message);Ok(bar)
}/// 下载文件进度条样式
fn add_download_bar(size: u64, message: String) -> Result<ProgressBar> {add_bar(size,message,"[{bar:50.cyan/blue}] [{msg}] [{bytes}/{total_bytes}] ({eta})",)
}/// 合并文件进度条样式
fn add_merge_bar(size: u64, message: String) -> Result<ProgressBar> {add_bar(size, message, "[{bar:50.magenta/cyan}] [{msg}] ({eta})")
}/// 下载文件
fn download_block(index: (usize, usize),start: usize,block_size: usize,bar: ProgressBar,
) -> JoinHandle<Result> {spawn(async move {let request = Request::builder().method(Method::GET).header("range",format!("bytes={}-{}", start, start + block_size - 1),).uri(&CONFIG.uri).body(Body::empty())?;let response = CLIENT.request(request).await?;write_file(response, index.0, &bar).await?;bar.finish_with_message(format!("任务 {} 下载完成", index.1));Ok(())})
}/// 写入文件
async fn write_file(mut response: Response<Body>, index: usize, bar: &ProgressBar) -> Result {let path_buf = CONFIG.temp_file_dir.join(index.to_string());// 数据流方式读取响应体let mut file = OpenOptions::new().create(true).append(true).open(path_buf).await?;while let Some(next) = response.data().await {let bytes = next?;bar.inc(bytes.len() as u64);file.write_all(&bytes).await?;}Ok(())
}/// 合并文件
async fn merge_file(size: u64) -> Result {let bar = add_merge_bar(size, "合并文件中".into())?;let mut file = OpenOptions::new().create(true).append(true).open(&CONFIG.file_path).await?;for i in 0..CONFIG.size {let mut block_file = File::open(CONFIG.temp_file_dir.join(i.to_string())).await?;let size = block_file.metadata().await?.len();const BUF_SIZE: u64 = 1024;let count = size / BUF_SIZE;let is_not_empty = count > 0;let mut first_buf_size = size % BUF_SIZE;if is_not_empty {first_buf_size += BUF_SIZE;}// count 为 0 时，第一个块获取 `余数` 个字节// count 大于 0 时，第一个块获取 `余数 + BUF_SIZE` 个字节let mut buffer = vec![0; first_buf_size as usize];block_file.read_exact(&mut buffer).await?;bar.inc(buffer.len() as u64);file.write_all(&buffer).await?;if is_not_empty {// 剩余块获取 `BUF_SIZE` 个字节let mut buffer = [0; BUF_SIZE as usize];for _ in 1..count {block_file.read_exact(&mut buffer).await?;bar.inc(BUF_SIZE);file.write_all(&buffer).await?;}}}bar.finish_with_message("合并文件完成");// 删除临时文件目录remove_dir_all(&CONFIG.temp_file_dir).await?;Ok(())
}pub async fn run() -> Result {let start = Instant::now();let request = Request::builder().method(Method::HEAD).uri(&CONFIG.uri).body(Body::empty())?;let response = CLIENT.request(request).await?;let headers = response.headers();let content_length = match headers.get(CONTENT_LENGTH) {None => return Err(anyhow!("{CONTENT_LENGTH} 为空")),Some(t) => t.to_str()?.parse::<usize>()?,};match headers.get(ACCEPT_RANGES) {None => return Err(anyhow!("不支持 {ACCEPT_RANGES} 请求")),Some(t) => {if t.to_str()? != "bytes" {return Err(anyhow!("不支持 {ACCEPT_RANGES} 请求"));}}};create_dir(&CONFIG.temp_file_dir).await?;// 单个任务下载的数据大小let block_size = content_length / CONFIG.size;let first_attach = content_length % CONFIG.size;let first_block_size = block_size + first_attach;let first_bar = add_download_bar(first_block_size as u64, "任务 1 下载中".into())?;// 第一个块获取 `block_size + 余数` 个字节let mut handles = vec![download_block((0, 1), 0, first_block_size, first_bar)];let bar_size = block_size as u64;// 剩余块获取 `block_size` 个字节for i in 1..CONFIG.size {let task_index = i + 1;let bar = add_download_bar(bar_size, format!("任务 {} 下载中", task_index))?;let start = i * block_size + first_attach;handles.push(download_block((i, task_index), start, block_size, bar));}// 等待所有任务结束for handle in handles {handle.await??;}merge_file(content_length as u64).await?;println!("耗时：{:?}", start.elapsed());Ok(())
}

src/main.rs

mod config;
mod http;use http::run;pub(crate) type Result<T = ()> = anyhow::Result<T>;#[tokio::main]
async fn main() -> Result {run().await?;Ok(())
}

Rust Tokio hyper 协程下载文件工具相关推荐

golang实现多协程下载文件(支持断点续传)
golang实现多协程下载文件(支持断点续传) 引言写这篇文章主要是周末休息太无聊,看了看别人代码,发现基本上要么是多协程下载文件要么就只有单协程的断点续传,所以就试了试有进度条的多协程下载文件(支 ...
linux上很方便的上传下载文件工具rz和sz
linux上很方便的上传下载文件工具rz和sz (本文适合linux入门的朋友) ######################################################### # ...
wget - Linux系统下载文件工具
wget - Linux系统下载文件工具 Linux系统下载文件工具补充说明 wget命令用来从指定的URL下载文件.wget非常稳定,它在带宽很窄的情况下和不稳定网络中有很强的适应性,如果是由 ...
python 实现多任务协程下载斗鱼平台图片
python 实现多任务协程下载斗鱼平台图片 import re import gevent from gevent import monkey, pool import time, random i ...
Retrofit 协程下载
demo: https://pan.baidu.com/s/1-u2Z7x9G19VPweK9VvwKtA 提取码:8zv2 整个下载过程需要用到Retrofit+协程+LiveData+Life ...
linux上很方便的上传下载文件工具rz和sz使用介绍
简单说就是,可以很方便地用这两个sz/rz工具,实现Linux下和Windows之间的文件传输(发送和接收),速度大概为10KB/s,适合中小文件.rz/sz 通过Zmodem协议传输数据一般来说, ...
Linux 命令之 wget -- 下载文件工具
文章目录一.命令介绍二.命令语法三.常用选项四.命令示例 (一)下载并以不同的文件名保存 (二)下载单个文件 (三)限速下载 (四)断点续传 (五)使用 wget 后台下载 (六)伪装代理名称 ...
linux用sz下载文件夹,linux上很方便的上传下载文件工具rz和sz使用介绍
一般来说,linux服务器大多是通过ssh客户端来进行远程的登陆和管理的,使用ssh登陆linux主机以后,如何能够快速的和本地机器进行文件的交互呢,也就是上传和下载文件到服务器和本地: 与ssh有关 ...
xshell的上传下载文件工具lrzsz
lrzsz lrzsz是一款xshell提供的一款小程序,在linux中可以代替ftp的上传和下载安装 yum install -y lrzsz.... Running transaction正在安 ...

Rust Tokio hyper 协程下载文件工具

Usage

Rust Tokio hyper 协程下载文件工具相关推荐

最新文章

热门文章