Google 會運用 AI 技術將內容翻譯成你偏好的語言，但可能會出錯。

加快 Cloud Storage 物件下載速度

在 Cloud Storage 中，透過單一連線依序下載大型物件，可能會導致可用的網路頻寬閒置。將物件分割成稱為「條帶」的位元組範圍區隔，並同時讀取這些區隔，即可使用 Rust 適用的 Cloud 用戶端程式庫，提高大型物件的下載處理量。

本文說明如何使用物件組合植入大型測試物件、使用 ReadRange::segment 同時下載位元組範圍，以及在固定偏移量將條帶寫入本機檔案。如果您不熟悉用戶端程式庫，請先參閱「搭配使用 Cloud Storage 和 Rust」。

事前準備

執行本文件中的程式碼前，請先完成下列必要條件：

建立或選取已啟用計費功能的 Google Cloud 專案。
在專案中建立專屬的測試 Cloud Storage bucket。
為環境設定應用程式預設憑證。
授予主體 Identity and Access Management (IAM) 角色，讓主體有權讀取、寫入、建立及刪除 bucket 中的物件，例如 Storage 管理員 (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

建立大型測試物件

如要測試分段下載，但不想透過網路上傳數 GB 的資料，請上傳小型種子物件，並使用 Cloud Storage compose 作業在伺服器端串連副本。在接受 client、control 和 bucket_name (格式為 projects/_/buckets/BUCKET_ID) 的非同步 seed 函式中，完成下列步驟：

如要在 compose 要求中參照現有物件，請匯入 SourceObject：
```
use google_cloud_storage::model::compose_object_request::SourceObject;
```

上傳 1 MiB 的種子物件，並命名為 1MiB.txt：

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?;

建立目的地檔案，並預先分配確切的物件大小，讓 worker 工作能在獨立的位元組偏移量寫入：
```
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 同時執行所有 Stripe 期貨，並傳播任何錯誤：

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)
```
如要將讀取作業限制在 Stripe 的位元組範圍內，請在要求建構工具上使用 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 速度快於網路讀取速度。
VM CPU 容量。Cloud Storage 會加密傳輸中的物件資料，因此用戶端會解密收到的每個區塊。請確保 VM 有足夠的 vCPU，可處理跨條帶的並行 TLS 解密作業。
bucket 和 VM 位置。跨區域下載會導致廣域網路延遲，並限制處理量。盡可能將 VM 與 bucket 放在相同區域。

您也可以嘗試調整 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，並使用 1 MiB stripe_size 時，輸出內容會與下列內容相似，回報下載大小、經過時間、條帶數量和有效頻寬：

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

清除所用資源

為避免系統向您的 Google Cloud 帳戶收取 seed 建立的測試物件費用，請在完成後刪除測試物件和刪除測試 bucket。一併移除本機目的地檔案。