🦀 RustFS 实战入门：基于 S3 兼容 SDK 构建对象存储应用

🦀 RustFS 实战入门：基于 S3 兼容 SDK 构建对象存储应用

Photos provided by Unsplash OR Pexels

RustFS 实战教程：使用 Rust SDK 进行对象存储处理

本教程将带你从零开始，使用 Rust 语言和 RustFS 对象存储构建一个完整的文件处理应用。你将学习如何安装 RustFS、配置 S3 兼容客户端，并实现文件上传、下载、删除等核心操作。

一、RustFS 简介

RustFS 是一个开源的、与 S3 兼容的高性能分布式对象存储系统，采用 Rust 语言开发。它具备以下核心特性：

  • 高性能：基于 Rust 语言构建，兼具内存安全与极致性能，显著提升 I/O 效率
  • S3 兼容性：完全兼容 Amazon S3 API，可无缝对接现有 S3 生态工具和应用
  • 分布式架构：支持横向扩展与容错，适用于大规模数据湖和 AI 场景
  • 零拷贝优化：通过 io_uring 实现 Linux 下的零拷贝传输，大幅提升大文件处理效率

⚠️ 注意：RustFS 目前正处于快速开发阶段（Alpha 版本），建议在测试环境使用，生产环境请谨慎评估。

二、环境搭建

2.1 使用 Docker 快速部署 RustFS

推荐使用 Docker 方式启动 RustFS，这是最快捷的部署方式：

# 拉取 RustFS 镜像
docker pull rustfs/rustfs:latest

# 创建数据目录
mkdir -p /data/rustfs/{data,logs}

# 启动 RustFS 容器
docker run -d \
  --name rustfs \
  --restart unless-stopped \
  -p 9000:9000 \
  -p 9001:9001 \
  -v /data/rustfs/data:/data \
  -v /data/rustfs/logs:/logs \
  -e RUSTFS_ACCESS_KEY=rustfsadmin \
  -e RUSTFS_SECRET_KEY=rustfsadmin \
  -e RUSTFS_CONSOLE_ENABLE=true \
  rustfs/rustfs:latest

启动成功后，你可以通过以下地址访问：

2.2 验证服务状态

# 检查容器运行状态
docker ps | grep rustfs

# 查看服务日志
docker logs rustfs

三、创建 Rust 项目并配置 SDK

RustFS 兼容 S3 API，因此我们可以使用官方 aws-sdk-s3 来操作。

3.1 新建项目并添加依赖

cargo new rustfs-demo
cd rustfs-demo

编辑 Cargo.toml，添加以下依赖：

[package]
name = "rustfs-demo"
version = "0.1.0"
edition = "2021"

[dependencies]
aws-sdk-s3 = "1.0"
aws-config = "1.0"
tokio = { version = "1", features = ["full"] }
anyhow = "1.0"
bytes = "1.0"
futures = "0.3"

3.2 配置 S3 客户端

由于 RustFS 是本地部署的 S3 兼容服务，需要配置自定义端点和强制路径样式：

use aws_sdk_s3::{Client, config::Region, config::Credentials};
use aws_sdk_s3::primitives::ByteStream;
use aws_sdk_s3::types::{
    CreateBucketConfiguration,
    BucketLocationConstraint,
    ObjectCannedAcl,
};
use anyhow::Result;
use bytes::Bytes;
use tokio::fs::File;
use tokio::io::AsyncReadExt;
use futures::StreamExt;

pub struct RustFSClient {
    client: Client,
    bucket: String,
}

impl RustFSClient {
    /// 初始化 RustFS 客户端
    pub async fn new(endpoint: &str, access_key: &str, secret_key: &str, bucket: &str) -> Self {
        let credentials = Credentials::new(
            access_key,
            secret_key,
            None,
            None,
            "rustfs-demo",
        );
        
        let config = aws_config::from_env()
            .region(Region::new("us-east-1"))
            .endpoint_url(endpoint)
            .credentials_provider(credentials)
            .load()
            .await;
        
        // 强制使用路径样式（path-style），而非虚拟主机样式
        let client = Client::from_conf(
            aws_sdk_s3::config::Builder::from(&config)
                .force_path_style(true)
                .build()
        );
        
        Self {
            client,
            bucket: bucket.to_string(),
        }
    }
}

关于 force_path_style(true) 的作用：它让请求路径变为 http://endpoint/bucket/key 格式，而非虚拟主机样式 http://bucket.endpoint/key，这对于非 AWS 的 S3 兼容服务是必需的。

四、核心功能实战

4.1 创建存储桶（Bucket）

impl RustFSClient {
    /// 创建存储桶
    pub async fn create_bucket(&self) -> Result<()> {
        let create_config = CreateBucketConfiguration::builder()
            .location_constraint(BucketLocationConstraint::UsEast1)
            .build();
        
        self.client
            .create_bucket()
            .bucket(&self.bucket)
            .create_bucket_configuration(create_config)
            .send()
            .await?;
        
        println!("✅ 存储桶 '{}' 创建成功", self.bucket);
        Ok(())
    }
    
    /// 检查存储桶是否存在
    pub async fn bucket_exists(&self) -> Result<bool> {
        match self.client.head_bucket().bucket(&self.bucket).send().await {
            Ok(_) => Ok(true),
            Err(_) => Ok(false),
        }
    }
}

4.2 上传文件（支持大文件流式上传）

impl RustFSClient {
    /// 上传小文件（Bytes）
    pub async fn upload_bytes(&self, key: &str, data: Vec<u8>) -> Result<()> {
        let body = ByteStream::from(data);
        
        self.client
            .put_object()
            .bucket(&self.bucket)
            .key(key)
            .body(body)
            .send()
            .await?;
        
        println!("✅ 文件 '{}' 上传成功", key);
        Ok(())
    }
    
    /// 从本地文件流式上传（适合大文件）
    pub async fn upload_file(&self, key: &str, file_path: &str) -> Result<()> {
        let mut file = File::open(file_path).await?;
        let mut buffer = Vec::new();
        file.read_to_end(&mut buffer).await?;
        
        let body = ByteStream::from(buffer);
        
        self.client
            .put_object()
            .bucket(&self.bucket)
            .key(key)
            .body(body)
            .send()
            .await?;
        
        println!("✅ 文件 '{}' 上传成功（本地路径：{}）", key, file_path);
        Ok(())
    }
    
    /// 分块上传大文件（推荐 > 100MB）
    pub async fn upload_large_file(&self, key: &str, file_path: &str) -> Result<()> {
        use aws_sdk_s3::types::CompletedMultipartUpload;
        use aws_sdk_s3::types::CompletedPart;
        
        let file = File::open(file_path).await?;
        let file_size = file.metadata().await?.len();
        let chunk_size = 5 * 1024 * 1024; // 5MB 每块
        let num_parts = (file_size + chunk_size - 1) / chunk_size;
        
        // 1. 初始化分块上传
        let create_mpu = self.client
            .create_multipart_upload()
            .bucket(&self.bucket)
            .key(key)
            .send()
            .await?;
        
        let upload_id = create_mpu.upload_id().unwrap();
        println!("📦 开始分块上传，共 {} 块，UploadId: {}", num_parts, upload_id);
        
        // 2. 上传各个分块
        let mut completed_parts = Vec::new();
        let mut file = File::open(file_path).await?;
        
        for part_number in 1..=num_parts {
            let mut chunk_buffer = vec![0u8; chunk_size as usize];
            let bytes_read = file.read(&mut chunk_buffer).await?;
            chunk_buffer.truncate(bytes_read);
            
            let part_data = ByteStream::from(chunk_buffer);
            
            let upload_part_res = self.client
                .upload_part()
                .bucket(&self.bucket)
                .key(key)
                .upload_id(upload_id)
                .part_number(part_number as i32)
                .body(part_data)
                .send()
                .await?;
            
            let etag = upload_part_res.e_tag().unwrap().to_string();
            completed_parts.push(
                CompletedPart::builder()
                    .part_number(part_number as i32)
                    .e_tag(etag)
                    .build()
            );
            
            println!("  ✅ 分块 {} / {} 上传完成", part_number, num_parts);
        }
        
        // 3. 完成分块上传
        let completed_upload = CompletedMultipartUpload::builder()
            .set_parts(Some(completed_parts))
            .build();
        
        self.client
            .complete_multipart_upload()
            .bucket(&self.bucket)
            .key(key)
            .upload_id(upload_id)
            .multipart_upload(completed_upload)
            .send()
            .await?;
        
        println!("✅ 大文件 '{}' 分块上传完成", key);
        Ok(())
    }
}

4.3 下载文件

impl RustFSClient {
    /// 下载文件到内存（适合小文件）
    pub async fn download_bytes(&self, key: &str) -> Result<Vec<u8>> {
        let response = self.client
            .get_object()
            .bucket(&self.bucket)
            .key(key)
            .send()
            .await?;
        
        let data = response.body.collect().await?.into_bytes();
        println!("✅ 文件 '{}' 下载成功，大小：{} bytes", key, data.len());
        Ok(data.to_vec())
    }
    
    /// 下载文件并保存到本地
    pub async fn download_to_file(&self, key: &str, save_path: &str) -> Result<()> {
        let response = self.client
            .get_object()
            .bucket(&self.bucket)
            .key(key)
            .send()
            .await?;
        
        let data = response.body.collect().await?.into_bytes();
        tokio::fs::write(save_path, data).await?;
        
        println!("✅ 文件 '{}' 已保存至：{}", key, save_path);
        Ok(())
    }
    
    /// 流式下载大文件（分块处理，避免内存溢出）
    pub async fn download_large_stream(&self, key: &str, save_path: &str) -> Result<()> {
        let response = self.client
            .get_object()
            .bucket(&self.bucket)
            .key(key)
            .send()
            .await?;
        
        let mut file = tokio::fs::File::create(save_path).await?;
        let mut stream = response.body;
        
        while let Some(chunk) = stream.next().await {
            let chunk = chunk?;
            tokio::io::copy(&mut chunk.as_ref(), &mut file).await?;
        }
        
        println!("✅ 大文件 '{}' 流式下载完成", key);
        Ok(())
    }
}

4.4 列出文件

impl RustFSClient {
    /// 列出存储桶中的所有文件
    pub async fn list_objects(&self, prefix: Option<&str>) -> Result<Vec<String>> {
        let mut list_objects = self.client
            .list_objects_v2()
            .bucket(&self.bucket);
        
        if let Some(p) = prefix {
            list_objects = list_objects.prefix(p);
        }
        
        let response = list_objects.send().await?;
        
        let mut objects = Vec::new();
        if let Some(contents) = response.contents() {
            for obj in contents {
                if let Some(key) = obj.key() {
                    objects.push(key.to_string());
                }
            }
        }
        
        println!("📁 找到 {} 个文件", objects.len());
        for obj in &objects {
            println!("  - {}", obj);
        }
        
        Ok(objects)
    }
}

4.5 删除文件

impl RustFSClient {
    /// 删除单个文件
    pub async fn delete_object(&self, key: &str) -> Result<()> {
        self.client
            .delete_object()
            .bucket(&self.bucket)
            .key(key)
            .send()
            .await?;
        
        println!("🗑️ 文件 '{}' 已删除", key);
        Ok(())
    }
    
    /// 批量删除文件
    pub async fn delete_objects(&self, keys: Vec<&str>) -> Result<()> {
        let objects: Vec<aws_sdk_s3::types::ObjectIdentifier> = keys
            .iter()
            .map(|key| {
                aws_sdk_s3::types::ObjectIdentifier::builder()
                    .key(key)
                    .build()
                    .unwrap()
            })
            .collect();
        
        let delete = aws_sdk_s3::types::Delete::builder()
            .set_objects(Some(objects))
            .build();
        
        self.client
            .delete_objects()
            .bucket(&self.bucket)
            .delete(delete)
            .send()
            .await?;
        
        println!("🗑️ 已批量删除 {} 个文件", keys.len());
        Ok(())
    }
}

4.6 获取对象元数据

impl RustFSClient {
    /// 获取文件元数据
    pub async fn get_object_metadata(&self, key: &str) -> Result<ObjectMetadata> {
        let head = self.client
            .head_object()
            .bucket(&self.bucket)
            .key(key)
            .send()
            .await?;
        
        let metadata = ObjectMetadata {
            key: key.to_string(),
            size: head.content_length().unwrap_or(0),
            last_modified: head.last_modified().map(|d| d.to_string()),
            content_type: head.content_type().map(|s| s.to_string()),
            etag: head.e_tag().map(|s| s.to_string()),
        };
        
        println!("📄 文件元数据：{:#?}", metadata);
        Ok(metadata)
    }
}

#[derive(Debug)]
pub struct ObjectMetadata {
    pub key: String,
    pub size: i64,
    pub last_modified: Option<String>,
    pub content_type: Option<String>,
    pub etag: Option<String>,
}

五、完整实战示例

下面是一个完整的示例程序，演示了所有核心功能：

// src/main.rs
use rustfs_demo::RustFSClient;
use anyhow::Result;

#[tokio::main]
async fn main() -> Result<()> {
    // 1. 初始化客户端
    let client = RustFSClient::new(
        "http://localhost:9000",
        "rustfsadmin",
        "rustfsadmin",
        "my-test-bucket"
    ).await;
    
    // 2. 检查并创建存储桶
    if !client.bucket_exists().await? {
        client.create_bucket().await?;
    }
    
    // 3. 上传小文件
    client.upload_bytes("hello.txt", b"Hello, RustFS!".to_vec()).await?;
    
    // 4. 上传本地文件
    client.upload_file("example.txt", "./example.txt").await?;
    
    // 5. 列出所有文件
    let objects = client.list_objects(None).await?;
    
    // 6. 下载文件
    let content = client.download_bytes("hello.txt").await?;
    println!("下载内容：{}", String::from_utf8_lossy(&content));
    
    // 7. 获取文件元数据
    let metadata = client.get_object_metadata("hello.txt").await?;
    
    // 8. 删除文件
    client.delete_object("hello.txt").await?;
    
    println!("\n🎉 所有操作执行成功！");
    Ok(())
}

六、高级技巧与最佳实践

6.1 并发上传/下载

利用 tokio::spawn 实现并发操作，大幅提升吞吐量：

use futures::future::try_join_all;

async fn concurrent_upload(client: &RustFSClient, files: Vec<(&str, &str)>) -> Result<()> {
    let tasks = files.into_iter().map(|(key, path)| {
        tokio::spawn(async move {
            client.upload_file(key, path).await
        })
    });
    
    try_join_all(tasks).await?;
    Ok(())
}

6.2 错误处理与重试机制

use backoff::{future::retry, ExponentialBackoff};

async fn upload_with_retry(client: &RustFSClient, key: &str, data: Vec<u8>) -> Result<()> {
    let operation = || async {
        client.upload_bytes(key, data.clone()).await
            .map_err(|e| backoff::Error::transient(e))
    };
    
    retry(ExponentialBackoff::default(), operation).await?;
    Ok(())
}

6.3 内存优化建议

  • 大文件处理：使用 upload_large_file 分块上传，避免一次性加载整个文件到内存
  • 流式下载：使用 download_large_stream 边读边写，控制内存峰值
  • 内存池管理：RustFS 底层使用自定义 Arena 分配器减少动态内存分配

七、参考资料

资源类型链接/说明
RustFS 官方文档http://docs.rustfs.com
RustFS Docker 部署指南Docker 安装教程
AWS SDK for Rusthttps://github.com/awslabs/aws-sdk-rust
S3 API 兼容性说明RustFS Object Creation
性能优化详解RustFS 零拷贝与异步 I/O 实现
社区项目参考Milvus + RustFS RAG 应用

总结

本教程全面介绍了如何使用 Rust SDK 操作 RustFS 对象存储，涵盖：

  1. ✅ Docker 快速部署 RustFS 服务
  2. ✅ Rust SDK 的配置与初始化（含 force_path_style 关键配置）
  3. ✅ 存储桶管理、文件上传下载、批量删除等核心操作
  4. ✅ 大文件分块上传和流式下载的实现
  5. ✅ 并发处理、错误重试等高级技巧

RustFS 凭借 Rust 语言的内存安全和高性能特性，结合完善的 S3 API 兼容性，为 Rust 开发者提供了一个优雅的对象存储解决方案。你可以基于本教程的代码，快速构建自己的文件管理系统、备份工具或数据湖应用。

版权声明：自由转载-非商用-非衍生-保持署名（创意共享3.0许可证）

Similar Posts

🦀 Rustls 实战：Axum HTTPS 配置，零漏洞上云

🦀 Rustls 实战：Axum HTTPS 配置，零漏洞上云

anyhow Context 高阶：链式捕获现场数据，日志一次定位到行

anyhow Context 高阶：链式捕获现场数据，日志一次定位到行

🦀 Tonic-Prost-Build 实战：从零构建高性能 Rust gRPC 服务

🦀 Tonic-Prost-Build 实战：从零构建高性能 Rust gRPC 服务

优雅入门：使用 Deadpool 管理 Rust 中的连接池

优雅入门：使用 Deadpool 管理 Rust 中的连接池

秋风携新锈，Rust 1.91.0 悄然绽放

秋风携新锈，Rust 1.91.0 悄然绽放