如何实现 Python 的并发编程？threading 与 multiprocessing-Python教程-PHP中文网

如何实现 Python 的并发编程？threading 与 multiprocessing

夜晨

发布： 2025-09-05 23:02:02

原创

181人浏览过

Python threading和multiprocessing的核心区别在于：threading受GIL限制，无法实现CPU并行，适合I/O密集型任务；multiprocessing创建独立进程，绕开GIL，可利用多核实现真正并行，适合CPU密集型任务。1. threading共享内存、开销小，但GIL导致多线程不能并行执行Python代码；2. multiprocessing进程隔离、通信复杂、启动开销大，但能充分发挥多核性能。因此，I/O密集型任务应选择threading以高效切换等待，CPU密集型任务应选择multiprocessing以实现并行计算。

如何实现 python 的并发编程？threading 与 multiprocessing

Python 的并发编程主要依赖两个核心模块：

threading

登录后复制

和

multiprocessing

登录后复制

。简单来说，如果你处理的是大量等待外部响应（比如网络请求、文件读写）的 I/O 密集型任务，

threading

登录后复制

常常是首选，因为它开销小。但如果你的任务是计算量巨大、需要榨干 CPU 性能的计算密集型任务，那么

multiprocessing

登录后复制

才是正解，因为它能让你真正利用多核 CPU，绕开那个著名的 GIL（全局解释器锁）的限制。

要实现 Python 的并发，我们通常会从这两个模块入手。它们代表了两种不同的并发模型：线程（

threading

登录后复制

）和进程（

multiprocessing

登录后复制

）。

先说说

threading

登录后复制

。它允许你在同一个进程内创建多个执行流，这些线程共享进程的内存空间。这听起来很美，内存共享意味着数据交换方便。但问题是，CPython 有个“全局解释器锁”——GIL。这个锁规定了在任何时刻，只有一个线程能执行 Python 字节码。所以，尽管你有多个线程，但它们在同一时间点上，只有一个能真正跑起来。这意味着，对于纯粹的 CPU 密集型任务，

threading

登录后复制

无法实现真正的并行计算，因为它本质上是并发而非并行。它更擅长的是在等待 I/O 时切换到另一个线程，这样 CPU 就不会闲着。

import threading
import time

def task_io_bound(name):
    print(f"线程 {name}: 开始执行 I/O 密集型任务...")
    time.sleep(2) # 模拟 I/O 等待
    print(f"线程 {name}: 任务完成。")

threads = []
for i in range(3):
    thread = threading.Thread(target=task_io_bound, args=(f"T{i}",))
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()
print("所有 I/O 密集型线程任务完成。")

登录后复制

然后是

multiprocessing

登录后复制

。这个模块就直接多了，它创建的是独立的进程，每个进程都有自己的 Python 解释器和内存空间。进程之间的数据是隔离的，所以它们不会受到 GIL 的限制。每个进程都能在自己的 CPU 核上独立运行，从而实现真正的并行计算。当然，进程间的通信就需要额外的机制，比如队列（

Queue

登录后复制

）或管道（

Pipe

登录后复制

）。它的缺点是启动开销相对大，因为要复制整个进程环境。

豆包AI编程

豆包推出的AI编程助手

483

查看详情

立即学习“Python免费学习笔记（深入）”；

import multiprocessing
import time
import os

def task_cpu_bound(name):
    print(f"进程 {name} (PID: {os.getpid()}): 开始执行 CPU 密集型任务...")
    result = 0
    for _ in range(10_000_000): # 模拟 CPU 密集计算
        result += 1
    print(f"进程 {name} (PID: {os.getpid()}): 任务完成，结果 {result}。")

if __name__ == '__main__': # 确保在 Windows 上能正常运行
    processes = []
    for i in range(3):
        process = multiprocessing.Process(target=task_cpu_bound, args=(f"P{i}",))
        processes.append(process)
        process.start()

    for process in processes:
        process.join()
    print("所有 CPU 密集型进程任务完成。")

登录后复制

选择哪种方式，说白了，就是看你的任务瓶颈在哪里。I/O 等待多，选