Python多线程与多进程全解：9 大核心知识点 + 实战案例，并发编程加速神器 – 小白编程笔记

图1：Python多线程与多进程 – 核心知识体系总览

一、进程 vs 线程：核心区别（必懂，并发基础）

学习Python多线程与多进程，首先要搞懂最基础的两个概念：进程和线程，这是并发编程的核心前提。

• 进程（Process）：是一个独立的程序运行实例，拥有自己独立的内存空间、系统资源，资源占用大，切换速度慢。比如你打开一个浏览器，就是启动了一个浏览器进程。
• 线程（Thread）：是进程内的执行单元，同一个进程下的多个线程共享进程的内存和资源，资源占用小，切换速度快。比如浏览器里的标签页、下载任务，都是同一个进程下的不同线程。
• 核心关系：一个进程可以包含多个线程，线程不能独立存在，必须依附于进程。
• Python GIL 锁（全局解释器锁）：这是Python特有的机制，它限制了同一个进程下的多线程，只能在单核CPU上并发执行，无法真正多核并行；而多进程因为有独立的内存空间，每个进程都有自己的GIL锁，所以能真正利用多核CPU并行计算。这是Python多线程与多进程最核心的区别。

怎么选择？（选型指南，新手必记）

• I/O 密集型任务（比如网络请求、文件读写、数据库操作）→ 用 多线程。这类任务大部分时间都在等待I/O完成，多线程能在等待时切换执行其他任务，大幅提升效率，而且资源占用小。
• CPU 密集型任务（比如大量计算、数据分析、视频编码）→ 用 多进程。这类任务需要持续占用CPU，多线程受GIL锁限制无法多核并行，多进程能真正利用多核CPU，实现真正的并行加速。

图2：Python多线程与多进程 – 进程vs线程核心区别对比

图3：Python多线程与多进程 – GIL锁对并行的影响示意图

二、多线程 threading 基础使用（I/O密集型首选）

Python多线程与多进程中，Python内置了 threading 模块来实现多线程，无需额外安装，直接导入就能使用，是I/O密集型任务的首选方案。

基础案例：创建并启动多线程

# 导入所需模块
import threading
import time

# 定义线程要执行的任务函数
def task(name):
    """
    线程任务函数，模拟I/O密集型任务（比如网络请求、文件读写）
    :param name: 线程名称，用于标识不同线程
    """
    print(f"线程 {name} 开始执行")
    time.sleep(2)  # 模拟I/O等待（比如等待网络响应、文件读取）
    print(f"线程 {name} 执行完毕")

# 多进程/多线程程序必须写在 if __name__ == '__main__': 内部，避免Windows系统无限创建进程
if __name__ == '__main__':
    # 1. 创建线程对象
    # target参数：指定线程要执行的函数
    # args参数：给函数传递参数，必须是元组形式（即使只有一个参数，也要加逗号）
    t1 = threading.Thread(target=task, args=("A",))
    t2 = threading.Thread(target=task, args=("B",))

    # 2. 启动线程（调用start()方法，线程才会真正开始执行）
    t1.start()
    t2.start()

    # 3. 等待线程执行完毕（join()方法会阻塞主线程，直到子线程执行完成）
    # 必须加join()，否则主线程会直接结束，子线程还没执行完就被强制终止
    t1.join()
    t2.join()

    print("所有线程执行完成，主线程结束")

线程常用方法（新手必记）

• start()：启动线程，线程进入就绪状态，等待CPU调度执行
• join([timeout])：阻塞主线程，等待子线程执行完毕，timeout是可选的超时时间
• is_alive()：判断线程是否存活（是否正在执行）
• threading.current_thread()：获取当前正在执行的线程对象
• threading.active_count()：获取当前存活的线程数量

三、线程安全 Lock 锁（解决多线程数据混乱问题）

Python多线程与多进程中，多线程共享进程的全局变量，当多个线程同时修改同一个全局变量时，会出现数据错乱的问题，必须加锁来保证线程安全。

import threading

# 定义全局变量，多个线程会同时修改这个变量
num = 0
# 创建互斥锁（Lock），用于保证线程安全
lock = threading.Lock()

def add_num():
    """
    对全局变量num进行累加操作，每个线程执行10万次累加
    """
    global num  # 声明使用全局变量num
    for _ in range(100000):
        # 加锁：同一时间只有一个线程能进入锁内的代码，其他线程阻塞等待
        lock.acquire()
        num += 1
        # 释放锁：释放后其他线程才能获取锁执行
        lock.release()

# 创建两个线程，同时执行add_num函数
t1 = threading.Thread(target=add_num)
t2 = threading.Thread(target=add_num)

# 启动线程
t1.start()
t2.start()

# 等待两个线程执行完毕
t1.join()
t2.join()

# 加锁后结果正确：200000（两个线程各累加10万次）
# 不加锁会出现数据错乱，结果远小于200000
print(f"最终num的值：{num}")

四、线程池 ThreadPoolExecutor（企业级推荐写法）

Python多线程与多进程中，手动创建线程的方式适合少量任务，当任务量很大时，手动创建线程会导致系统资源耗尽、程序卡死。企业级开发推荐用 ThreadPoolExecutor 线程池，由系统自动管理线程，代码更简洁、高效，还能控制线程数量。

# 从concurrent.futures模块导入线程池
from concurrent.futures import ThreadPoolExecutor
import time

def task(name):
    """
    线程任务函数，模拟I/O密集型任务
    :param name: 任务编号
    :return: 任务执行结果
    """
    print(f"任务 {name} 开始执行")
    time.sleep(1)  # 模拟I/O等待
    return f"任务 {name} 执行完成"

if __name__ == '__main__':
    # 创建线程池，max_workers参数指定最多同时运行的线程数量（根据任务量和系统配置调整）
    # 用with上下文管理器，自动管理线程池，任务完成后自动关闭
    with ThreadPoolExecutor(max_workers=2) as executor:
        # 方式1：submit()提交单个任务，返回Future对象，可通过result()获取结果
        futures = [executor.submit(task, i) for i in range(5)]
        
        # 遍历Future对象，获取任务执行结果
        for f in futures:
            print(f.result())
        
        # 方式2：map()批量提交任务，直接返回结果迭代器（更简洁）
        # results = executor.map(task, range(5))
        # for res in results:
        #     print(res)

五、多进程 multiprocessing（CPU密集型任务专用）

Python多线程与多进程中，多进程能避开GIL锁的限制，真正利用多核CPU并行计算，是CPU密集型任务的唯一选择。Python内置了 multiprocessing 模块实现多进程，用法和threading几乎一致，上手成本低。

import multiprocessing
import time

def cpu_task():
    """
    模拟CPU密集型任务：大量循环计算，持续占用CPU
    """
    count = 0
    for i in range(100000000):  # 1亿次循环，模拟大量计算
        count += 1

if __name__ == '__main__':
    # 创建进程对象，用法和threading.Thread完全一致
    p1 = multiprocessing.Process(target=cpu_task)
    p2 = multiprocessing.Process(target=cpu_task)

    # 启动进程
    p1.start()
    p2.start()

    # 等待进程执行完毕
    p1.join()
    p2.join()

    print("多进程执行完成，两个进程利用双核CPU并行计算，速度大幅提升")

六、进程池 ProcessPoolExecutor（CPU密集型企业级写法）

和线程池一样，多进程也有对应的进程池 ProcessPoolExecutor，由系统自动管理进程，控制进程数量，避免手动创建进程的繁琐，适合大量CPU密集型任务的并行处理。

from concurrent.futures import ProcessPoolExecutor

def calc(num):
    """
    计算任务：求平方，模拟CPU密集型计算
    :param num: 输入数字
    :return: 平方结果
    """
    return num * num

if __name__ == '__main__':
    # 创建进程池，默认max_workers为CPU核心数，充分利用多核
    with ProcessPoolExecutor() as executor:
        # map()批量提交任务，直接返回结果迭代器
        results = executor.map(calc, [1, 2, 3, 4, 5])
        # 遍历结果
        for res in results:
            print(f"计算结果：{res}")

七、经典实战案例（直接套用，小白友好）

下面给大家整理了Python多线程与多进程最常用的2个实战案例，每个案例都加了详细注释，新手可以直接复制使用，也可以根据需求修改。

案例1：多线程批量下载网页（I/O密集型，线程池实战）

# 导入所需模块
import requests
from concurrent.futures import ThreadPoolExecutor

def download(url):
    """
    下载网页内容，模拟I/O密集型任务
    :param url: 要下载的网页URL
    """
    # 发送GET请求，获取网页内容
    res = requests.get(url, timeout=10)
    # 打印下载结果：URL和内容大小
    print(f"下载完成：{url}, 内容大小：{len(res.content)} 字节")

if __name__ == '__main__':
    # 待下载的URL列表（这里用百度URL重复5次，模拟多个下载任务）
    urls = ["https://www.baidu.com"] * 5
    # 创建线程池，max_workers=3，最多同时3个线程下载，避免请求过于频繁被封
    with ThreadPoolExecutor(max_workers=3) as executor:
        # 批量提交任务，map()自动处理结果
        executor.map(download, urls)

案例2：多进程加速大数据计算（CPU密集型，进程池实战）

import time
from concurrent.futures import ProcessPoolExecutor

# 模拟CPU密集型任务：2500万次累加计算
def heavy_calc_task(n):
    """
    大量循环计算，持续占用CPU，模拟CPU密集型任务
    :param n: 任务编号
    :return: 任务执行结果
    """
    count = 0
    for _ in range(25000000):  # 2500万次循环
        count += 1
    return f"任务{n}计算完成，最终结果：{count}"

if __name__ == '__main__':
    # ========== 单进程串行执行（对比用） ==========
    start_time = time.time()
    # 4个任务串行执行，总1亿次计算
    for i in range(4):
        heavy_calc_task(i)
    single_process_time = time.time() - start_time
    print(f"单进程串行执行总耗时：{single_process_time:.2f} 秒")

    # ========== 多进程并行执行（利用4核CPU） ==========
    start_time = time.time()
    # 创建进程池，max_workers=4，对应4核CPU，充分利用多核
    with ProcessPoolExecutor(max_workers=4) as executor:
        # 批量提交4个任务，并行执行
        results = executor.map(heavy_calc_task, range(4))
        # 打印任务结果
        for res in results:
            print(res)
    multi_process_time = time.time() - start_time
    print(f"多进程并行执行总耗时：{multi_process_time:.2f} 秒")

    # 打印加速比
    print(f"多进程加速比：{single_process_time / multi_process_time:.2f} 倍")

运行效果：单进程串行执行总耗时约4秒，多进程并行执行耗时约1秒，实现了接近4倍的加速效果，完美适配数据分析、科学计算、视频处理等CPU密集型场景，这就是Python多线程与多进程的核心实战价值。

八、高频易错点（避坑指南，新手必看）

整理了Python多线程与多进程开发中，新手最容易踩的4个坑，每个坑都标了原因和解决方法，看完这些，再也不会因为这些问题报错。

九、核心知识点总结（快速复习）