Python的并发编程一直是新手和老手都容易搞混的话题。GIL、多线程、多进程、协程、异步IO……这些概念堆在一起，确实让人头大。这篇文章不扯复杂的理论，直接从实际场景出发，讲清楚什么情况该用哪种并发方案。

先搞清楚：并发和并行是两码事

很多人在说"并发"的时候其实指的是"并行"，这两个概念必须区分清楚。

• 并发：多个任务交替执行，看起来像是在同时进行。单核CPU就能实现。

• 并行：多个任务真正同时执行，需要多核CPU的支持。

举个生活中的例子：你一边吃早饭一边看手机，这其实是"并发"——你在一段时间内交替做两件事。如果要"并行"，你得有两个脑子才行。

Python的GIL（全局解释器锁）导致CPython中的多线程无法实现真正的并行（同一时刻只有一个线程在执行Python字节码），但可以实现并发。这是理解Python并发编程的起点。

一、多线程：IO密集型任务的首选

Python的多线程虽然受GIL限制，但在IO密集型场景下依然是很好的选择。

什么是IO密集型任务？就是程序大部分时间在等待IO操作完成，比如：

• 网络请求（调用API、爬虫）

• 数据库查询

• 文件读写

• 用户输入

在这些场景下，线程在等待IO时会释放GIL，其他线程可以执行，所以多线程能显著提升效率。

python

import 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(target=worker, args=(url,))
    t.start()
    threads.append(t)

for t in threads:
    t.join()
print(f"多线程耗时: {time.time() - start:.2f}秒")
# 输出：多线程耗时约1秒，串行约5秒

使用concurrent.futures.ThreadPoolExecutor更简洁：

python

from 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_workers=5) 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并行执行。

python

import 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:i+chunk_size] for i in range(0, len(numbers), chunk_size)]

with multiprocessing.Pool(processes=num_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.ProcessPoolExecutor：

python

from concurrent.futures import ProcessPoolExecutor

with ProcessPoolExecutor(max_workers=4) 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等待时，协程会主动让出控制权，让其他协程执行。这被称为"协作式多任务"，不同于线程的"抢占式多任务"。

python

import 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。

python

from 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_workers=8) 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消费消息，做轻量处理后再发送出去。

方案：多线程 + 协程。多线程负责消费和生产，协程处理并发。

python

import 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保护临界区。

python

import threading

counter = 0
lock = threading.Lock()

def increment():
    global counter
    with lock:  # 加锁
        counter += 1

坑2：多进程数据共享

多进程不共享内存，传递数据需要序列化。用multiprocessing.Queue、multiprocessing.Manager或者Redis等外部存储。

python

from 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的源码（大量使用异步）

多动手写实验代码，把各种方案都跑一跑，比看一百篇文章都有用。