多进程下载并解压.tar.bz2文件异常问题排查

核心疑问解答 #

并发进程数不受可用CPU数量的强制限制——操作系统允许创建超过CPU核心数的进程，但过多CPU密集型进程会因上下文切换导致整体效率下降。你的情况并非CPU数量直接限制了进程启动，更可能是代码逻辑存在问题。

常见代码问题排查 #

结合你描述的现象（URL数量等于vCPU数时解压启动，数量多时无反应），以下是最可能的原因：

1. 多进程代码未受主模块保护（Windows系统特有） #

在Windows中，Python多进程会通过导入主模块创建子进程，如果多进程相关代码未放在if __name__ == '__main__':块内，子进程会重复执行整个脚本，导致进程无法正常启动。当URL数量较少时，可能因执行顺序侥幸绕过问题，但数量多时会彻底卡住。

2. 任务提交逻辑阻塞 #

如果你的代码是先等待所有下载线程完成，再提交解压任务，那么当URL数量大时，下载耗时久，解压自然不会启动。但你提到“下载过程正常且较快，但解压从未启动”，这种可能性较低，但仍需确认是否存在等待全部下载完成的逻辑。

3. 线程内创建进程的错误操作 #

若你在下载线程中直接提交解压任务到进程池，可能导致进程创建失败——虽然Python允许线程内创建进程，但部分场景下会因资源竞争或初始化问题导致子进程无法启动。

4. 文件写入未完成就提交解压 #

下载线程可能还未将文件完全写入磁盘，就提交了解压任务，导致解压进程因文件损坏或占用而卡住，但这种情况通常会抛出异常，而非完全不启动。

修正后的示例代码 #

采用生产者-消费者模式，用队列传递下载完成的文件路径，实现边下载边解压，同时规避多进程的常见坑：

import threading
from multiprocessing import Pool, Queue
import requests
import tarfile
import os

def download_worker(url_queue, file_queue, save_dir):
    """下载线程工作函数：从URL队列取任务，下载后将文件路径放入文件队列"""
    os.makedirs(save_dir, exist_ok=True)
    while not url_queue.empty():
        url = url_queue.get()
        try:
            fname = url.split('/')[-1]
            save_path = os.path.join(save_dir, fname)
            # 流式下载大文件
            with requests.get(url, stream=True) as resp:
                resp.raise_for_status()
                with open(save_path, 'wb') as f:
                    for chunk in resp.iter_content(chunk_size=1024*1024):
                        f.write(chunk)
            # 下载完成后放入解压队列
            file_queue.put(save_path)
        except Exception as e:
            print(f"下载失败 {url}: {str(e)}")
        finally:
            url_queue.task_done()

def extract_worker(file_path, dest_dir):
    """解压进程工作函数"""
    os.makedirs(dest_dir, exist_ok=True)
    try:
        with tarfile.open(file_path, 'r:bz2') as tar:
            tar.extractall(dest_dir)
        print(f"解压完成 {file_path}")
    except Exception as e:
        print(f"解压失败 {file_path}: {str(e)}")

def main():
    # 配置参数
    urls = ["URL1", "URL2", "..."]  # 你的URL列表
    save_dir = "./downloads"
    dest_dir = "./extracted"
    thread_num = 5  # IO密集型任务，线程数可设为CPU数的5-10倍
    process_num = 2  # CPU密集型任务，进程数设为vCPU数即可

    # 初始化队列
    url_queue = Queue()
    for url in urls:
        url_queue.put(url)
    file_queue = Queue()

    # 启动下载线程
    download_threads = []
    for _ in range(thread_num):
        t = threading.Thread(target=download_worker, args=(url_queue, file_queue, save_dir))
        t.daemon = True
        t.start()
        download_threads.append(t)

    # 启动解压进程池，循环从文件队列取任务
    with Pool(processes=process_num) as pool:
        while True:
            # 等待下载线程全部完成且文件队列为空时退出
            if url_queue.empty() and file_queue.empty() and all(not t.is_alive() for t in download_threads):
                break
            # 非阻塞获取文件路径，避免卡住
            try:
                file_path = file_queue.get(block=False)
                pool.apply_async(extract_worker, args=(file_path, dest_dir))
            except:
                pass

if __name__ == '__main__':
    main()

关键优化点 #

• 用队列实现下载与解压的解耦，下载完成一个就解压一个，无需等待全部下载完成。
• 多进程代码完全放在if __name__ == '__main__':块内，规避Windows系统的进程创建问题。
• 下载线程设为守护线程，确保主进程退出时线程能自动结束。
• 非阻塞读取文件队列，避免主进程因等待队列而卡住。

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