如何限制Python Boto3 S3大文件的上传带宽？

Boto3本身没有内置的带宽限速功能，但可以通过自定义文件流包装或利用botocore工具类实现绝对限速或相对限速，以下是具体实现方案：

一、绝对限速（固定10 Mbps） #

首先转换单位：10 Mbps = 1.25 MB/s（1 Mbps = 128 KB/s，10×128 = 1280 KB/s = 1.25 MB/s），通过控制文件读取速度间接限制上传带宽。

方法1：使用botocore内置RateLimiter #

利用botocore.utils.RateLimiter实现精准限速：

import boto3
from botocore.utils import RateLimiter

# 配置10 Mbps对应的字节/秒
max_bytes_per_sec = 10 * 1024 * 1024 // 8

s3 = boto3.client('s3')
bucket_name = 'your-bucket-name'
local_file_path = '5gb_large_file.bin'
s3_key = 'uploads/large_file.bin'

# 创建限流器
rate_limiter = RateLimiter(max_bytes_per_sec)

# 自定义带限速的读取函数
def limited_read(file_obj, chunk_size):
    rate_limiter.acquire(chunk_size)
    return file_obj.read(chunk_size)

# 包装文件对象的read方法
with open(local_file_path, 'rb') as f:
    original_read = f.read
    f.read = lambda size: limited_read(f, size)
    
    s3.upload_fileobj(f, bucket_name, s3_key)

方法2：自定义限速文件类 #

更灵活的手动限速实现，适合自定义逻辑：

import boto3
import time

class RateLimitedFile:
    def __init__(self, file_obj, max_bytes_per_sec):
        self.file_obj = file_obj
        self.max_bytes_per_sec = max_bytes_per_sec
        self.last_read_ts = time.time()
        self.bytes_read_since_last = 0

    def read(self, size=-1):
        if size == -1:
            # 读取全部分块处理
            data = b''
            while True:
                chunk = self.read(4096)
                if not chunk:
                    break
                data += chunk
            return data
        
        # 计算需要等待的时间以满足限速
        current_ts = time.time()
        elapsed = current_ts - self.last_read_ts
        if elapsed > 0:
            expected_time = self.bytes_read_since_last / self.max_bytes_per_sec
            if elapsed < expected_time:
                time.sleep(expected_time - elapsed)
        
        # 读取数据并更新统计
        data = self.file_obj.read(size)
        self.bytes_read_since_last += len(data)
        self.last_read_ts = current_ts
        return data

# 配置10 Mbps限速
max_bytes_per_sec = 10 * 1024 * 1024 // 8

s3 = boto3.client('s3')
bucket_name = 'your-bucket-name'
local_file_path = '5gb_large_file.bin'
s3_key = 'uploads/large_file.bin'

with open(local_file_path, 'rb') as f:
    limited_file = RateLimitedFile(f, max_bytes_per_sec)
    s3.upload_fileobj(limited_file, bucket_name, s3_key)

二、相对限速（可用带宽的10%） #

需要先检测当前系统的可用带宽，再按比例分配上传速度。这里用psutil监控系统级网络流量：

import boto3
import psutil
import time

class DynamicRateLimitedFile:
    def __init__(self, file_obj, bandwidth_percent=10):
        self.file_obj = file_obj
        self.bandwidth_percent = bandwidth_percent
        self.last_read_ts = time.time()
        self.bytes_read_since_last = 0
        self.total_bandwidth = self._get_total_interface_bandwidth()

    def _get_total_interface_bandwidth(self):
        # 获取第一个活跃非回环接口的总带宽（转换为字节/秒）
        for iface, stats in psutil.net_if_stats().items():
            if not stats.isup or iface == 'lo':
                continue
            # stats.speed单位是Mbps，转换为字节/秒：Mbps × 1024×1024 ÷8
            return stats.speed * 1024 * 1024 // 8
        # 默认 fallback 100 Mbps
        return 100 * 1024 * 1024 // 8

    def _get_available_bandwidth(self):
        # 计算当前可用上传带宽（总带宽 - 其他进程上传速度）
        net_io_old = psutil.net_io_counters()
        time.sleep(0.1)
        net_io_new = psutil.net_io_counters()
        current_upload_speed = (net_io_new.bytes_sent - net_io_old.bytes_sent) / 0.1
        return max(self.total_bandwidth - current_upload_speed, 0)

    def read(self, size=-1):
        if size == -1:
            data = b''
            while True:
                chunk = self.read(4096)
                if not chunk:
                    break
                data += chunk
            return data
        
        # 动态计算允许的最大上传速度
        available_bw = self._get_available_bandwidth()
        max_bytes_per_sec = available_bw * (self.bandwidth_percent / 100)
        
        if max_bytes_per_sec <= 0:
            time.sleep(0.1)
            return self.file_obj.read(size)
        
        # 应用限速逻辑
        current_ts = time.time()
        elapsed = current_ts - self.last_read_ts
        if elapsed > 0:
            expected_time = self.bytes_read_since_last / max_bytes_per_sec
            if elapsed < expected_time:
                time.sleep(expected_time - elapsed)
        
        data = self.file_obj.read(size)
        self.bytes_read_since_last += len(data)
        self.last_read_ts = current_ts
        return data

s3 = boto3.client('s3')
bucket_name = 'your-bucket-name'
local_file_path = '5gb_large_file.bin'
s3_key = 'uploads/large_file.bin'

with open(local_file_path, 'rb') as f:
    limited_file = DynamicRateLimitedFile(f, bandwidth_percent=10)
    s3.upload_fileobj(limited_file, bucket_name, s3_key)

注意事项 #

• 大文件上传建议优先使用upload_fileobj或upload_file，它们会自动分块处理，配合限速更稳定。
• 相对限速的精度依赖于系统网络统计的准确性，psutil会统计所有进程的流量，若需更精准可结合专业带宽测试工具（如speedtest-cli）预检测带宽。

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