Python多线程ThreadPool执行任务耗时超单线程的性能问题

你的ThreadPool实现未达到预期提速效果，核心原因如下：

1. GIL（全局解释器锁）的限制 #

Python的线程在执行CPU密集型任务时，同一时间只有一个线程能持有GIL并执行代码。多线程本质是并发切换而非并行执行，反而会增加线程上下文切换的额外开销，导致总耗时上升。你的getPairs函数属于CPU密集型任务，ThreadPool完全无法发挥多核优势。

2. 任务类型与线程池不匹配 #

ThreadPool仅适合IO密集型任务（如文件读写、网络请求）——这类任务执行时线程会主动释放GIL，让其他线程有机会运行。但你的基准测试核心是CPU计算，线程池完全不适用于此场景。

3. 线程池的额外开销 #

线程池的创建、任务分发、结果收集过程本身存在开销，当单个任务执行时间较短时，该开销会被进一步放大，拉低整体效率。

解决方案：改用多进程池（multiprocessing.Pool） #

多进程的每个子进程拥有独立的Python解释器和GIL，能真正利用多核CPU并行执行CPU密集型任务。

修改后的基准测试代码 #

from multiprocessing import Pool
import os
from math import floor
import time
from queue import Queue

# 全局变量，用于多进程共享字典数据
global_words = None

def init_worker(words):
    """多进程初始化函数，传递共享的字典数据"""
    global global_words
    global_words = words

def get_set_from_dict_file(filename):
    # 保持原实现不变
    pass

def getPairs(words):
    # 保持原实现不变
    pass

def main_with_shared_data(print_results=True):
    """修改后的main函数，使用预先加载的共享数据"""
    results = Queue()
    start_time = time.time()

    words = global_words
    results.put(f"Total words read: {len(words)}")
    results.put(f"Total time taken to read the file: 0 ms")  # 已提前读取
    start_time_2 = time.time()

    pairs = getPairs(words)
    results.put(f"Number of words that can be built with 3 letter word + letter + 3 letter word: {len(pairs)}")

    results.put(f"Total time taken to find the pairs: {round((time.time() - start_time_2) * 1000)} ms")
    results.put(f"Time taken: {round((time.time() - start_time) * 1000)}ms")

    if print_results:
        [print(x) for x in results.queue]
    return (time.time() - start_time) * 1000

def benchmark(n=1000):
    core_count = os.cpu_count()
    process_num = floor(core_count * 0.9)
    # 提前读取字典文件，避免每个进程重复IO
    words = get_set_from_dict_file("usa.txt")
    
    with Pool(process_num, initializer=init_worker, initargs=(words,)) as pool:
        results = pool.map(main_with_shared_data, [False] * n)
    
    avg_time_ms = round(sum(results) / len(results))
    return avg_time_ms, -1

# 测试代码保持不变
if __name__ == "__main__":
    print("Do you want to benchmark? (y/n)")
    if input().upper() == "Y":
        print("Benchmark n times: (int)")
        n = input()
        n = int(n) if (n.isdigit() and 0 < int(n) <= 1000) else 100
        start = time.time()
        bench = benchmark(n)
        end = time.time()
        print("\n----------Multi-Process Benchmark----------")
        print(f"Average time taken: {bench[0]} ms")
        print(f"Best time taken yet: {bench[1]} ms")
        print(f"Total bench time: {end - start:0.5} s")

        start = time.time()
        non_t_results = [main_with_shared_data(False) for _ in range(n)]
        end = time.time()
        print("\n----------Single-Thread Benchmark----------")
        print(f"Average time taken: {round(sum(non_t_results) / len(non_t_results))} ms")
        print(f"Total bench time: {end - start:0.5} s")

    else:
        # 单进程运行时需要先加载数据
        global_words = get_set_from_dict_file("usa.txt")
        main_with_shared_data()

优化说明 #

1. 共享字典数据：提前读取usa.txt并传递给所有子进程，避免重复IO开销
2. 多进程并行：每个子进程在独立CPU核心上执行getPairs，真正利用多核优势
3. 避免GIL限制：多进程绕过了GIL的约束，CPU密集型任务的执行效率会随核心数增加而显著提升

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