Python多线程并发：利用ThreadPool高效处理大规模任务队列

class UrlConverter:
    def load(self, filename: str):
        # ...
        queue = Queue(maxsize=10) # 设定了最大容量
        with open(urls_file_path, 'r', encoding="utf-8") as txt_file:
            for line in txt_file:
                line = line.strip()
                queue.put(line) # 在这里尝试填充队列
        return queue

# ...
def main():
    url_converter = UrlConverter()
    urls_queue = url_converter.load('urls.txt') # 生产者在这里一次性填充队列
    fetcher_threads.execute(urls_queue) # 消费者（线程）在这里才开始从队列取数据

from multiprocessing.pool import ThreadPool
import requests
from pathlib import Path
import time

# 辅助函数：生成示例urls.txt文件
def create_sample_urls_file(filename="urls.txt"):
    urls_content = """
https://en.wikipedia.org/wiki/Sea-level_rise
https://en.wikipedia.org/wiki/Sequoia_National_Park
https://en.wikipedia.org/wiki/Serengeti
https://en.wikipedia.org/wiki/Sierra_Nevada_(Utah)
https://en.wikipedia.org/wiki/Sonoran_Desert
https://en.wikipedia.org/wiki/Steppe
https://en.wikipedia.org/wiki/Swiss_Alps
https://en.wikipedia.org/wiki/Taiga
https://en.wikipedia.org/wiki/Tatra_Mountains
https://en.wikipedia.org/wiki/Temperate_rainforest
https://en.wikipedia.org/wiki/Tropical_rainforest
https://en.wikipedia.org/wiki/Tundra
https://en.wikipedia.org/wiki/Ural_Mountains
https://en.wikipedia.org/wiki/Wetland
https://en.wikipedia.org/wiki/Wildlife_conservation
https://en.wikipedia.org/wiki/Salt_marsh
https://en.wikipedia.org/wiki/Savanna
https://en.wikipedia.org/wiki/Scandinavian_Mountains
https://en.wikipedia.org/wiki/Subarctic_tundra
https://en.wikipedia.org/wiki/Stream_(freshwater)
    """
    file_path = Path(__file__).parent / Path(filename)
    if not file_path.exists():
        file_path.write_text(urls_content.strip(), encoding="utf-8")
        print(f"创建了示例文件: {filename}")
    else:
        print(f"文件 {filename} 已存在，跳过创建。")


# 生成器函数：惰性地从文件中读取URL
def get_urls(file_name):
    urls_file_path = str(Path(__file__).parent / Path(file_name))
    try:
        with open(urls_file_path, 'r', encoding="utf-8") as f_in:
            for url in map(str.strip, f_in):
                if url: # 过滤掉空行
                    yield url
    except FileNotFoundError:
        print(f"错误: 文件 '{file_name}' 未找到。请确保文件存在。")
        return # 返回空生成器

# 工作函数：处理单个URL任务
def process_url(url):
    try:
        # 模拟网络请求，并设置超时以防止长时间阻塞
        response = requests.get(url, timeout=10)
        return url, response.status_code
    except requests.exceptions.Timeout:
        return url, "Error: Request timed out"
    except requests.exceptions.RequestException as e:
        return url, f"Error: {e}"
    except Exception as e:
        return url, f"Unexpected Error: {e}"

if __name__ == "__main__":
    # 确保urls.txt文件存在
    create_sample_urls_file("urls.txt")

    num_workers = 5 # 设定线程池的大小，例如5个工作线程

    print(f"开始使用 {num_workers} 个线程处理URL任务...")
    start_time = time.time()

    # 使用ThreadPool上下文管理器，确保线程池正确关闭
    with ThreadPool(processes=num_workers) as pool:
        # imap_unordered 惰性地从 get_urls 获取任务，并将它们分发给线程池中的工作线程。
        # 结果会以任务完成的顺序返回，而不是输入的顺序。
        for url, result in pool.imap_unordered(process_url, get_urls("urls.txt")):
            print(f"处理完成: {url} -> {result}")

    end_time = time.time()
    print(f"\n所有URL任务处理完毕。总耗时: {end_time - start_time:.2f} 秒。")