如何在Python类中正确调用多线程方法？解决仅执行首个任务问题

你的代码出现仅下载第一个文件就停止的核心原因是ThreadPool实例未被持有引用，导致被Python垃圾回收机制提前销毁，线程池随之关闭，后续任务无法执行。此外还有几处细节问题需要修正：

关键问题点 #

• 在_download_parallel方法中，直接创建ThreadPool实例并调用imap_unordered，但未将实例赋值给变量保存引用。Python的引用计数机制会立即回收无引用对象，ThreadPool的__del__方法会触发线程池关闭，导致仅第一个任务完成后就停止处理。
• zip对象是一次性迭代器，若后续误重复使用self.data会导致无数据可用，建议转为列表。
• _download_url的异常处理中使用pass但无返回值，会导致results中出现None，打印时会引发IndexError。

修正后的代码 #

from multiprocessing import cpu_count
from multiprocessing.pool import ThreadPool
import os
from requests import Session
import time


class OSM:  # 修复缺失的冒号

    def __init__(self):
        self.url_root = "https://my.site/index.html"
        self.s = self._mount_session()
        self.data = None  # 改为列表存储，避免zip迭代器的单次迭代问题
        self.download_path = "C:/Temp"
        
    def _download_parallel(self, args):
        # 保存ThreadPool实例引用，避免被垃圾回收
        pool = ThreadPool(cpu_count() - 1)
        results = pool.imap_unordered(self._download_url, args)
        for result in results:
            if result:  # 过滤异常情况下的空返回
                print(f"URL: {result[0]} | Time (s): {result[1]}")
        # 关闭线程池并等待所有任务完成
        pool.close()
        pool.join()

    def _download_url(self, args):
        t0 = time.time()
        url, fn = args[0], args[1]
        try:
            r = self.s.get(url)
            r.raise_for_status()  # 显式触发HTTP错误，避免静默失败
            with open(fn, 'wb') as f:
                f.write(r.content)
            return (url, round(time.time() - t0, 2))
        except Exception as e:
            print(f"Exception in _download_url() for {url}: {e}")
            return (url, None)  # 返回明确的错误标记，避免None引发后续问题

    def _mount_session(self):
        return Session()  # 这里可以添加自定义的session配置，比如headers、cookies等

    def download(self):
        if not os.path.exists(self.download_path):
            os.makedirs(self.download_path)
        return self._download_parallel(self.data)

    def do_stuff_with_files(self):
        # 处理文件的逻辑
        pass

    def get_file_list(self):
        dl_links = []
        local_files = []
        # 这里替换为实际获取下载链接和本地路径的逻辑
        # 示例：模拟多个下载任务
        dl_links = ["https://example.com/file1.txt", "https://example.com/file2.txt"]
        local_files = [os.path.join(self.download_path, "file1.txt"), os.path.join(self.download_path, "file2.txt")]
        # 将zip对象转为列表，避免单次迭代问题
        self.data = list(zip(dl_links, local_files))
        

if __name__ == "__main__":
    o = OSM()
    o.get_file_list()
    o.download()

修正说明 #

1. 持有ThreadPool引用：将ThreadPool实例赋值给pool变量，遍历完结果后调用close()和join()，确保所有线程任务完成后再销毁线程池。
2. zip转列表：将self.data从zip迭代器转为列表，避免后续重复使用时无数据的问题，同时让调试更直观。
3. 完善异常处理：添加r.raise_for_status()触发HTTP错误，异常时返回明确的标记值，避免None引发打印错误。
4. 修复语法错误：补充class OSM后的冒号。

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