Python多线程任务队列的优化实践：避免死锁与高效任务分发

class UrlConverter:
    def load(self, filename: str):
        # ...
        queue = Queue(maxsize=10) # 队列最大容量为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

# 获取urls.txt文件的路径
def get_urls_file_path(filename: str):
    return str(Path(__file__).parent / Path(filename))

# 定义每个线程要执行的任务
def process_url(url: str):
    try:
        # 实际的网络请求操作
        resp = requests.get(url, timeout=5) # 增加超时，避免长时间等待
        return url, resp.status_code
    except requests.exceptions.RequestException as e:
        return url, f"Error: {e}"

# 定义一个生成器，惰性地从文件中读取URL
def get_urls_lazy(file_name: str):
    urls_file_path = get_urls_file_path(file_name)
    with open(urls_file_path, "r", encoding="utf-8") as f_in:
        for line in f_in:
            url = line.strip()
            if url:  # 忽略空行
                yield url

if __name__ == "__main__":
    # 使用ThreadPool，指定并发线程数为10
    # with语句确保Pool资源在任务完成后被正确关闭
    with ThreadPool(processes=10) as pool:
        # imap_unordered 接受一个函数和一个可迭代对象
        # 它会惰性地从 get_urls_lazy 获取URL，并提交给线程池处理
        # 结果是无序的，一旦任务完成就立即返回
        print("开始处理URL...")
        for url, status_code in pool.imap_unordered(process_url, get_urls_lazy("urls.txt")):
            print(f"{url}: {status_code}")
    print("所有URL处理完毕。")

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)

from multiprocessing import Pool
import requests # 假设requests库在多进程环境中也能正常工作，通常可以
from pathlib import Path

# ... (process_url 和 get_urls_lazy 函数与 ThreadPool 示例相同) ...

if __name__ == "__main__":
    # 使用multiprocessing.Pool，指定并发进程数为10
    with Pool(processes=10) as pool:
        print("开始处理URL (使用进程池)...")
        for url, status_code in pool.imap_unordered(process_url, get_urls_lazy("urls.txt")):
            print(f"{url}: {status_code}")
    print("所有URL处理完毕 (使用进程池)。")