Google 会使用 AI 技术将内容翻译成您偏好的语言。AI 翻译可能包含错误。

加快 Cloud Storage 对象下载速度

通过单个连接依序下载 Cloud Storage 中的大型对象可能会导致可用网络带宽未得到充分利用。通过将对象划分为称为条带的字节范围段并并发读取这些段，您可以使用 Rust 版 Cloud 客户端库提高大型对象的下载吞吐量。

本文档介绍了如何使用对象组合为大型测试对象添加种子、使用 ReadRange::segment 并发下载字节范围，以及在固定偏移量处将条带写入本地文件。如果您不熟悉客户端库，请先参阅将 Cloud Storage 与 Rust 搭配使用。

准备工作

在运行本文档中的代码之前，请先完成以下前提条件：

创建或选择一个已启用结算功能的 Google Cloud 项目。
在项目中创建专用的测试 Cloud Storage 存储桶。
为您的环境配置应用默认凭证。
向您的主账号授予具有读取、写入、组合和删除存储桶中对象权限的 Identity and Access Management (IAM) 角色，例如 Storage Admin (roles/storage.admin)。

依赖项

将 Cloud Storage 客户端库和支持 crate 添加到 Cargo.toml 文件中：

cargo add google-cloud-storage google-cloud-gax anyhow bytes futures
cargo add tokio --features macros,rt-multi-thread

创建大型测试对象

如需测试条带化下载，而无需通过网络上传数千兆字节的数据，请上传一个小型种子对象，并使用 Cloud Storage 组合操作在服务器端串联副本。在接受 client、control 和 bucket_name（格式为 projects/_/buckets/BUCKET_ID）的异步 seed 函数中，完成以下步骤：

如需在组合请求中引用现有对象，请导入 SourceObject：

use google_cloud_storage::model::compose_object_request::SourceObject;

上传名为 1MiB.txt 的 1 MiB 种子对象：

let buffer = String::from_iter(('a'..='z').cycle().take(1024 * 1024));
let seed = client
    .write_object(bucket_name, "1MiB.txt", bytes::Bytes::from_owner(buffer))
    .set_if_generation_match(0)
    .send_unbuffered()
    .await?;
println!(
    "Uploaded object {}, size={}KiB",
    seed.name,
    seed.size / 1024
);

如需创建名为 32MiB.txt 的 32 MiB 对象，请在服务器上串联 32 个 1MiB.txt 副本：

let seed_32 = control
    .compose_object()
    .set_destination(Object::new().set_bucket(bucket_name).set_name("32MiB.txt"))
    .set_source_objects((0..32).map(|_| {
        SourceObject::new()
            .set_name(&seed.name)
            .set_generation(seed.generation)
    }))
    .set_if_generation_match(0)
    .with_idempotency(true)
    .send()
    .await?;
println!(
    "Created object {}, size={}MiB",
    seed.name,
    seed.size / (1024 * 1024)
);

如需创建名为 1GiB.txt 的 1 GiB 对象，请串联 32 个 32MiB.txt 副本：

let seed_1024 = control
    .compose_object()
    .set_destination(Object::new().set_bucket(bucket_name).set_name("1GiB.txt"))
    .set_source_objects((0..32).map(|_| {
        SourceObject::new()
            .set_name(&seed_32.name)
            .set_generation(seed_32.generation)
    }))
    .set_if_generation_match(0)
    .with_idempotency(true)
    .send()
    .await?;
println!(
    "Created object {}, size={}MiB",
    seed.name,
    seed.size / (1024 * 1024)
);

如需通过 1GiB.txt（2GiB.txt、4GiB.txt、8GiB.txt、16GiB.txt 和 32GiB.txt）构建 2 的幂次方测试对象，请重复执行组合操作：

for s in [2, 4, 8, 16, 32] {
    let name = format!("{s}GiB.txt");
    let target = control
        .compose_object()
        .set_destination(Object::new().set_bucket(bucket_name).set_name(&name))
        .set_source_objects((0..s).map(|_| {
            SourceObject::new()
                .set_name(&seed_1024.name)
                .set_generation(seed_1024.generation)
        }))
        .set_if_generation_match(0)
        .with_idempotency(true)
        .send()
        .await?;
    println!(
        "Created object {} size={} MiB",
        target.name,
        target.size / (1024 * 1024)
    );
}

协调并发条带下载

如需并行下载对象，请查询其大小和世代，预分配本地目标文件，并为每个条带生成一个 future。在接受 client、control、bucket_name、object_name、stripe_size 和 destination 的 download 函数中，完成以下步骤：

使用 StorageControl::get_object 提取对象元数据，以便检查对象总大小并将后续读取锁定到单个世代：

let metadata = control
    .get_object()
    .set_bucket(bucket_name)
    .set_object(object_name)
    .send()
    .await?;

创建目标文件并将其预分配到确切的对象大小，以便工作器任务可以在独立的字节偏移量处写入：
```
let file = Arc::new(std::fs::File::create(destination)?);
file.set_len(metadata.size as u64)?;
```

使用 stripe_size 作为 limit 计算每个条带偏移量，并为每个片段创建一个 write_stripe future，当对象大小不能被 limit 整除时，附加最终的部分条带：

let size = metadata.size as u64;
let limit = stripe_size as u64;
let count = size / limit;
let mut stripes = (0..count)
    .map(|i| write_stripe(client.clone(), &file, i * limit, limit, &metadata))
    .collect::<Vec<_>>();
if !size.is_multiple_of(limit) {
    stripes.push(write_stripe(
        client.clone(),
        &file,
        count * limit,
        limit,
        &metadata,
    ))
}

使用 futures::future::join_all 并发运行所有条带 future，并传播所有错误：

futures::future::join_all(stripes)
    .await
    .into_iter()
    .collect::<anyhow::Result<Vec<_>>>()?;

下载并写入单个条带

每个 write_stripe 任务都会从 Cloud Storage 读取单个字节范围，并将收到的数据块写入共享本地文件句柄中的相应偏移量。如需实现 write_stripe，请完成以下步骤：

导入 ReadRange 并定义 write_stripe 函数：

async fn write_stripe(
    client: Storage,
    file: &Arc<std::fs::File>,
    offset: u64,
    limit: u64,
    metadata: &Object,
) -> anyhow::Result<()> {
    use google_cloud_storage::model_ext::ReadRange;

使用 Storage::read_object 创建 ReadObject 请求构建器，并在按照以下步骤配置生成和读取范围选项后发送请求：
```
let mut reader = client
    .read_object(&metadata.bucket, &metadata.name)
    .send()
    .await?;
```
如需将所有条带固定到同一对象世代，并防止在其他进程覆盖对象时出现不一致的读取，请在调用 send 之前对请求构建器调用 set_generation：
```
.set_generation(metadata.generation)
```
如需将读取操作限制为条带的字节范围，请在请求构建器上使用 ReadRange::segment(offset, limit) 调用 set_read_range：
```
.set_read_range(ReadRange::segment(offset, limit))
```

从读取器流式传输数据块，并将每个数据块写入 tokio::task::spawn_blocking 内 current_pos 处的目标文件。由于 Rust 的标准库缺少跨平台位置写入方法，因此本示例在 Unix 上使用 FileExt::write_all_at，在 Windows 上使用 FileExt::seek_write：

let mut current_pos = offset;
while let Some(b) = reader.next().await.transpose()? {
    let chunk_len = b.len() as u64;
    let handle = file.clone();
    #[cfg(unix)]
    tokio::task::spawn_blocking(move || {
        use std::os::unix::fs::FileExt;
        handle.write_all_at(&b, current_pos)
    })
    .await??;
    #[cfg(windows)]
    tokio::task::spawn_blocking(move || {
        use std::os::windows::fs::FileExt;
        handle.seek_write(&b, current_pos)
    })
    .await??;
    current_pos += chunk_len;
}

性能考虑因素

有几个因素会决定条带化下载的有效吞吐量：

本地磁盘速度。将分块写入速度较慢的本地存储空间可能会成为瓶颈。使用高吞吐量本地 SSD 存储有助于确保磁盘 I/O 速度快于网络读取速度。
虚拟机 CPU 容量。Cloud Storage 会对传输中的对象数据进行加密，因此客户端会解密收到的每个数据块。确保您的虚拟机有足够的 vCPU 来处理跨条带的并发 TLS 解密。
存储桶和虚拟机位置。跨区域下载会引入广域网延迟并限制吞吐量。尽可能将虚拟机与存储桶置于同一区域。

您还可以尝试调整 stripe_size，以避免留下非常小的剩余条带或创建过多微小的范围请求。

完整代码

以下示例将 seed、download 和 write_stripe 函数组合成一个完整的模块：

use std::sync::Arc;

use google_cloud_gax::options::RequestOptionsBuilder;
use google_cloud_storage::client::Storage;
use google_cloud_storage::client::StorageControl;
use google_cloud_storage::model::Object;

async fn seed(client: Storage, control: StorageControl, bucket_name: &str) -> anyhow::Result<()> {
    use google_cloud_storage::model::compose_object_request::SourceObject;

    let buffer = String::from_iter(('a'..='z').cycle().take(1024 * 1024));
    let seed = client
        .write_object(bucket_name, "1MiB.txt", bytes::Bytes::from_owner(buffer))
        .set_if_generation_match(0)
        .send_unbuffered()
        .await?;
    println!(
        "Uploaded object {}, size={}KiB",
        seed.name,
        seed.size / 1024
    );

    let seed_32 = control
        .compose_object()
        .set_destination(Object::new().set_bucket(bucket_name).set_name("32MiB.txt"))
        .set_source_objects((0..32).map(|_| {
            SourceObject::new()
                .set_name(&seed.name)
                .set_generation(seed.generation)
        }))
        .set_if_generation_match(0)
        .with_idempotency(true)
        .send()
        .await?;
    println!(
        "Created object {}, size={}MiB",
        seed.name,
        seed.size / (1024 * 1024)
    );

    let seed_1024 = control
        .compose_object()
        .set_destination(Object::new().set_bucket(bucket_name).set_name("1GiB.txt"))
        .set_source_objects((0..32).map(|_| {
            SourceObject::new()
                .set_name(&seed_32.name)
                .set_generation(seed_32.generation)
        }))
        .set_if_generation_match(0)
        .with_idempotency(true)
        .send()
        .await?;
    println!(
        "Created object {}, size={}MiB",
        seed.name,
        seed.size / (1024 * 1024)
    );

    for s in [2, 4, 8, 16, 32] {
        let name = format!("{s}GiB.txt");
        let target = control
            .compose_object()
            .set_destination(Object::new().set_bucket(bucket_name).set_name(&name))
            .set_source_objects((0..s).map(|_| {
                SourceObject::new()
                    .set_name(&seed_1024.name)
                    .set_generation(seed_1024.generation)
            }))
            .set_if_generation_match(0)
            .with_idempotency(true)
            .send()
            .await?;
        println!(
            "Created object {} size={} MiB",
            target.name,
            target.size / (1024 * 1024)
        );
    }

    Ok(())
}

async fn download(
    client: Storage,
    control: StorageControl,
    bucket_name: &str,
    object_name: &str,
    stripe_size: usize,
    destination: &str,
) -> anyhow::Result<()> {
    let metadata = control
        .get_object()
        .set_bucket(bucket_name)
        .set_object(object_name)
        .send()
        .await?;

    let file = Arc::new(std::fs::File::create(destination)?);
    file.set_len(metadata.size as u64)?;
    let start = std::time::Instant::now();

    let size = metadata.size as u64;
    let limit = stripe_size as u64;
    let count = size / limit;
    let mut stripes = (0..count)
        .map(|i| write_stripe(client.clone(), &file, i * limit, limit, &metadata))
        .collect::<Vec<_>>();
    if !size.is_multiple_of(limit) {
        stripes.push(write_stripe(
            client.clone(),
            &file,
            count * limit,
            limit,
            &metadata,
        ))
    }

    futures::future::join_all(stripes)
        .await
        .into_iter()
        .collect::<anyhow::Result<Vec<_>>>()?;

    let elapsed = start.elapsed();
    let mib = metadata.size as f64 / (1024.0 * 1024.0);
    let bw = mib / elapsed.as_secs_f64();
    println!(
        "Completed {mib:.2} MiB download in {elapsed:?}, using {count} stripes, effective bandwidth = {bw:.2} MiB/s"
    );

    Ok(())
}

async fn write_stripe(
    client: Storage,
    file: &Arc<std::fs::File>,
    offset: u64,
    limit: u64,
    metadata: &Object,
) -> anyhow::Result<()> {
    use google_cloud_storage::model_ext::ReadRange;
    let mut reader = client
        .read_object(&metadata.bucket, &metadata.name)
        .set_generation(metadata.generation)
        .set_read_range(ReadRange::segment(offset, limit))
        .send()
        .await?;
    let mut current_pos = offset;
    while let Some(b) = reader.next().await.transpose()? {
        let chunk_len = b.len() as u64;
        let handle = file.clone();
        #[cfg(unix)]
        tokio::task::spawn_blocking(move || {
            use std::os::unix::fs::FileExt;
            handle.write_all_at(&b, current_pos)
        })
        .await??;
        #[cfg(windows)]
        tokio::task::spawn_blocking(move || {
            use std::os::windows::fs::FileExt;
            handle.seek_write(&b, current_pos)
        })
        .await??;
        current_pos += chunk_len;
    }
    Ok(())
}

当您在 32 MiB 对象上运行 download（stripe_size 为 1 MiB）时，输出类似于以下内容，报告下载大小、经过的时间、条带计数和有效带宽：

Completed 32.00 MiB download in 412.5ms, using 32 stripes, effective bandwidth = 77.58 MiB/s

清理

为避免因 seed 创建的测试对象导致您的 Google Cloud 账号产生费用，请在完成后删除测试对象并删除测试存储桶。同时移除本地目标文件。