Python多线程任务队列优化：避免阻塞与高效处理大数据

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

from multiprocessing.pool import ThreadPool # 或者 multiprocessing.Pool
import requests
from pathlib import Path

# 辅助函数：从文件中获取URL的生成器
def get_urls(filename: str):
    """
    从指定文件中逐行读取URL，并作为生成器返回。
    避免一次性加载所有URL到内存。
    """
    urls_file_path = str(Path(__file__).parent / Path(filename))
    with open(urls_file_path, "r", encoding="utf-8") as f_in:
        for url in map(str.strip, f_in):
            if url: # 忽略空行
                yield url

# 工作函数：处理单个URL的任务
def process_url(url: str):
    """
    模拟对单个URL进行处理，例如发起HTTP请求并返回状态码。
    """
    try:
        response = requests.get(url, timeout=5) # 增加超时，防止长时间阻塞
        return url, response.status_code
    except requests.exceptions.RequestException as e:
        return url, f"Error: {e}"
    except Exception as e:
        return url, f"Unexpected Error: {e}"

# 主执行逻辑
def main():
    num_threads = 10  # 工作者线程数量
    urls_file = 'urls.txt'

    print(f"开始使用 {num_threads} 个线程处理URL...")

    # 使用ThreadPool上下文管理器，确保池资源被正确释放
    with ThreadPool(processes=num_threads) as pool:
        # imap_unordered 会从 get_urls 生成器中按需获取任务
        # 并将它们分发给池中的线程进行 process_url 处理
        for url, result in pool.imap_unordered(process_url, get_urls(urls_file)):
            print(f"{url}: {result}")

    print("所有URL处理完毕。")

if __name__ == "__main__":
    # 创建一个示例 urls.txt 文件，如果它不存在
    if not Path('urls.txt').exists():
        sample_urls = [
            "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)", # 可能404
            "https://en.wikipedia.org/wiki/Sonoran_Desert",
            "https://en.wikipedia.org/wiki/Steppe",
            "https://en.wikipedia.org/wiki/Stream_(freshwater)", # 可能404
            "https://en.wikipedia.org/wiki/Subarctic_tundra", # 可能404
            "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"
        ]
        with open('urls.txt', 'w', encoding='utf-8') as f:
            for url in sample_urls:
                f.write(url + '\n')
        print("已创建示例 urls.txt 文件。")

    main()