Python使用multithreading多线程requests批量上传文件夹视频到AWS S3问询

完整改造代码 #

import requests
import base64
import hashlib
import os
import threading

# 配置最大并发上传数，可根据带宽和接口限流调整
MAX_WORKERS = 5
semaphore = threading.Semaphore(MAX_WORKERS)

tools_dir = os.path.dirname(os.path.realpath(__file__))
project_root = os.path.dirname(tools_dir)
videos_dir = os.path.join(project_root, "output", "videos")

# Basic auth settings
basic_auth_user = "abc123"
basic_auth_password = "abc456"
auth_str = f"{basic_auth_user}:{basic_auth_password}"

# Base64 编码鉴权信息
auth_encode_bytes = auth_str.encode("utf-8")
auth_encode_base64 = base64.b64encode(auth_encode_bytes)
auth_header = auth_encode_base64.decode("utf-8")

str1 = "_"

def upload_single_video(file_path, file_name, serial_number):
    # 信号量控制并发数
    with semaphore:
        try:
            # 计算文件MD5
            m = hashlib.md5()
            with open(file_path, 'rb') as f:
                for chunk in iter(lambda: f.read(4096), b""):
                    m.update(chunk)
            file_hash = m.hexdigest()

            url = "http://localhost:8000/api/awss3/upload"
            payload = {
                'fileHash': file_hash,
                'device': {'serialNumber': serial_number}
            }
            # 上传文件
            with open(file_path, 'rb') as f:
                files = [('file', (file_name, f, 'application/octet-stream'))]
                headers = {
                    'Authorization': f'Basic {auth_header}'
                }
                # 注意：如果接口要求接收JSON格式参数，将data改为json即可
                response = requests.post(url, headers=headers, data=payload, files=files)
                print(f"【{file_name}】上传结果：\n{response.text}\n")
        except Exception as e:
            print(f"【{file_name}】上传失败，错误：{str(e)}\n")

if __name__ == "__main__":
    thread_list = []
    for file in os.listdir(videos_dir):
        file_full_path = os.path.join(videos_dir, file)
        # 跳过目录，只处理文件
        if not os.path.isfile(file_full_path):
            continue
        # 提取设备序列号
        try:
            serial_number = file[:file.index(str1)]
        except ValueError:
            print(f"【{file}】文件名不符合规则，跳过\n")
            continue
        # 创建线程
        t = threading.Thread(target=upload_single_video, args=(file_full_path, file, serial_number))
        thread_list.append(t)
        t.start()
    
    # 等待所有线程执行完成
    for t in thread_list:
        t.join()
    print("所有文件上传任务执行完毕")

核心改造点 #

• 把单文件的MD5计算、上传逻辑封装为独立的线程执行函数，每个文件对应一个独立线程处理
• 引入threading.Semaphore控制最大并发数，避免并发过高触发接口限流、占满本地带宽或者导致系统资源耗尽
• 增加了异常捕获逻辑，单个文件上传失败不会影响其他文件的上传任务，同时会输出明确的错误提示
• 优化了文件名合法性校验，不符合命名规则的文件会直接跳过，避免程序运行报错
• 优化了文件句柄管理，所有文件操作都用with语法自动释放资源，避免文件句柄泄漏

调整建议 #

• 如果上传的视频文件普遍较大，建议适当调低MAX_WORKERS的值，避免单线程占用带宽过高导致其他线程上传超时
• 如果接口有明确的QPS限制，可配合time.sleep做更精细的频率控制
• 需要上传结果统计的话，可以额外加线程安全的全局计数器，记录成功、失败的文件数量

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