推荐线程数 = CPU核数

黄金法则：在Python中，线程主要用于I/O并发，CPU并行请用进程。

GIL 锁：同一时刻只有一个线程能执行 Python 字节码。GIL 并非 Python 语言的特性，而是 CPython 解释器的设计选择

GIL 本质是一把互斥锁（Mutex），任何线程要执行 Python 代码，必须先获取 GIL 锁，执行一段字节码后（默认约 100 次操作）或遇到 I/O 阻塞（如网络请求、文件读写，解释器检测到这是一个可能阻塞的 I/O 请求，先主动释放 GIL）时，会主动释放 GIL，让其他线程有机会获取锁并执行。

CPU 密集型任务为何不适合多线程？

多个线程会频繁抢占 GIL；

切换开销大，效率反而更低；

推荐使用多进程（如 multiprocessing 模块），每个进程有独立 GIL，能真正并行执行。

1.进程（Process）

通过 multiprocessing 模块或 concurrent.futures.ProcessPoolExecutor 创建：

主进程和子进程通过 multiprocessing 底层的管道 / 队列通信，所有传递的参数和结果都必须是可序列化的（pickle 支持）

from multiprocessing import Process
import time

def process_task(name):
    print(f"进程 {name} 开始")
    time.sleep(2)  # 模拟计算
    print(f"进程 {name} 结束")

if __name__ == "__main__":
    p1 = Process(target=process_task, args=("A",))
    p2 = Process(target=process_task, args=("B",))
    p1.start()
    p2.start()
    p1.join() #如果没有 join()，主进程可能会在子进程（p1、p2）还没执行完就提前结束
    p2.join()

2. 线程（Thread）

import threading
import time

def thread_task(name):
    print(f"线程 {name} 开始")
    time.sleep(2)  # 模拟I/O等待（释放GIL）
    print(f"线程 {name} 结束")

t1 = threading.Thread(target=thread_task, args=("A",))
t2 = threading.Thread(target=thread_task, args=("B",))
t1.start()
t2.start()
t1.join()
t2.join()

3.协程（Coroutine）

import asyncio
# 用 async def 定义协程函数
async def coro_task(name):  
    print(f"协程 {name} 开始")
    await asyncio.sleep(2)  # 主动交出控制权，事件循环切换到其他协程
    print(f"协程 {name} 结束")

async def main():
    # 创建两个协程任务，由事件循环调度
    task1 = asyncio.create_task(coro_task("A"))
    task2 = asyncio.create_task(coro_task("B"))
    await asyncio.gather(task1, task2)  # 等待所有协程完成

asyncio.run(main())