Python多进程池下载图片PIL处理异常的解决咨询

问题背景 #

使用multiprocessing.Pool批量下载数千张图片，结合PIL进行拼接或处理。核心问题是部分下载的图片会损坏，导致PIL抛出异常；即便检查了返回数据数组的长度，后续仍会因损坏图片报错。需要实现两种重跑机制：仅重下损坏图片，或是重启整个进程池（单池总数据约15MB）。

重试方案实现 #

方案1：单任务精准重试（仅重下损坏图片） #

在下载+处理的任务函数中加入重试逻辑，针对单张图片的下载/验证失败做重试，避免因个别任务失败中断整个流程：

from multiprocessing import Pool
from PIL import Image
import requests
from io import BytesIO

def download_and_process(img_url, max_retries=3):
    for attempt in range(max_retries):
        try:
            # 下载图片
            resp = requests.get(img_url, timeout=10)
            resp.raise_for_status()  # 捕获HTTP请求错误
            # 提前验证图片完整性，避免后续处理报错
            img_buffer = BytesIO(resp.content)
            with Image.open(img_buffer) as img:
                img.verify()  # 验证图片格式与完整性
            # 验证通过后重新加载图片（verify后文件指针会移到末尾）
            img_buffer.seek(0)
            img = Image.open(img_buffer)
            # 这里添加你的图片处理逻辑（裁剪、尺寸调整等）
            return (img_url, img, True)
        except (requests.exceptions.RequestException, 
                Image.UnidentifiedImageError, IOError) as e:
            if attempt == max_retries - 1:
                # 最后一次重试失败，返回失败标识
                return (img_url, None, False)
            # 非最后一次失败，继续重试
            continue

def main():
    img_urls = ["https://example.com/img1.jpg", ...]  # 你的URL列表
    # 首次执行任务
    with Pool(processes=4) as pool:
        results = pool.map(download_and_process, img_urls)
    
    # 筛选失败任务，发起重试
    failed_urls = [url for url, _, success in results if not success]
    if failed_urls:
        print(f"检测到{len(failed_urls)}张图片下载/处理失败，开始重试")
        with Pool(processes=2) as retry_pool:
            retry_results = retry_pool.map(download_and_process, failed_urls)
            # 合并结果（根据业务需求调整，比如替换原失败项或追加）
            for url, img, success in retry_results:
                for idx, (res_url, _, _) in enumerate(results):
                    if res_url == url:
                        results[idx] = (url, img, success)
                        break

if __name__ == "__main__":
    main()

方案2：重启整个进程池 #

由于单池数据量仅15MB，资源消耗极低，可在检测到任务失败时直接重启整个进程池：

from multiprocessing import Pool

def run_pool_task(url_list):
    with Pool(processes=4) as pool:
        task_results = pool.map(download_and_process, url_list)
    # 检查是否存在失败任务
    has_failures = any(not success for _, _, success in task_results)
    return task_results, has_failures

def main():
    img_urls = ["https://example.com/img1.jpg", ...]
    max_restart_attempts = 2
    current_attempt = 0
    final_results = None

    while current_attempt < max_restart_attempts:
        final_results, has_failures = run_pool_task(img_urls)
        if not has_failures:
            break
        current_attempt += 1
        print(f"任务执行失败，第{current_attempt}次重启进程池")
    
    if current_attempt >= max_restart_attempts:
        print("已达最大重启次数，仍存在失败任务")
    # 后续处理final_results

if __name__ == "__main__":
    main()

实际排查过程与根因解决 #

排查时逐步排除常见问题：

• 最初在任务函数中添加try-except并限制重试次数，问题未解决
• 将下载的分片图片保存到本地目录，发现本地打开全部正常，但约1/20的图片在粘贴到大图时失败，一度怀疑是内存不足
• 修改下载函数返回失败标识，尝试重新触发任务逻辑，仍未解决
• 最终定位根因：存储图片的VM中Samba服务器崩溃，导致图片读写过程中出现隐性损坏；给VM增加资源后，问题完全消失

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