文盘Rust —— rust连接oss | 京东云技术团队

rust,连接,oss,京东,技术,团队 · 浏览次数 : 155

小编点评

**基本连接与操作** 1. 获取 SdkConfig 配置对象。 2. 建立 SdkClient 对象。 3. 获取对象列表。 4. 使用 `put_object` 方法上传文件。 5. 使用 `get_object` 方法下载文件。 6. 使用 `delete_objects` 方法删除文件。 **技巧** * 使用 `list_objects_v2` 方法可以控制返回列表的长度。 * 使用 `upload` 方法可以进行分段上传。 * 使用 `delete_objects` 方法可以批量删除文件。 * 在从对象存储服务端下载文件的过程中,可以采用分段写入的方式节约带宽和内存。

正文

作者:京东科技 贾世闻

对象存储是云的基础组件之一,各大云厂商都有相关产品。这里跟大家介绍一下rust与对象存储交到的基本套路和其中的一些技巧。

基本连接

我们以 [S3 sdk](
https://github.com/awslabs/aws-sdk-rust)为例来说说基本的连接与操作,作者验证过aws、京东云、阿里云。主要的增删改查功能没有什么差别。

  • 建立客户端
let shared_config = SdkConfig::builder()
         .credentials_provider(SharedCredentialsProvider::new(Credentials::new(
            "LTAI5t7NPuPKsXm6UeSa1",
            "DGHuK03ESXQYqQ83buKMHs9NAwz",
             None,
             None,
             "Static",
         )))
         .endpoint_url("http://oss-cn-beijing.aliyuncs.com")
         .region(Region::new("oss-cn-beijing"))
         .build();
     let s3_config_builder = aws_sdk_s3::config::Builder::from(&shared_config);
     let client = aws_sdk_s3::Client::from_conf(s3_config_builder.build());

建立Client所需要的参数主要有你需要访问的oss的AK、SK,endpoint url 以及服务所在的区域。以上信息都可以在服务商的帮助文档查询到。

  • 对象列表
let mut obj_list = client
     .list_objects_v2()
     .bucket(bucket)
     .max_keys(max_keys)
     .prefix(prefix_str)
     .continuation_token(token_str);

let list = obj_list.send().await.unwrap();
println!("{:?}",list.contents());
println!("{:?}",list.next_continuation_token());

使用list_objects_v2函数返回对象列表,相比list_objects函数,list_objects_v2可以通过continuation_token和max_keys控制返回列表的长度。list.contents()返回对象列表数组,
list.next_continuation_token()返回继续查询的token。

  • 上传文件
let content = ByteStream::from("content in file".as_bytes());
 let exp = aws_smithy_types::DateTime::from_secs(100);
let upload = client
    .put_object()
    .bucket("bucket")
    .key("/test/key")
    .expires(exp)
    .body(content);
upload.send().await.unwrap();

指定bucket及对象路径,body接受ByteStream类型作为文件内容,最后设置过期时间expires,无过期时间时不指定该配置即可。

  • 下载文件
let key = "/tmp/test/key".to_string();
let resp = client
    .get_object()
    .bucket("bucket")
    .key(&key)
    .send()
    .await.unwrap();
let data = resp.body.collect().await.unwrap();
let bytes = data.into_bytes();

let path = std::path::Path::new("/tmp/key")
if let Some(p) = path.parent() {
    std::fs::create_dir_all(p).unwrap();
}
let mut file = OpenOptions::new()
    .write(true)
    .truncate(true)
    .create(true)
    .open(path).unwrap();
let _ = file.write(&*bytes);
file.flush().unwrap();


通过get_object()函数获取GetObjectOutput。返回值的body 就是文件内容,将 body 转换为 bytes,最后打开文件写入即可。

  • 删除文件
let mut keys = vec![];
let key1 = ObjectIdentifier::builder()
    .set_key(Some("/tmp/key1".to_string()))
    .build();
let key2 = ObjectIdentifier::builder()
    .set_key(Some("/tmp/key2".to_string()))
    .build()
keys.push(key1);
keys.push(key2)
client
    .delete_objects()
    .bucket(bucket)
    .delete(Delete::builder().set_objects(Some(keys)).build())
    .send()
    .await
    .unwrap();

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 = 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流式读取对象字节,分段写入文件。

对象存储的相关话题今天先聊到这儿,下期见。

与文盘Rust —— rust连接oss | 京东云技术团队相似的内容:

文盘Rust —— rust连接oss | 京东云技术团队

对象存储是云的基础组件之一,各大云厂商都有相关产品。这里跟大家介绍一下rust与对象存储交到的基本套路和其中的一些技巧。

文盘Rust -- rust 连接云上数仓 starwift

最近想看看 rust 如何集成 clickhouse,又犯了好吃懒做的心理(不想自己建环境),刚好京东云发布了兼容ck 的云原生数仓 Starwfit,于是搞了个实例折腾一番。 Starwfit 是京东云自主研发的新一代云原生数据仓库,通过存算分离降低了存储成本,同时兼具性能和扩展弹性。其写入和查询速度可达到传统数据仓库的数倍,为用户提供实时数据分析能力。广泛应用于流量分析、精准营销、用户画像、广

文盘Rust -- 安全连接 TiDB/Mysql

最近在折腾rust与数据库集成,选了Tidb Cloud Serverless Tier 作为数据源。Tidb 无疑是近五年来最优秀的国产开源分布式数据库,Tidb Cloud Serverless Tier作为pingcap旗下的云产品方便又经济,这次使用还有一些小惊喜。

文盘Rust -- 领域交互模式如何实现

书接上文,上回说到如何通过interactcli-rs四步实现一个命令行程序。但是 shell 交互模式在有些场景下用户体验并不是很好。比如我们要连接某个服务,比如 mysql 或者 redis 这样的服务。如果每次交互都需要输入地址、端口、用户名等信息,交互起来太麻烦。通常的做法是一次性输入和连接相关的信息或者由统一配置文件进行管理,然后进入领域交互模式,所有的命令和反馈都和该领域相关。inte

文盘Rust -- r2d2 实现redis连接池

作者:贾世闻 我们在开发应用后端系统的时候经常要和各种数据库、缓存等资源打交道。这一期,我们聊聊如何访问redis 并将资源池化。 在一个应用后端程序访问redis主要要做的工作有两个,单例和池化。 在后端应用集成redis,我们主要用到以下几个crate:​ ​once_cell​​​、​ ​re

文盘Rust -- tonic-Rust grpc初体验

gRPC 是开发中常用的开源高性能远程过程调用(RPC)框架,tonic 是基于 HTTP/2 的 gRPC 实现,专注于高性能、互操作性和灵活性。该库的创建是为了对 async/await 提供一流的支持,并充当用 Rust 编写的生产系统的核心构建块。今天我们聊聊通过使用tonic 调用grpc的的具体过程。

文盘Rust -- FFI 浅尝

rust FFI 是rust与其他语言互调的桥梁,通过FFI rust 可以有效继承 C 语言的历史资产。本期通过几个例子来聊聊rust与C 语言交互的具体步骤

文盘Rust -- 本地库引发的依赖冲突

clickhouse 的原生 rust 客户端目前比较好的有两个clickhouse-rs 和 clickhouse.rs 。两个库在单独使用时没有任何问题,但是,在同一工程同时引用时会报错。本篇内容主要讲解如何用rust语言解决本地库引发的依赖冲突问题

文盘Rust -- 用Tokio实现简易任务池

Tokio 无疑是 Rust 世界中最优秀的异步Runtime实现。非阻塞的特性带来了优异的性能,但是在实际的开发中我们往往需要在某些情况下阻塞任务来实现某些功能。

文盘Rust -- 把程序作为守护进程启动

当我们写完一个服务端程序,需要上线部署的时候,或多或少都会和操作系统的守护进程打交道,毕竟谁也不希望shell关闭既停服。今天我们就来聊聊这个事儿。 最早大家部署应用的通常操作是 “nohup xxxx &”,别说像weblogic 或者其他java 容器有启动脚本,里面其实也差不多;很喜欢 ngi