一、先搞懂核心区别：用「工厂」类比

1. 进程：独立的「工厂」

原理：操作系统给每个工厂分配独立的厂房、设备、原材料（内存、CPU 资源），工厂之间互不干扰，各自独立生产。

关键：每个工厂有自己的「营业执照」（PID 进程号），工厂倒闭（崩溃）不影响其他工厂。

通俗特点：重量级选手，建工厂（创建进程）成本高、慢，但抗造（稳定）。

2. 线程：工厂里的「工人」

原理：一个工厂（进程）里的多个工人（线程），共享厂房、设备（进程内存），一起干活。工人的调度由「工厂老板」（操作系统内核）说了算，你没法精准控制。

关键：工人抢设备（共享变量）时容易打架，需要「排队规则」（锁机制）；一个工人出事（崩溃），整个工厂都得停。

通俗特点：中量级选手，雇工人（创建线程）成本中等，沟通（共享数据）方便，但容易出冲突。

3. 协程：工人手里的「任务清单」

原理：一个工人（单个线程）手里有多个任务清单（协程），干 A 任务需要等物料（IO 等待）时，手动切换去干 B 任务，全程自己掌控节奏，不用老板插手。

关键：所有任务都在一个工人手里，不用跟人抢设备（无需加锁），切换任务（协程）只是换个清单，几乎没成本。

通俗特点：轻量级选手，建清单（创建协程）成本极低，能同时干超多任务，但只能干「等物料」的活（IO 密集），干重活（CPU 密集）会累死一个工人。

二、各自的使用场合：看「活的类型」

1. 进程：干「重体力活」（CPU 密集型）

适用场景：需要全力烧 CPU 的活，比如：大规模数值计算、数据排序、图片 / 视频处理、算法运算。
原因：突破 Python 的 GIL 锁限制，多个工厂（进程）能同时用多核 CPU，相当于多工厂一起生产，效率翻倍。

import multiprocessing
import time

# 定义「工厂的生产任务」：计算某段数字的和
def calc_sum(start, end):
    total = 0
    for i in range(start, end):
        total += i  # 纯重体力计算，无等待
    return total

if __name__ == "__main__":
    start_time = time.time()
    # 拆分任务：4个工厂各算2500万个数（适配4核CPU）
    task_ranges = [(0, 25000000), (25000000, 50000000), 
                   (50000000, 75000000), (75000000, 100000000)]
    
    # 开4个工厂（进程池）
    pool = multiprocessing.Pool(processes=4)
    results = []
    for start, end in task_ranges:
        # 给每个工厂分配任务
        res = pool.apply_async(calc_sum, args=(start, end))
        results.append(res)
    
    # 汇总结果
    total = 0
    for res in results:
        total += res.get()
    
    pool.close()
    pool.join()
    print(f"1亿以内整数和：{total}")
    print(f"多进程耗时：{time.time()-start_time:.2f}秒")  # 约2秒（单进程约8秒）

2.线程：干「等物料的活」（低并发 IO 密集型）

适用场景：需要频繁等外部资源的活（低并发），比如：普通文件读写、少量接口请求、串口通信。
原因：工人等物料（IO 等待）时，其他工人能接着干，不用干等着，比单工人（单线程）效率高，且实现简单。

import threading
import time
import os

# 定义「工人的任务」：写文件（等磁盘IO）
def write_file(file_path, content):
    print(f"工人{threading.current_thread().ident}开始写文件：{file_path}")
    time.sleep(1)  # 模拟等磁盘响应（IO等待）
    with open(file_path, "w", encoding="utf-8") as f:
        f.write(content)
    print(f"工人{threading.current_thread().ident}写完了：{file_path}")

if __name__ == "__main__":
    start_time = time.time()
    # 5个写文件任务
    tasks = [(f"./test_{i}.txt", f"测试内容{i}") for i in range(5)]
    threads = []
    
    # 雇5个工人（线程）
    for file_path, content in tasks:
        t = threading.Thread(target=write_file, args=(file_path, content))
        threads.append(t)
        t.start()
    
    # 等所有工人干完活
    for t in threads:
        t.join()
    
    # 清理测试文件
    for file_path, _ in tasks:
        if os.path.exists(file_path):
            os.remove(file_path)
    
    print(f"多线程耗时：{time.time()-start_time:.2f}秒")  # 约1秒（单线程约5秒）

3.协程，干「海量等物料的活」（高并发 IO 密集型）

适用场景：需要同时处理海量等待型任务，比如：高并发接口请求、爬虫、消息队列消费、实时数据推送。
原因：一个工人能同时管数万份任务清单，切换清单的成本几乎为 0，比多线程效率高 10 倍以上。

import asyncio
import aiohttp  # 需先装：pip install aiohttp
import time

# 定义「任务清单」：异步请求接口（等网络响应）
async def fetch_url(session, url):
    print(f"开始请求：{url}")
    async with session.get(url) as response:
        await response.text()  # 等网络IO响应（非阻塞）
    print(f"完成请求：{url}")

async def main():
    start_time = time.time()
    # 30个接口请求任务（实际可扩展到数万）
    urls = ["https://www.baidu.com"] * 30
    
    # 创建异步会话（相当于工人的工具包）
    async with aiohttp.ClientSession() as session:
        # 把所有任务清单列出来
        tasks = [fetch_url(session, url) for url in urls]
        # 让工人按清单干活（并发执行）
        await asyncio.gather(*tasks)
    
    print(f"协程耗时：{time.time()-start_time:.2f}秒")  # 约1秒（单线程同步约30秒）

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

三 总结

1. CPU 密集选进程：比如计算、排序、处理数据，靠多工厂（多核）提效率；
2. IO 低并发选线程：比如少量文件读写、几个接口请求，实现简单成本低；
3. IO 高并发选协程：比如爬虫、高并发接口、消息消费，单个工人干海量活，极致高效。