如何向ThreadPoolExecutor传递列表参数并获取索引与值实现多线程视频转换

原代码核心问题 #

• 线程池初始化放在了for循环内部，每次循环都会新建一个线程池，无法复用4个工作线程，等于串行执行任务，多线程完全失效
• 没有做任务异常捕获，转码过程中出现报错时无法定位到对应索引的视频

正确实现代码 #

from concurrent.futures import ThreadPoolExecutor, as_completed

class Transcode:
    # 这里保留你原有的Transcode类逻辑，以下是hls方法的参考实现
    def hls(self, video_path: str, index: int):
        # 用索引和视频名创建独立文件夹的参考逻辑
        # 你可以根据自己的需求调整目录命名规则
        import os
        video_name = os.path.basename(video_path).rsplit('.', 1)[0]
        save_dir = f"./transcode_output/video_{index}_{video_name}"
        os.makedirs(save_dir, exist_ok=True)
        
        # 此处写你原本的视频转码逻辑，save_dir可直接作为输出目录参数使用
        # 以下是模拟转码耗时的示例代码，可删除
        import time
        time.sleep(3)
        return f"索引{index}的视频{video_path}转码完成，输出目录：{save_dir}"

def main(videos):
    transcode = Transcode()
    # 线程池初始化移到循环外部，复用4个工作线程处理所有转码任务
    with ThreadPoolExecutor(max_workers=4) as executor:
        future_map = {}
        for index, video in enumerate(videos):
            future = executor.submit(transcode.hls, video, index)
            # 存储任务和对应索引、视频的映射，方便后续定位异常
            future_map[future] = (index, video)
        
        # 可选：获取转码结果、捕获执行异常
        for future in as_completed(future_map):
            index, video = future_map[future]
            try:
                result = future.result()
                print(result)
            except Exception as e:
                print(f"索引{index}的视频{video}转码失败，错误信息：{str(e)}")

# 测试调用示例
if __name__ == "__main__":
    # 替换成你实际的视频路径列表即可
    video_list = ["./test1.mp4", "./test2.mp4", "./test3.mp4", "./test4.mp4", "./test5.mp4"]
    main(video_list)

关键逻辑说明 #

• 线程池ThreadPoolExecutor的上下文管理器移动到循环外，4个工作线程会持续复用处理所有转码任务，相比原代码执行效率提升数倍
• 直接通过enumerate遍历拿到的索引和视频路径会准确传递给每个转码任务，可直接用于生成独立的输出目录
• 增加了异常捕获逻辑，转码过程中出现的任何错误都可以直接定位到对应索引的视频，方便调试排查问题

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