原文来源:https://tidb.net/blog/fd002ec8
notice”Rust is a trademark of the Mozilla Foundation in the US and other countries.”
对象存储是云的基础组件之一,各大云厂商都有相关产品。这里跟大家介绍一下 rust 与对象存储交到的基本套路和其中的一些技巧。
基本连接
我们以 aws 对象存储的 sdk 为例来说说基本的连接与操作,目前对象存储的客户端除了 aws 的还有一些第三方库,不过从功能和工程完整度上来讲 aws 的 rust 库最为优秀,而且公有云市场上的大部分对象存储产品都与其兼容。作者验证过 aws、京东云、阿里云。主要的增删改查功能没有什么差别。
建立 Client 所需要的参数主要有你需要访问的 oss 的 AK、SK,endpoint url 以及服务所在的区域。以上信息都可以在服务商的帮助文档查询到。
使用 list_objects_v2 函数返回对象列表,相比 list_objects 函数,list_objects_v2 可以通过 continuation_token 和 max_keys 控制返回列表的长度。list.contents() 返回对象列表数组,list.next_continuation_token() 返回继续查询的 token。
指定 bucket 及对象路径,body 接受 ByteStream 类型作为文件内容,最后设置过期时间 expires,无过期时间时不指定该配置即可。
通过 get_object() 函数获取 GetObjectOutput。返回值的 body 就是文件内容,将 body 转换为 bytes,最后打开文件写入即可。
delete_objects 批量删除对象。首先构建 keys vector,定义要删除的对象,然后通过 Delete::builder(),构建 Delete model。
大文件上传
let mut file = fs::File::open("/tmp/file_name").unwrap();
let chunk_size = 1024*1024;
let mut part_number = 0;
let mut upload_parts: Vec<CompletedPart> = Vec::new();
//获取上传id
let multipart_upload_res: CreateMultipartUploadOutput = self
.client
.create_multipart_upload()
.bucket("bucket")
.key("/tmp/key")
.send()
.await.unwrap();
let upload_id = match multipart_upload_res.upload_id() {
Some(id) => id,
None => {
return Err(anyhow!("upload id is None"));
}
};
//分段上传文件并记录completer_part
loop {
let mut buf = vec![0; chuck_size];
let read_count = file.read(&mut buf)?;
part_number += 1;
if read_count == 0 {
break;
}
let body = &buf[..read_count];
let stream = ByteStream::from(body.to_vec());
let upload_part_res = self
.client
.upload_part()
.key(key)
.bucket(bucket)
.upload_id(upload_id)
.body(stream)
.part_number(part_number)
.send()
.await.unwrap();
let completer_part = CompletedPart::builder()
.e_tag(upload_part_res.e_tag.unwrap_or_default())
.part_number(part_number)
.build();
upload_parts.push(completer_part);
if read_count != chuck_size {
break;
}
}
// 完成上传文件合并
let completed_multipart_upload: CompletedMultipartUpload =
CompletedMultipartUpload::builder()
.set_parts(Some(upload_parts))
.build();
let _complete_multipart_upload_res = self
.client
.complete_multipart_upload()
.bucket("bucket")
.key(key)
.multipart_upload(completed_multipart_upload)
.upload_id(upload_id)
.send()
.await.unwrap();
复制代码
有时候面对大文件,比如几百兆甚至几个 G 的文件,为了节约带宽和内存,我才采取分段上传的方案,然后在对象存储的服务端做合并。基本流程是:指定 bucket 和 key,获取一个上传 id;按流读取文件,分段上传字节流,并记录 CompletedPart; 通知服务器按照 CompletedPart 集合来合并文件。具体过程代码已加注释,这里不再累述。
大文件下载
let mut file = match OpenOptions::new()
.truncate(true)
.create(true)
.write(true)
.open("/tmp/target_file");
let key = "/tmp/test/key".to_string();
let resp = client
.get_object()
.bucket("bucket")
.key(&key)
.send()
.await.unwrap();
let content_len = resp.content_length();
let mut byte_stream_async_reader = resp.body.into_async_read();
let mut content_len_usize: usize = content_len.try_into().unwrap();
loop {
if content_len_usize > chunk_size {
let mut buffer = vec![0; chunk_size];
let _ = byte_stream_async_reader.read_exact(&mut buffer).await.unwrap();
file.write_all(&buffer).unwrap();
content_len_usize -= chunk_size;
continue;
} else {
let mut buffer = vec![0; content_len_usize];
let _ = byte_stream_async_reader.read_exact(&mut buffer).await.unwrap();
file.write_all(&buffer).unwrap();
break;
}
}
file.flush().unwrap();
复制代码
在从对象存储服务端下载文件的过程中也会遇到大文件问题。为了节约带宽和内存,我们采取读取字节流的方式分段写入文件。首先 get_object() 函数获取 ByteStream,通过 async_reader 流式读取对象字节,分段写入文件。
对象存储的相关话题今天先聊到这儿,下期见。
评论