如何在boto3的upload_part中限制带宽并实现断点续传？

boto3的upload_part方法确实不支持TransferConfig里的max_bandwidth参数——这个配置是高层transfer manager（也就是upload_file依赖的逻辑）封装的功能，底层API没有对应的参数，强行传入会触发ParamValidationError。要实现断点续传+带宽限制，你需要在自定义分片上传的代码里手动控制数据传输速率，具体方案如下：

核心思路 #

直接在读取文件分片的环节做带宽限制，通过控制每次读取的字节数和等待时间，确保整体上传速率不超过设定值。不需要修改boto3 client的属性，只需要包装文件读取对象即可。

自定义带宽限制包装器 #

写一个简单的类来包装文件对象，实现速率控制：

import time

class BandwidthLimiter:
    def __init__(self, file_obj, max_bandwidth):
        self.file_obj = file_obj
        self.max_bytes_per_sec = max_bandwidth  # 单位：字节/秒
        self.last_read_time = time.time()
        self.bytes_read_since_last = 0

    def read(self, size=-1):
        if self.max_bytes_per_sec <= 0:
            return self.file_obj.read(size)
        
        current_time = time.time()
        elapsed = current_time - self.last_read_time

        # 计算当前周期内允许读取的字节数
        allowed_bytes = self.max_bytes_per_sec * elapsed
        if self.bytes_read_since_last >= allowed_bytes:
            # 超出速率，等待到下一个周期
            wait_time = (self.bytes_read_since_last - allowed_bytes) / self.max_bytes_per_sec
            time.sleep(wait_time)
            self.last_read_time = time.time()
            self.bytes_read_since_last = 0

        data = self.file_obj.read(size)
        self.bytes_read_since_last += len(data)
        return data

整合到断点续传代码中 #

把你的自定义分片上传代码里的文件对象用BandwidthLimiter包装，就能自动实现带宽限制。以下是整合后的示例（假设你已有断点续传的基础逻辑）：

import boto3

s3 = boto3.client('s3')
bucket_name = 'your-bucket-name'
object_key = 'large-file.tar.gz'
local_file_path = '/path/to/your/large/file'
part_size = 10 * 1024 * 1024  # 10MB 分片大小
max_bandwidth = 5 * 1024 * 1024  # 限制为5MB/s（字节数：5*1024*1024）

# 1. 获取已有的上传会话和已上传分片（断点续传核心逻辑）
upload_id = 'your-existing-upload-id'  # 从之前的上传会话获取
existing_parts = {}
# 调用list_parts获取已上传的分片编号
parts_response = s3.list_parts(
    Bucket=bucket_name,
    Key=object_key,
    UploadId=upload_id
)
for part in parts_response.get('Parts', []):
    existing_parts[part['PartNumber']] = part['ETag']

# 2. 用带宽限制包装文件对象，开始上传未完成的分片
with open(local_file_path, 'rb') as raw_file:
    # 包装文件对象，启用带宽限制
    limited_file = BandwidthLimiter(raw_file, max_bandwidth)
    part_number = 1
    uploaded_parts = []

    while True:
        # 跳过已上传的分片
        if part_number in existing_parts:
            uploaded_parts.append({
                'PartNumber': part_number,
                'ETag': existing_parts[part_number]
            })
            # 移动文件指针到下一个分片的起始位置
            limited_file.file_obj.seek(part_size * part_number)
            part_number += 1
            continue

        # 读取分片数据（自动受带宽限制）
        part_data = limited_file.read(part_size)
        if not part_data:
            break

        # 上传当前分片
        upload_response = s3.upload_part(
            Bucket=bucket_name,
            Key=object_key,
            PartNumber=part_number,
            UploadId=upload_id,
            Body=part_data
        )
        uploaded_parts.append({
            'PartNumber': part_number,
            'ETag': upload_response['ETag']
        })
        part_number += 1

    # 3. 完成分片上传
    s3.complete_multipart_upload(
        Bucket=bucket_name,
        Key=object_key,
        UploadId=upload_id,
        MultipartUpload={'Parts': uploaded_parts}
    )

注意事项 #

• max_bandwidth的单位是字节/秒，如果要限制为10MB/s，需要传入10 * 1024 * 1024。
• 这个限制是全局的，会覆盖所有分片的读取速率，确保整体上传速率不超过设定值。
• 如果不想自己实现速率控制，也可以用第三方库（如pylimiter），但上面的自定义类足够满足需求且无额外依赖。

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