Python的线程、进程与协程分别实现多任务详细过程以及区别对比Python小知识系列又来了这次想认真聊一个让很多人困惑的地方线程、进程、协程到底分别怎么用它们的区别又是啥。我见过太多人一上来就写threading结果发现跑起来没快多少换multiprocessing又踩了序列化的坑后面学asyncio又发现语法好绕搞得初学者直接劝退。这篇文章我想把三者在Python里实现多任务的完整过程讲透从最基础的写法开始再逐步深入到原理和适用场景。我不打算整高深源码分析而是用“够用、能用、不出错”的思路把该知道的细节都铺开讲一遍。最后用一个实际对比的表和几个常见坑的记录收尾希望你读完能直接照着写不用再翻无数篇零散的资料。顺便说明一下我的运行环境是 Python 3.10操作系统是 Windows 11不过代码在 Linux/macOS 上基本一套走只有个别行为和fork相关的地方会略微不同我会在文中标出。1. 先把基础概念搞清楚什么是多任务为什么要学这三种1.1 从单任务程序说起假设你写了一个最简单的下载器功能是从某个接口读数据然后保存到本地文件。在没有并发的情况下代码大概是这样的import time def download(url): print(f开始下载{url}) time.sleep(2) # 模拟网络IO等待 print(f下载完成{url}) return url if __name__ __main__: start time.time() urls [url_1, url_2, url_3, url_4] for url in urls: download(url) print(f总耗时{time.time() - start:.2f}秒)这段代码的耗时差不多是 8 秒4 个任务每个 2 秒。整个程序是一行一行跑下去的一个任务不结束另一个任务永远没机会开始。这就是所谓的“单任务串行”。但实际操作中你要处理的往往是几百上千个这样的任务爬很多页面、批量处理图片、同时监听多个端口、并发发请求。如果还是串行那耗时会基本呈线性增长。这时候就需要“多任务并发”来插手了。1.2 Python里三种实现多任务的方式各是什么我用比较口语的方式来定义线程Thread多个任务在同一个进程内来回切换执行。切换由操作系统调度周期非常短看起来像“同时跑”实际上每个时刻 CPU 上只有一个线程在执行至少在 CPython 下是这样。进程Process每个任务都是独立的一个进程拥有独立的内存空间和系统资源。多个进程可以真正地同时跑在多个 CPU 核心上。协程Coroutine用async/await语法写出来的任务由程序自身控制切换不依赖操作系统。它的切换只发生在await点比线程更轻量。这三种方式都能实现“多任务”但适用场景和性能表现差异极大。先把这个概念放在心里后面写到具体案例时你会感受更明显。1.3 为什么不能只学一种有些朋友会有“反正能跑就行会一种就够”的想法。但真实业务里IO密集型和CPU密集型任务需要完全不同的工具大量网络请求、文件读写 —— 适合协程或线程。大量计算、压缩、加密 —— 适合多进程绕开GIL限制。既要并发又要共享大量状态 —— 可能要用线程加锁来管理。所以三种都得有个基础认知并且知道什么时候选哪种。这篇文章后面会用同一个“模拟多任务”的例子分别用线程、进程、协程实现一遍方便对照。2. 线程实现多任务从入门写法到锁和队列避坑2.1 最简单的线程写法threading.ThreadPython 自带的threading模块是老牌方案。最基础的写法是手动创建Thread对象然后start()启动join()等待结束。import threading import time def download(url): print(f开始下载{url}) time.sleep(2) print(f下载完成{url}当前线程{threading.current_thread().name}) if __name__ __main__: start time.time() urls [url_1, url_2, url_3, url_4] threads [] for url in urls: t threading.Thread(targetdownload, args(url,), namef线程-{url}) threads.append(t) t.start() # 启动线程 for t in threads: t.join() # 等待所有线程结束 print(f总耗时{time.time() - start:.2f}秒)运行结果里你能看到“开始下载”4条打印几乎是瞬间出来的随后 2 秒后陆续输出“下载完成”总耗时大约 2 秒。对比串行的 8 秒效果立竿见影。这里有个细节join()必须要写。如果不写主线程会直接跑完后面的代码程序退出时可能把还没跑完的子线程直接带走。具体表现是有些“下载完成”没打印程序就结束了。这个问题在新手里极常见。2.2 线程池别傻傻地手动创建成百上千个线程当任务量变大比如 500 个请求手动创建 500 个线程绝对是个糟糕的主意。每个线程都有额外的系统资源开销栈空间、调度实体而且 Python 的 GIL 会让大量线程争抢全局锁调度开销暴涨。正确的姿势是使用线程池。Python 提供两种常用方式concurrent.futures.ThreadPoolExecutor高层API好用推荐。threadingqueue.Queue手动实现生产者消费者模型适合需要精确控制的场景。用ThreadPoolExecutor改写上面的例子核心代码如下from concurrent.futures import ThreadPoolExecutor import time def download(url): print(f开始下载{url}) time.sleep(2) return f下载完成{url} if __name__ __main__: start time.time() urls [furl_{i} for i in range(20)] with ThreadPoolExecutor(max_workers4) as executor: results list(executor.map(download, urls)) for r in results: print(r) print(f总耗时{time.time() - start:.2f}秒)max_workers4表示同时最多跑 4 个线程。20 个任务每个 2 秒串行需要 40 秒这个池化版本大约是 5 批每批 2 秒总共 10 秒左右。executor.map会按传入顺序返回结果适合结果收集。如果你更关心“谁先完成谁先处理”可以用executor.submitas_completedfrom concurrent.futures import ThreadPoolExecutor, as_completed with ThreadPoolExecutor(max_workers4) as executor: futures [executor.submit(download, url) for url in urls] for future in as_completed(futures): print(future.result())这两者的使用场景不同map适合顺序结算as_completed适合动态处理先完成的任务。你在实际项目里按需求选即可。2.3 线程共享变量的坑GIL并不是万能锁有朋友听说 Python 有 GIL全局解释器锁会以为线程写共享变量是安全的。其实 GIL 只保证单个字节码指令的执行不被其他线程抢占并不保证一段多步操作的原子性。看看一个经典例子import threading count 0 def add(): global count for _ in range(100000): count 1 if __name__ __main__: threads [threading.Thread(targetadd) for _ in range(10)] for t in threads: t.start() for t in threads: t.join() print(fcount {count})count 1在字节码层面其实是读取、相加、写回三步。多个线程同时操作时会发生互相覆盖导致最终结果远小于 1000000。这就是经典的“竞态条件”。解决方案是用锁lock threading.Lock() def add(): global count for _ in range(100000): with lock: count 1加了锁以后多个线程对count的修改会排队最终结果必然正确。锁的开销在于它会让并发程度下降但数据正确性优先级显然更高。注意with lock是一种非常推荐的写法它能自动释放锁避免程序员忘记release()导致的死锁问题。手动acquire()/release()很容易出错。如果业务场景是生产者消费者模型用queue.Queue可以帮助你把锁的管理放在更安全的位置队列本身就是线程安全的内部已经帮你处理好加锁解锁的逻辑。import queue import threading import time q queue.Queue(maxsize10) def producer(): for i in range(20): q.put(i) print(f生产{i}) time.sleep(0.1) def consumer(): while True: item q.get() if item is None: # 约定 None 为结束信号 break print(f消费{item}) q.task_done() p threading.Thread(targetproducer) c threading.Thread(targetconsumer) p.start() c.start() p.join() q.put(None) # 让消费者退出 c.join()这个模式在实际业务里非常常见比如爬虫负责把 URL 放队列多个下载线程从队列取任务执行。3. 进程实现多任务突破GIL限制的硬件级并行3.1 多进程不是“更好的多线程”很多 Python 新手在学习多进程时想法是我要用多进程替代多线程因为它能利用多核 CPU。方向是对的但要注意进程的开销远大于线程创建/销毁一个进程的成本非常高而且进程之间不共享内存数据要通过特殊机制传递队列、管道、共享内存等。所以它并不适合拿来做大量轻量的 IO 任务而更适合做 CPU 密集计算。先看官方multiprocessing模块最简单用法import multiprocessing import time def calculate(n): print(f进程 {multiprocessing.current_process().name} 开始计算 {n}) time.sleep(1) return n * n if __name__ __main__: start time.time() with multiprocessing.Pool(processes4) as pool: results pool.map(calculate, [1, 2, 3, 4, 5, 6, 7, 8]) print(results) print(f总耗时{time.time() - start:.2f}秒)这里的Pool是进程池processes4会创建 4 个进程。pool.map的作用相当于把可迭代对象分发到进程池里执行再收集结果。这里有一点必须强调如果在 Windows 上使用多进程必须把创建进程的代码放在if __name__ __main__:里面否则会无限递归启动子进程导致异常甚至崩溃。在 Linux 上由于fork的存在这个问题没有那么致命但规范写法是任何平台都加上这个保护。3.2 进程间通信Queue 和 Pipe 常见玩法进程之间不共享内存所以你想知道某个子进程算出了什么必须通过通信机制传回来。最常用的是multiprocessing.Queue用法和线程版本很像但它内部是为进程设计的底层走的是操作系统提供的管道或共享内存。import multiprocessing def worker(q, name): result f{name} 的处理结果 q.put(result) if __name__ __main__: q multiprocessing.Queue() processes [] for i in range(5): p multiprocessing.Process(targetworker, args(q, f任务{i})) processes.append(p) p.start() for p in processes: p.join() while not q.empty(): print(q.get())另一种是Pipe()它返回连接的两端适合两个进程之间一对一通信。import multiprocessing def child(conn): conn.send(来自子进程的消息) conn.close() if __name__ __main__: parent_conn, child_conn multiprocessing.Pipe() p multiprocessing.Process(targetchild, args(child_conn,)) p.start() print(parent_conn.recv()) p.join()用Queue做任务分发的模式和线程版的生产者消费者非常相似区别在于进程队列的传输内容必须能被pickle序列化。3.3 使用 ProcessPoolExecutor 的麻烦与解法和多线程类似也有一个高层APIconcurrent.futures.ProcessPoolExecutor。它的用法和ThreadPoolExecutor几乎一样很多代码只需要换一个类名。实际使用时有一个致命痛点任务函数和参数必须可以被 pickle。如果参数里有 lambda、函数对象内部定义、某些自定义类示例会直接抛出PicklingError。我自己就踩过这个坑写了一个内部函数到ProcessPoolExecutor.map里本地运行没问题一加并发立刻崩。解决办法是把内部函数改到模块顶层用def定义成全局函数或者把参数改成简单数据类型。配合ProcessPoolExecutor实现 CPU 密集任务的示例from concurrent.futures import ProcessPoolExecutor def fib(n): if n 2: return 1 return fib(n-1) fib(n-2) if __name__ __main__: nums [35] * 6 with ProcessPoolExecutor(max_workers3) as executor: results list(executor.map(fib, nums)) print(results)fib是顶层函数可以被 pickle 正常传输。如果你的程序算的是纯计算密集型任务用多进程一般能看到接近核心数的线性提速。4. 协程实现多任务单线程内的异步王者4.1 什么是协程它和线程有什么区别协程的核心思想是任务主动让出控制权等待IO完成后再回来继续跑。这种让出的动作发生在代码里的await处。它不是由操作系统调度而是由事件循环event loop调度。拿现实生活类比线程切换像公司里的主管不断给不同员工安排任务每个安排都有资源开销协程则像一个人同时开很多文档等某个文档的内容加载时先切去处理别的文档等加载完再回来继续看。这段理解的要点是协程遇到IO等待不会傻等而是立刻切换到另一个就绪的协程。所以它的并发能力很强一个线程可以维护上万协程而线程只能维持几百个。4.2 用 asyncio 写第一个异步程序Python 异步编程的核心模块是asyncio。先看一个最简示例import asyncio import time async def download(url): print(f开始下载{url}) await asyncio.sleep(2) # 模拟IO等待 print(f下载完成{url}) async def main(): tasks [download(furl_{i}) for i in range(4)] await asyncio.gather(*tasks) if __name__ __main__: start time.time() asyncio.run(main()) print(f总耗时{time.time() - start:.2f}秒)这里的关键点async def定义一个协程函数。await asyncio.sleep(2)模拟IO等待在这里把控制权交还事件循环。asyncio.gather并发运行多个协程。asyncio.run创建事件循环并运行主协程。跑下来总耗时大约 2 秒4 个任务几乎同时完成效果和线程版类似但这里根本没有创建线程只有一个主线程在跑。值得注意的是time.sleep(2)和await asyncio.sleep(2)是两个完全不同的东西前者会真的阻塞整个线程后者则会挂起当前协程、让事件循环调度其他任务。写异步代码时如果不小心用了同步阻塞库那么整个并发就失效了。4.3 协程下如何真正发起IO请求aiohttp 实践asyncio.sleep只是模拟。真正常见的场景是发 HTTP 请求。标准库的requests是同步的会阻塞整个事件循环异步代码里一定要避免。改用aiohttp或httpx.AsyncClient才能享受到协程并发的好处。下面用aiohttp写一个并发请求例程import asyncio import aiohttp async def fetch(session, url): async with session.get(url) as resp: return await resp.text() async def main(): urls [ https://example.com for _ in range(10) ] async with aiohttp.ClientSession() as session: tasks [fetch(session, url) for url in urls] results await asyncio.gather(*tasks) print(f成功获取 {len(results)} 个页面)用aiohttp的好处是在await session.get()和await resp.text()等待网络返回时事件循环可以切到其他任务实现真正的高并发 IO。如果你用requests在异步里硬写代码会变成同步阻塞并发效果瞬间消失。4.4 多个异步任务之间的协作方式除了asyncio.gather你还需要了解另外两个常用工具asyncio.create_task把协程封装成 Task让其“在后台”运行。asyncio.wait可以自定义等待策略比如FIRST_COMPLETED。示例async def worker(name, delay): await asyncio.sleep(delay) return name async def main(): task1 asyncio.create_task(worker(任务A, 1)) task2 asyncio.create_task(worker(任务B, 2)) done, pending await asyncio.wait( {task1, task2}, return_whenasyncio.ALL_COMPLETED ) for task in done: print(task.result()) asyncio.run(main())在多个任务里如果你需要处理“超时控制”可以用asyncio.wait_fortry: result await asyncio.wait_for(worker(任务C, 3), timeout2) except asyncio.TimeoutError: print(任务超时)编写真正的异步业务时这些控制能力比线程更有优势——你不需要依赖操作系统去强杀或超时管理代码层面就能优雅地控制。5. 进程、线程、协程的全面对比与选择建议5.1 一个表格看明白核心技术差异对比维度线程进程协程并发单位同一进程内的多个执行流多个独立进程单线程内的多个任务是否受GIL限制受限制CPU密集无优势不受限制可多核并行受限制但IO密集影响不大创建/切换开销中等高非常低内存共享共享进程内存独立内存需IPC同线程共享内存通信方式共享变量 锁 / queueQueue、Pipe、共享内存直接函数调用、Future/Task适合场景IO密集、任务量中等的场景CPU密集、需要真正并行高并发IO、网络服务代码复杂度低到中中到高中需要理解 async/await调试难度中等死锁、竞态较高进程隔离较低单线程内调度这张表是选型的核心。看了很多推荐其实有一句话可以概括IO任务优先协程CPU任务优先多进程没有明确上限且你希望代码简单就线程。5.2 什么时候该用线程如果你没有精力学习asyncio或者你依赖的第三方库全是同步阻塞实现比如一堆requests和pandas混合逻辑这时用线程最省心。有几个典型场景调用别人写的同步SDK内部没有异步版本。需要对已有阻塞代码做少量并发提速。程序规模不大不想引入过多异步心智负担。不过要注意线程的数量最好控制在“任务的IO等待时长/任务总耗时比例”决定的合适范围内。不用刻意追求几千个线程ThreadPoolExecutor(max_workers8)这种就够了。线程多了反而让上下文切换开销变大。5.3 什么时候该用进程满足以下条件的场景优先考虑多进程计算量巨大图像处理、矩阵运算、数据变形、没有现成C扩展帮你释放GIL。逻辑本身可以天然切成独立模块模块之间只需要极少的通信。你需要更强的稳定性隔离某个子进程崩溃不影响主进程和其他子进程。同时要意识到进程间通信成本很高。频繁地传大对象是糟糕的设计。如果任务小但数量多Pool.map非常合适如果任务大且每个子进程内部有复杂流程考虑用Queue做任务队列让进程启动后长期存活消费。5.4 什么时候该用协程协程最适合的场景是“大量IO等待”高并发网络爬虫一次抓成千上万页面。微服务网关转发大量请求中间全是网络IO。聊天服务、WebSocket管理连接总数多活跃度较低。文件批量处理等待磁盘读写时释放CPU。协程也有一些限制需要注意必须有配套的异步库如aiohttp、asyncpg、aiofiles否则会卡住整个事件循环。调试时打印堆栈和查看任务状态没那么直观。部分老代码和同步库混用困难需要asyncio.to_thread之类的适配手段。5.5 混合使用也不是不行我见过不少人认为三种方式只能选一种实际生产里往往是混合的。比如外层用asyncio调度并发请求内部对某个CPU密集片段使用asyncio.to_thread或ProcessPoolExecutor。这样既能享受协程的轻量调度又能绕开GIL解决计算块。举个混合例子import asyncio from concurrent.futures import ProcessPoolExecutor def cpu_bound(x): return sum(i * i for i in range(x)) async def main(): loop asyncio.get_running_loop() with ProcessPoolExecutor(max_workers2) as pool: result await loop.run_in_executor(pool, cpu_bound, 100000) print(result) asyncio.run(main())这样写异步负责调度的同时CPU密集任务也能真正并行执行。6. 常见问题与排查实录我踩过的坑和解决方法6.1 多线程没变快反而更慢了可能性一任务本身是CPU密集GIL导致线程间互相争夺开销比串行更大。 排查方法如果代码里极少发生IO操作基本都是循环和计算那么把ThreadPoolExecutor换成ProcessPoolExecutor试试通常能解决问题。可能性二任务最小粒度太小线程创建与切换开销远大于任务本身耗时。这时候可以增大单个任务的耗时或者使用协程来避免线程切换成本。6.2 多进程程序在 Windows 上闪退或无限重启这是初学者最常遇到忘了if __name__ __main__:保护。Windows 下用spawn方式启动子进程时它会重新导入主模块如果代码直接调用Process就会陷入无限递归。正确做法是把所有入口逻辑放到if __name__ __main__:的下面就像我在前面的示例代码里写的那样。6.3 协程里用了 requests发现所有请求变成串行这个问题很隐蔽requests是同步库它会阻塞当前线程整个事件循环暂停。等这个慢请求结束后下一个请求才开始。表现就是代码看着是异步的其实一直是串行。解决办法替换成aiohttp或httpx.AsyncClient。如果第三方库实在没有异步版本就使用await asyncio.to_thread(sync_func, arg)把它放到线程池里执行不阻塞事件循环。6.4 共享变量用锁后性能变得极差锁导致的竞争太多。比如多个线程同时高频更新同一个计数器锁的争抢可能成为系统瓶颈。优化思路使用threading.local让每个线程维护自己的副本处理完再汇总。使用collections.Counter等并发友好的数据结构。降低加锁频率在临界区外先做局部计算最后合并。6.5 进程池的 map 结果顺序不对其实map的结果是按输入顺序返回的如果你发现“顺序不对”通常是用错了方法。as_completed返回的顺序是任务完成的顺序而不是提交顺序。如果你想让最终结果按提交顺序输出用map即可如果你关心“谁先完成先处理”用as_completed。关于ProcessPoolExecutor.map还有一个容易忽略的点它的 chunksize 参数对性能影响不小。对于大量轻量任务调高chunksize可以减少进程间通讯次数。with ProcessPoolExecutor(max_workers4) as executor: results list(executor.map(func, data, chunksize20))6.6 协程任务泄漏没有 await 就结束写异步代码时最常犯的是创建了协程对象但忘记把它加入事件循环。例如async def handle(x): pass coro handle(1) # 这里只是创建了一个协程对象它不会自动执行要让它执行必须await coro或用asyncio.create_task(coro)。否则协程对象会被垃圾回收并抛出一个RuntimeWarning提示coroutine was never awaited。养成习惯创建协程对象后立刻用create_task或直接await不要存着不用。6.7 Windows 多进程里 print 输出混在一起排查时发现多进程的print输出比较乱原因是每个进程独立缓冲写入顺序无法保证。这不是严重bug只要业务逻辑正确输出杂乱可以接受。如果需要结构化日志建议给每个进程设置独立日志文件或用日志模块并按进程号区分。7. 结尾我从项目实战中总结的一些经验聊到这里基本把线程、进程、协程从写法到原理都过了一遍。最后还是想分享几个我在项目实战里的体会希望对你未来的代码有帮助。第一个体会是不要炫技要选最稳妥的模型。有些团队成员习惯性用协程结果整个项目里全是异步API到后面想引入一个同步库却发现处处受阻。反过来也有团队只敢用多线程结果大量并发请求时线程池排队严重。最好是根据团队维护能力和任务类型决定。第二个体会是性能调优时先看瓶颈在哪。我遇到过一个人写的爬虫用协程跑一万个请求速度很快但内存爆炸。后来分析发现问题是每次请求都构造新的ClientSession资源没有复用。类似这种不是模型选错而是代码细节没处理好。如果你发现自己代码运行效果和理论差距很大先检查资源复用、连接复用、变量生命周期再考虑换并发模型。第三个体会是写代码时给并发部分预留好日志和超时。无论用什么模型一旦并发规模上去日志和错误处理都会变成大头。比如线程池里某个任务抛异常如果不用future.result()获取结果异常会被吞掉排查时极其痛苦。协程里也要注意gather的return_exceptionsTrue参数否则一个协程挂了就可能影响大局。这三个并发模型不是三个相互替代的关系更像是三种工具线程像电动螺丝刀进程像冲击钻协程像精密螺丝批。不同场景用不同工具这才是现代 Python 开发者的基本素养。希望这篇内容能让你少走弯路遇到问题时有更多可以依赖的经验而不是重新从零开始。