Python的并发编程一直是新手和老手都容易搞混的话题。GIL、多线程、多进程、协程、异步IO……这些概念堆在一起确实让人头大。这篇文章不扯复杂的理论直接从实际场景出发讲清楚什么情况该用哪种并发方案。先搞清楚并发和并行是两码事很多人在说并发的时候其实指的是并行这两个概念必须区分清楚。并发多个任务交替执行看起来像是在同时进行。单核CPU就能实现。并行多个任务真正同时执行需要多核CPU的支持。举个生活中的例子你一边吃早饭一边看手机这其实是并发——你在一段时间内交替做两件事。如果要并行你得有两个脑子才行。Python的GIL全局解释器锁导致CPython中的多线程无法实现真正的并行同一时刻只有一个线程在执行Python字节码但可以实现并发。这是理解Python并发编程的起点。一、多线程IO密集型任务的首选Python的多线程虽然受GIL限制但在IO密集型场景下依然是很好的选择。什么是IO密集型任务就是程序大部分时间在等待IO操作完成比如网络请求调用API、爬虫数据库查询文件读写用户输入在这些场景下线程在等待IO时会释放GIL其他线程可以执行所以多线程能显著提升效率。pythonimport threading import time import requests def fetch_url(url): 模拟网络请求 print(f开始请求: {url}) response requests.get(url) print(f完成请求: {url}状态码: {response.status_code}) return response.status_code # 串行执行 — 总耗时 所有请求耗时之和 urls [https://httpbin.org/delay/1] * 5 start time.time() for url in urls: fetch_url(url) print(f串行耗时: {time.time() - start:.2f}秒) # 多线程执行 — 总耗时 ≈ 最慢的那个请求耗时 def worker(url): fetch_url(url) start time.time() threads [] for url in urls: t threading.Thread(targetworker, args(url,)) t.start() threads.append(t) for t in threads: t.join() print(f多线程耗时: {time.time() - start:.2f}秒) # 输出多线程耗时约1秒串行约5秒使用concurrent.futures.ThreadPoolExecutor更简洁pythonfrom concurrent.futures import ThreadPoolExecutor, as_completed def fetch_url(url): response requests.get(url) return url, response.status_code urls [https://httpbin.org/delay/1] * 5 with ThreadPoolExecutor(max_workers5) as executor: # 提交所有任务 futures {executor.submit(fetch_url, url): url for url in urls} # 处理完成的结果 for future in as_completed(futures): url, status future.result() print(f{url} 完成状态码: {status})多线程的优势共享内存方便、切换开销小多线程的局限受GIL限制CPU密集型任务无法利用多核二、多进程CPU密集型任务的选择如果你的任务是CPU密集型的比如图像处理、大数据计算、加密解密、数学运算那么多进程是更好的选择。多进程会创建独立的子进程每个进程有自己的Python解释器和内存空间可以真正利用多核CPU并行执行。pythonimport multiprocessing import time import math def is_prime(n): 判断一个数是否是质数CPU密集型 if n 2: return False for i in range(2, int(math.sqrt(n)) 1): if n % i 0: return False return True def find_primes(numbers): 从列表中找出所有质数 return [n for n in numbers if is_prime(n)] # 生成大量数字 numbers list(range(100000, 200000)) # 串行执行 start time.time() result find_primes(numbers) print(f串行耗时: {time.time() - start:.2f}秒) # 多进程执行 def worker(numbers_chunk): return find_primes(numbers_chunk) start time.time() # 获取CPU核心数 num_cores multiprocessing.cpu_count() chunk_size len(numbers) // num_cores # 分片 chunks [numbers[i:ichunk_size] for i in range(0, len(numbers), chunk_size)] with multiprocessing.Pool(processesnum_cores) as pool: results pool.map(worker, chunks) # 合并结果 all_primes [] for r in results: all_primes.extend(r) print(f多进程耗时: {time.time() - start:.2f}秒) # 多进程通常比串行快3-4倍取决于CPU核心数同样可以用concurrent.futures.ProcessPoolExecutorpythonfrom concurrent.futures import ProcessPoolExecutor with ProcessPoolExecutor(max_workers4) as executor: results executor.map(find_primes, chunks) all_primes [] for r in results: all_primes.extend(r)多进程的优势真并行、利用多核多进程的局限进程创建开销大、数据共享复杂需要序列化、内存占用高三、协程高并发IO的新方案协程是Python 3.4引入asyncio、3.5正式支持async/await语法的并发方案。它比线程更轻量一个线程里可以运行成千上万个协程。协程的核心是主动让出CPU遇到IO等待时协程会主动让出控制权让其他协程执行。这被称为协作式多任务不同于线程的抢占式多任务。pythonimport asyncio import aiohttp import time async def fetch_url(session, url): 异步请求 async with session.get(url) as response: return url, response.status async def main(): urls [https://httpbin.org/delay/1] * 10 async with aiohttp.ClientSession() as session: tasks [fetch_url(session, url) for url in urls] results await asyncio.gather(*tasks) for url, status in results: print(f{url} 完成状态码: {status}) start time.time() asyncio.run(main()) print(f异步耗时: {time.time() - start:.2f}秒) # 10个请求每个延迟1秒总耗时约1秒协程特别适合高并发网络IO场景Web爬虫、实时API网关、聊天服务器、游戏服务器等。常用的异步库aiohttp异步HTTP客户端/服务器aiomysql、aiopg异步数据库驱动aiofiles异步文件操作httpx支持异步的HTTP客户端四、实际项目中的选择策略说了这么多到底怎么选我用一张表帮你决策场景推荐方案原因爬虫、大量API调用协程 或 多线程IO密集型协程更轻量Web后端Django/Flask多线程默认框架自带够用Web后端高并发协程FastAPI/Starlette性能更好资源占用少图像处理、计算密集多进程利用多核CPU混合任务IO计算多进程协程组合各取所长简单定时任务单线程协程够用且简单五、三个真实案例案例一爬虫项目需求每天爬取10000个商品页面的价格信息。方案协程 aiohttp。单机几百个并发连接每秒爬取几百个页面CPU和内存占用都很低。python# 伪代码示意 async def crawl_all(): semaphore asyncio.Semaphore(100) # 限制并发数避免被封 async def crawl_one(url): async with semaphore: async with aiohttp.ClientSession() as session: async with session.get(url) as resp: return await resp.text() tasks [crawl_one(url) for url in urls] pages await asyncio.gather(*tasks)案例二图像批处理需求将一批图片缩放到指定尺寸并添加水印。方案多进程。图片处理是CPU密集型的每个进程处理一批图片充分利用多核CPU。pythonfrom PIL import Image from concurrent.futures import ProcessPoolExecutor def process_image(image_path, output_path): img Image.open(image_path) img img.resize((800, 600)) # 添加水印... img.save(output_path) with ProcessPoolExecutor(max_workers8) as executor: futures [] for img_path, out_path in image_pairs: futures.append(executor.submit(process_image, img_path, out_path)) for future in futures: future.result() # 等待所有完成案例三实时数据流处理需求从Kafka消费消息做轻量处理后再发送出去。方案多线程 协程。多线程负责消费和生产协程处理并发。pythonimport threading import asyncio def kafka_consumer(): 线程从Kafka拉取消息 for message in kafka_consumer_loop(): # 提交到协程事件循环处理 asyncio.run_coroutine_threadsafe( process_message(message), loop ) async def process_message(message): 协程处理单条消息可能涉及多个外部API调用 data await fetch_enrichment(message) await send_to_downstream(data)六、常见坑和避坑指南坑1线程安全多个线程共享变量时要注意数据竞争。使用threading.Lock保护临界区。pythonimport threading counter 0 lock threading.Lock() def increment(): global counter with lock: # 加锁 counter 1坑2多进程数据共享多进程不共享内存传递数据需要序列化。用multiprocessing.Queue、multiprocessing.Manager或者Redis等外部存储。pythonfrom multiprocessing import Manager def worker(shared_list, value): shared_list.append(value) manager Manager() shared_list manager.list()坑3协程中不要用阻塞代码在异步函数中调用requests.get()或time.sleep()会阻塞整个事件循环。必须用异步版本的库。python# ❌ 错误在协程中调用阻塞函数 async def bad_example(): response requests.get(http://example.com) # 阻塞 time.sleep(1) # 阻塞 # ✅ 正确用异步库 async def good_example(): async with aiohttp.ClientSession() as session: async with session.get(http://example.com) as resp: await resp.text() await asyncio.sleep(1) # 异步版本写在最后并发编程是Python进阶路上的一道坎但掌握了核心原理就没那么可怕。记住几个关键点IO密集型用多线程或协程协程更轻量但生态稍弱CPU密集型用多进程真的能利用多核协程不能和阻塞代码混用要用异步版本的库没有银弹根据场景选择最合适的方案最后推荐几个进一步学习的资源官方文档threading、multiprocessing、asyncio模块文档《Python并发编程实战》开源项目FastAPI的源码大量使用异步多动手写实验代码把各种方案都跑一跑比看一百篇文章都有用。