面试题：Python多线程在计算密集型任务中效率低的原因及优化

原因

全局解释器锁（GIL）：Python 的 CPython 解释器中存在 GIL，它是一个互斥锁，在任何时刻，只有一个线程能在 CPU 上执行字节码。这意味着即使在多核 CPU 环境下，多线程也无法真正利用多核优势并行执行计算密集型任务，线程之间频繁切换上下文会带来额外开销，导致效率不如单线程。

优化方案

使用多进程：
- 原理：多进程可以利用多核 CPU 的优势，每个进程都有独立的 Python 解释器和内存空间，不存在 GIL 限制。
- 示例：

import multiprocessing


def compute(n):
    result = 0
    for i in range(n):
        result += i
    return result


if __name__ == '__main__':
    num_processes = multiprocessing.cpu_count()
    pool = multiprocessing.Pool(processes=num_processes)
    tasks = [10000000] * num_processes
    results = pool.map(compute, tasks)
    pool.close()
    pool.join()
    total_result = sum(results)
    print(total_result)

使用 C 扩展模块：
- 原理：将计算密集型部分用 C 语言实现，C 语言不受 GIL 限制，可以充分利用多核资源，然后通过 Python 的 C 扩展机制在 Python 中调用。
- 示例：以 cython 为例，首先编写 compute.pyx 文件：

def compute(int n):
    cdef int result = 0
    cdef int i
    for i in range(n):
        result += i
    return result

然后编写 setup.py 文件：

from setuptools import setup
from Cython.Build import cythonize


setup(
    ext_modules = cythonize("compute.pyx")
)

在命令行运行 python setup.py build_ext --inplace 生成 C 扩展模块，就可以在 Python 中像普通模块一样导入使用。 3. 使用异步编程： - 原理：对于 I/O 密集型任务可以将其与计算密集型任务结合，通过 asyncio 等库实现异步操作，在等待 I/O 时释放线程，使计算与 I/O 操作重叠，提高整体效率。虽然不能直接解决计算密集型任务多线程效率低的问题，但可以在混合任务场景下优化。 - 示例：

import asyncio


async def io_bound_task():
    await asyncio.sleep(1)
    return "IO task completed"


async def compute_task():
    result = 0
    for i in range(10000000):
        result += i
    return result


async def main():
    task1 = asyncio.create_task(io_bound_task())
    task2 = asyncio.create_task(compute_task())
    result1, result2 = await asyncio.gather(task1, task2)
    print(result1)
    print(result2)


if __name__ == '__main__':
    asyncio.run(main())

星途面试题库

面试题：Python多线程在计算密集型任务中效率低的原因及优化

知识考点

面试题答案

原因

优化方案