作者：京东科技 贾世闻
对象存储是云的基础组件之一，各大云厂商都有相关产品。这里跟大家介绍一下rust与对象存储交到的基本套路和其中的一些技巧。
基本连接
我们以 [S3 sdk](
https://github.com/awslabs/aws-sdk-rust)为例来说说基本的连接与操作，作者验证过aws、京东云、阿里云。主要的增删改查功能没有什么差别。
- 建立客户端
1let shared_config = SdkConfig::builder() 2 .credentials_provider(SharedCredentialsProvider::new(Credentials::new( 3 "LTAI5t7NPuPKsXm6UeSa1", 4 "DGHuK03ESXQYqQ83buKMHs9NAwz", 5 None, 6 None, 7 "Static", 8 ))) 9 .endpoint_url("http://oss-cn-beijing.aliyuncs.com") 10 .region(Region::new("oss-cn-beijing")) 11 .build(); 12 let s3_config_builder = aws_sdk_s3::config::Builder::from(&shared_config); 13 let client = aws_sdk_s3::Client::from_conf(s3_config_builder.build()); 14
建立Client所需要的参数主要有你需要访问的oss的AK、SK，endpoint url 以及服务所在的区域。以上信息都可以在服务商的帮助文档查询到。
- 对象列表
1let mut obj_list = client 2 .list_objects_v2() 3 .bucket(bucket) 4 .max_keys(max_keys) 5 .prefix(prefix_str) 6 .continuation_token(token_str); 7 8let list = obj_list.send().await.unwrap(); 9println!("{:?}",list.contents()); 10println!("{:?}",list.next_continuation_token()); 11
使用list_objects_v2函数返回对象列表，相比list_objects函数，list_objects_v2可以通过continuation_token和max_keys控制返回列表的长度。list.contents()返回对象列表数组，
list.next_continuation_token()返回继续查询的token。
- 上传文件
1let content = ByteStream::from("content in file".as_bytes()); 2 let exp = aws_smithy_types::DateTime::from_secs(100); 3let upload = client 4 .put_object() 5 .bucket("bucket") 6 .key("/test/key") 7 .expires(exp) 8 .body(content); 9upload.send().await.unwrap(); 10
指定bucket及对象路径，body接受ByteStream类型作为文件内容，最后设置过期时间expires，无过期时间时不指定该配置即可。
- 下载文件
1let key = "/tmp/test/key".to_string(); 2let resp = client 3 .get_object() 4 .bucket("bucket") 5 .key(&key) 6 .send() 7 .await.unwrap(); 8let data = resp.body.collect().await.unwrap(); 9let bytes = data.into_bytes(); 10 11let path = std::path::Path::new("/tmp/key") 12if let Some(p) = path.parent() { 13 std::fs::create_dir_all(p).unwrap(); 14} 15let mut file = OpenOptions::new() 16 .write(true) 17 .truncate(true) 18 .create(true) 19 .open(path).unwrap(); 20let _ = file.write(&*bytes); 21file.flush().unwrap(); 22 23
通过get_object()函数获取GetObjectOutput。返回值的body 就是文件内容，将 body 转换为 bytes，最后打开文件写入即可。
- 删除文件
1let mut keys = vec![]; 2let key1 = ObjectIdentifier::builder() 3 .set_key(Some("/tmp/key1".to_string())) 4 .build(); 5let key2 = ObjectIdentifier::builder() 6 .set_key(Some("/tmp/key2".to_string())) 7 .build() 8keys.push(key1); 9keys.push(key2) 10client 11 .delete_objects() 12 .bucket(bucket) 13 .delete(Delete::builder().set_objects(Some(keys)).build()) 14 .send() 15 .await 16 .unwrap(); 17
delete_objects 批量删除对象。首先构建keys vector，定义要删除的对象，然后通过Delete::builder()，构建 Delete model。
大文件上传
1let mut file = fs::File::open("/tmp/file_name").unwrap(); 2let chunk_size = 1024*1024; 3let mut part_number = 0; 4let mut upload_parts: Vec = Vec::new(); 5 6//获取上传id 7let multipart_upload_res: CreateMultipartUploadOutput = self 8 .client 9 .create_multipart_upload() 10 .bucket("bucket") 11 .key("/tmp/key") 12 .send() 13 .await.unwrap(); 14let upload_id = match multipart_upload_res.upload_id() { 15 Some(id) => id, 16 None => { 17 return Err(anyhow!("upload id is None")); 18 } 19}; 20 21//分段上传文件并记录completer_part 22loop { 23 let mut buf = vec![0; chuck_size]; 24 let read_count = file.read(&mut buf)?; 25 part_number += 1; 26 27 if read_count == 0 { 28 break; 29 } 30 31 let body = &buf[..read_count]; 32 let stream = ByteStream::from(body.to_vec()); 33 34 let upload_part_res = self 35 .client 36 .upload_part() 37 .key(key) 38 .bucket(bucket) 39 .upload_id(upload_id) 40 .body(stream) 41 .part_number(part_number) 42 .send() 43 .await.unwrap(); 44 45 let completer_part = CompletedPart::builder() 46 .e_tag(upload_part_res.e_tag.unwrap_or_default()) 47 .part_number(part_number) 48 .build(); 49 50 upload_parts.push(completer_part); 51 52 if read_count != chuck_size { 53 break; 54 } 55} 56// 完成上传文件合并 57let completed_multipart_upload: CompletedMultipartUpload = 58 CompletedMultipartUpload::builder() 59 .set_parts(Some(upload_parts)) 60 .build(); 61 62let _complete_multipart_upload_res = self 63 .client 64 .complete_multipart_upload() 65 .bucket("bucket") 66 .key(key) 67 .multipart_upload(completed_multipart_upload) 68 .upload_id(upload_id) 69 .send() 70 .await.unwrap(); 71
有时候面对大文件，比如几百兆甚至几个G的文件，为了节约带宽和内存，我才采取分段上传的方案，然后在对象存储的服务端做合并。基本流程是：指定bucket和key，获取一个上传id；按流读取文件，分段上传字节流，并记录CompletedPart;通知服务器按照CompletedPart 集合来合并文件。具体过程代码已加注释，这里不再累述。
大文件下载
1let mut file = match OpenOptions::new() 2 .truncate(true) 3 .create(true) 4 .write(true) 5 .open("/tmp/target_file"); 6let key = "/tmp/test/key".to_string(); 7let resp = client 8 .get_object() 9 .bucket("bucket") 10 .key(&key) 11 .send() 12 .await.unwrap(); 13 14let content_len = resp.content_length(); 15let mut byte_stream_async_reader = resp.body.into_async_read(); 16let mut content_len_usize: usize = content_len.try_into().unwrap(); 17loop { 18 if content_len_usize > chunk_size { 19 let mut buffer = vec![0; chunk_size]; 20 let _ = byte_stream_async_reader.read_exact(&mut buffer).await.unwrap(); 21 file.write_all(&buffer).unwrap(); 22 content_len_usize -= chunk_size; 23 continue; 24 } else { 25 let mut buffer = vec![0; content_len_usize]; 26 let _ = byte_stream_async_reader.read_exact(&mut buffer).await.unwrap(); 27 file.write_all(&buffer).unwrap(); 28 break; 29 } 30} 31file.flush().unwrap(); 32
在从对象存储服务端下载文件的过程中也会遇到大文件问题。为了节约带宽和内存，我们采取读取字节流的方式分段写入文件。首先get_object()函数获取ByteStream，通过async_reader流式读取对象字节，分段写入文件。
对象存储的相关话题今天先聊到这儿，下期见。