带索引图片URL批量多文件下载：requests多线程实现问询

嘿，你的思路方向是对的——用字典映射文件名和对应URL，结合多线程来提速批量图片下载。不过你贴的代码还缺了几个关键部分：没有实际创建/启动线程、没做异常处理（网络出错或者文件写入失败会直接崩）、requests.get没设置超时（遇到卡壳的请求会让线程无限挂着），还有分片的逻辑可以更清晰。

下面是完善后的完整代码，附带详细说明：

完整可运行代码 #

import requests
import threading
import os
from typing import Dict

# 先模拟你的文件名和URL字典（替换成你实际的数据）
name = {0: "cat_0.jpg", 1: "dog_1.jpg", 2: "bird_2.jpg"}  # 索引对应文件名
d2 = {0: "https://example.com/cat0.jpg", 1: "https://example.com/dog1.jpg", 2: "https://example.com/bird2.jpg"}  # 索引对应图片URL

def download_single_image(index: int):
    """负责单张图片的下载，包含完整异常处理"""
    try:
        url = d2[index]
        filename = name[index]
        
        # 设置10秒超时，避免请求卡壳导致线程挂起
        response = requests.get(url, timeout=10)
        # 检查HTTP状态码，非200直接抛出异常
        response.raise_for_status()
        
        # 确保下载目录存在，不存在就创建（不会报错）
        os.makedirs("assayImage", exist_ok=True)
        
        # 写入文件
        with open(f'assayImage/{filename}', 'wb') as f:
            f.write(response.content)
        print(f"✅ 成功下载: {filename}")
    except requests.exceptions.RequestException as e:
        print(f"❌ 下载{name.get(index, '未知文件')}失败: {str(e)}")
    except Exception as e:
        print(f"⚠️ 处理第{index}号文件时出错: {str(e)}")

def main():
    batch_size = 1500  # 每批次启动的线程数，避免一次性开太多线程占资源
    total_tasks = len(d2)
    
    # 按批次拆分任务，启动线程
    for start in range(0, total_tasks, batch_size):
        stop = min(start + batch_size, total_tasks)
        current_threads = []
        
        # 创建并启动当前批次的所有线程
        for i in range(start, stop):
            thread = threading.Thread(target=download_single_image, args=(i,))
            current_threads.append(thread)
            thread.start()
        
        # 等待当前批次所有线程完成，再处理下一批
        for thread in current_threads:
            thread.join()

if __name__ == "__main__":
    main()

关键改进点说明 #

• 异常兜底：不管是网络请求失败（比如超时、404）还是文件写入出错，都会捕获并打印错误信息，不会让单个任务搞崩整个程序。
• 超时保护：给requests.get加了timeout=10，遇到慢请求或者死链时，线程会及时退出，不会一直挂着浪费资源。
• 目录自动创建：用os.makedirs确保assayImage目录存在，不用手动提前创建，避免文件写入时报错。
• 线程分批管理：每批次只启动1500个线程，等这批都跑完再开下一批，防止一次性创建几千个线程导致系统CPU/内存过载。

更简洁的进阶方案：用线程池 #

如果觉得手动管理线程批次太麻烦，推荐用concurrent.futures.ThreadPoolExecutor，它会自动帮你管理线程池大小，代码更清爽：

import requests
import os
from concurrent.futures import ThreadPoolExecutor

# 替换成你的实际数据
name = {0: "cat_0.jpg", 1: "dog_1.jpg"}
d2 = {0: "https://example.com/cat0.jpg", 1: "https://example.com/dog1.jpg"}

def download_single_image(index: int):
    try:
        url = d2[index]
        filename = name[index]
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        
        os.makedirs("assayImage", exist_ok=True)
        with open(f'assayImage/{filename}', 'wb') as f:
            f.write(response.content)
        print(f"✅ 成功下载: {filename}")
    except Exception as e:
        print(f"❌ 下载失败({index}): {str(e)}")

def main():
    # 线程池大小建议设为32-64（根据你的网络带宽调整，别太大）
    with ThreadPoolExecutor(max_workers=32) as executor:
        # 把所有索引丢给线程池，自动分配任务
        executor.map(download_single_image, d2.keys())

if __name__ == "__main__":
    main()

线程池的好处是不用自己拆分批次，它会自动控制并发数，避免资源浪费，代码也更简洁易维护，适合大多数批量下载场景。

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