如何通过AWS SDK获取S3运行中上传并控制并行上传数量？

方法1：基于AWS SDK的上传对象跟踪 #

不管你使用Java、Python还是其他语言的AWS SDK，每个S3上传请求（比如PutObject或分段上传的CreateMultipartUpload）都会返回可跟踪的任务句柄。你可以通过维护线程安全的任务集合来精准控制并行上传数量：

• 初始化一个线程安全的集合（比如Python中用threading.Lock保护的集合，Java中用ConcurrentHashMap），用于存储正在运行的上传任务标识。
• 发起上传前先检查集合内的任务数量：
  • 若数量已达maxAuthorizeUpload，直接拒绝新请求；
  • 若未达上限，将新任务标识加入集合，再执行上传操作。
• 上传完成（无论成功/失败）后，务必从集合中移除对应任务，释放名额。

以下是Python（boto3）的示例代码：

import boto3
import threading
from botocore.exceptions import ClientError

s3 = boto3.client('s3')
max_uploads = 5
running_uploads = set()
lock = threading.Lock()

def upload_to_s3(bucket, key, file_path):
    with lock:
        if len(running_uploads) >= max_uploads:
            print("并行上传数量已达上限，拒绝本次上传")
            return False
        # 生成唯一任务标识，可结合文件key和线程ID
        task_id = f"{key}_{threading.get_ident()}"
        running_uploads.add(task_id)
    
    try:
        s3.upload_file(file_path, bucket, key)
        print(f"上传成功: {key}")
        return True
    except ClientError as e:
        print(f"上传失败: {key}, 错误信息: {e}")
        return False
    finally:
        with lock:
            running_uploads.discard(task_id)

方法2：分段上传的全局状态查询 #

如果你的场景涉及大文件分段上传，可以通过list_multipart_uploads API查询指定存储桶内所有未完成的分段上传任务：

def get_running_multipart_uploads(bucket):
    response = s3.list_multipart_uploads(Bucket=bucket)
    return response.get('Uploads', [])

# 获取当前运行中的分段上传数量
running_count = len(get_running_multipart_uploads('your-target-bucket'))
if running_count >= max_uploads:
    # 阻止新的上传请求
    pass

注意：这个API会返回存储桶内所有未完成的分段上传（包括其他客户端发起的），适合跨客户端的全局数量限制；如果是单客户端内的控制，方法1的精准度更高。

关键注意事项 #

• 线程安全：必须用锁或线程安全的数据结构维护任务列表，避免并发场景下的计数错误。
• 异常清理：上传失败、超时的任务必须从集合中移除，否则会长期占用名额，导致无法发起新上传。
• 分段上传资源清理：未完成的分段上传会占用S3存储资源，建议定期用abort_multipart_upload清理超时任务。

内容的提问来源于stack exchange，提问作者damien le borgne