Python中multiprocessing.ThreadPool.imap触发空deque弹出IndexError导致冻结如何解决

• 补全全链路异常捕获
  multiprocessing.ThreadPool的imap、imap_unordered方法会将子线程抛出的异常延迟到迭代对应结果项时才触发，如果迭代过程中未走到异常项，或异常触发了线程池内部静默错误，就会出现无报错终止的情况。需要在download函数内部用try-except包裹所有逻辑，配合traceback模块打印完整异常栈，不要遗漏任何异常日志。
• 检查requests超时配置
  所有requests.get调用必须加超时参数，如果未配置timeout，碰到网络波动时请求会无限等待，直接导致线程卡死，最终引发程序冻结。建议配置timeout=(连接超时, 读取超时)，例如timeout=(10, 30)，分别限制连接建立时间和数据读取时间。
• 排查资源泄露问题
  如果requests返回的响应对象未正常关闭，频繁请求会快速占满端口、文件句柄等系统资源，触发系统主动杀进程。推荐用with requests.get(...) as resp:的上下文管理器写法自动释放连接，同时给requests配置连接池，限制最大连接数。
• 加日志定位卡住节点
  分别在任务提交前、download函数入口、结果返回前、迭代取结果的位置打印日志，确认是卡在任务执行阶段、迭代结果阶段还是pool.join阶段：如果卡在join阶段，基本可以判定是有线程出现无限等待或死锁。
• 验证共享资源线程安全
  如果download函数内存在操作全局变量、写入同一文件等跨线程共享资源的逻辑，未加锁会触发竞态条件，引发偶发崩溃或死锁，需要排查所有共享资源操作是否加了线程锁。

基础改造方案（保留原有ThreadPool写法） #

1. 补全download函数的异常捕获和超时配置

import traceback
import requests

def download(url):
    try:
        # 加超时，用raise_for_status捕获4xx、5xx响应码
        resp = requests.get(url, timeout=(10, 30))
        resp.raise_for_status()
        # 此处替换为你的图片保存逻辑
        return f"success: {url}"
    except Exception as e:
        traceback.print_exc()
        return f"failed: {url}, error: {str(e)}"

2. 迭代结果时也加异常捕获，避免子线程异常直接终止主进程

pool = ThreadPool(args.threads)
res = pool.imap_unordered(download, urls)
while True:
    try:
        item = next(res)
        print(item)
    except StopIteration:
        break
    except Exception as e:
        traceback.print_exc()
pool.close()
pool.join()

更稳定的替换方案 #

multiprocessing.ThreadPool是较早期的实现，异常处理和资源管理的稳定性较差，推荐替换为concurrent.futures.ThreadPoolExecutor，无序返回可以配合as_completed实现：

import traceback
from concurrent.futures import ThreadPoolExecutor, as_completed

with ThreadPoolExecutor(max_workers=args.threads) as executor:
    futures = [executor.submit(download, url) for url in urls]
    for future in as_completed(futures):
        try:
            item = future.result()
            print(item)
        except Exception as e:
            traceback.print_exc()

附加优化 #

配置requests自动重试，减少网络波动带来的偶发失败：

from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# 全局初始化带重试的session，download函数内复用该session
session = requests.Session()
retry_strategy = Retry(
    total=3, # 最多重试3次
    backoff_factor=1, # 重试间隔按1/2/4秒递增
    status_forcelist=[429, 500, 502, 503, 504] # 碰到这些状态码自动重试
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
session.mount("http://", adapter)

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