如何在Python中限制Google Cloud Storage的Blob下载速率？

官方google-cloud-storage库和GCSFS确实没有内置的下载速率限制功能，你提到的分块下载思路是可行的，但可以通过更高效的方式实现：

1. 推荐方案：流式下载+自定义速率限制文件类 #

相比手动分块请求，通过流式下载配合自定义的速率限制包装类，能复用TCP连接、减少请求开销，且速率控制更平滑。实现方式如下：

import time
import io
from google.cloud import storage

class RateLimitedFile(io.BufferedWriter):
    def __init__(self, underlying_file, max_bytes_per_sec):
        super().__init__(underlying_file)
        self.max_bytes_per_sec = max_bytes_per_sec
        self._last_write_ts = time.time()
        self._bytes_since_last_check = 0

    def write(self, byte_data):
        current_ts = time.time()
        elapsed = current_ts - self._last_write_ts

        # 计算当前已写字节对应的理论耗时，若实际耗时不足则等待
        if elapsed > 0:
            expected_time = self._bytes_since_last_check / self.max_bytes_per_sec
            if expected_time > elapsed:
                time.sleep(expected_time - elapsed)
        
        # 重置计时与计数
        self._last_write_ts = time.time()
        self._bytes_since_last_check = len(byte_data)
        return super().write(byte_data)

# 使用示例
client = storage.Client()
bucket = client.bucket("你的存储桶名称")
blob = bucket.blob("目标Blob路径")

with open("本地保存路径", "wb") as local_file:
    # 限制为1MB/s，可根据需求调整
    rate_limited_stream = RateLimitedFile(local_file, max_bytes_per_sec=1024*1024)
    blob.download_to_file(rate_limited_stream)

2. 备选方案：分块请求+节流控制 #

如果坚持使用分块下载的思路，建议采用GCS推荐的分块大小（如64MB），并通过节流函数控制请求间隔，避免过多小请求带来的开销：

import time
from google.cloud import storage

def throttle(rate_bytes_per_sec):
    def decorator(func):
        last_request_ts = 0
        def wrapper(*args, **kwargs):
            nonlocal last_request_ts
            current_ts = time.time()
            elapsed = current_ts - last_request_ts
            # 计算当前分块的字节数
            chunk_size = kwargs["end"] - kwargs["start"] + 1
            required_time = chunk_size / rate_bytes_per_sec
            
            if required_time > elapsed:
                time.sleep(required_time - elapsed)
            
            result = func(*args, **kwargs)
            last_request_ts = time.time()
            return result
        return wrapper
    return decorator

# 初始化客户端与Blob
client = storage.Client()
bucket = client.bucket("你的存储桶名称")
blob = bucket.blob("目标Blob路径")
blob_size = blob.size
chunk_size = 64 * 1024 * 1024  # 64MB分块
rate_limit = 1024 * 1024  # 1MB/s限制

# 为下载方法添加节流控制
throttled_download = throttle(rate_limit)(blob.download_as_bytes)

# 分块下载并写入本地文件
with open("本地保存路径", "wb") as local_file:
    for start in range(0, blob_size, chunk_size):
        end = min(start + chunk_size - 1, blob_size - 1)
        chunk_data = throttled_download(start=start, end=end)
        local_file.write(chunk_data)

补充说明 #

• GCS服务端本身不提供下载速率限制的配置，所有速率控制都需要在客户端实现。
• 流式下载方案的性能优于分块请求，因为它避免了多次HTTP请求的连接开销，更适合大文件下载场景。

内容的提问来源于stack exchange，提问作者Theo