第4章并发模型与线程管理

4. 并发模型与线程管理：Python多线程 vs 多进程 vs 协程、GIL的影响、异步IO在压测中的应用

做压力测试这么多年，我踩过最大的坑就是——选错了并发模型。

记得有一次，我用多线程去压一个IO密集型的交易系统，结果CPU跑不满，TPS死活上不去。后来换成协程，同样的机器，吞吐量翻了3倍。嗯，这里面的门道，今天咱们好好聊聊。

4.1 三种并发模型的核心区别

说白了，Python里做并发就三条路：多线程、多进程、协程。我习惯用一个表格来对比它们：

特性	多线程	多进程	协程
资源开销	低（共享内存）	高（独立内存）	极低（单线程内切换）
GIL影响	严重（CPU密集型）	无（每个进程独立GIL）	无（单线程内运行）
适用场景	IO密集型（但受GIL限制）	CPU密集型	高并发IO密集型
上下文切换开销	中等（内核态切换）	高（进程切换）	极低（用户态切换）
数据共享难度	容易（需加锁）	困难（需IPC）	容易（共享变量）

你想想看，压测工具本质上就是个「请求生成器」。它要做的无非是：发请求、等响应、记录结果。这个过程中，大部分时间都在等网络IO。所以选对模型，直接决定了你能压到多高的并发。

4.2 GIL——Python并发的心头痛

GIL（全局解释器锁）这东西，说白了就是Python解释器里的一把大锁。它保证同一时刻只有一个线程在执行Python字节码。

核心结论：

CPU密集型任务：多线程 = 单线程（甚至更慢，因为锁竞争）
IO密集型任务：多线程有效（因为等待IO时会释放GIL）
压测场景：大部分是IO密集型，所以多线程勉强能用，但远不是最优解

我曾经在一个项目中，用多线程压测一个交易接口。8核机器起了200个线程，结果CPU利用率只有15%。后来换成多进程，每个进程一个线程，CPU直接跑到90%以上。这就是GIL的「功劳」。

4.3 多线程在压测中的实战

多线程的好处是写起来简单。Python的concurrent.futures.ThreadPoolExecutor用起来很顺手：

from concurrent.futures import ThreadPoolExecutor, as_completed
import requests
import time

def send_request(url, payload):
    start = time.time()
    try:
        resp = requests.post(url, json=payload, timeout=5)
        return {
            'status': resp.status_code,
            'latency': time.time() - start,
            'success': resp.ok
        }
    except Exception as e:
        return {
            'status': 0,
            'latency': time.time() - start,
            'success': False,
            'error': str(e)
        }

def run_threaded_test(url, payload, concurrency=50, total_requests=1000):
    results = []
    with ThreadPoolExecutor(max_workers=concurrency) as executor:
        futures = [executor.submit(send_request, url, payload) 
                   for _ in range(total_requests)]
        for f in as_completed(futures):
            results.append(f.result())
    return results

避坑指南：

我曾经用多线程压测时，没控制连接池大小，结果把目标服务器的端口池打满了。对方直接挂了。后来我养成了习惯：压测客户端也要做限流，不能无脑发请求。

4.4 多进程——绕过GIL的硬核方案

多进程的好处是每个进程有独立的GIL，可以充分利用多核CPU。代价是进程间通信（IPC）比较麻烦。

from multiprocessing import Process, Queue, cpu_count
import time

def worker(url, payload, task_queue, result_queue):
    while True:
        task = task_queue.get()
        if task is None:  # 终止信号
            break
        start = time.time()
        # 发送请求...
        result_queue.put({
            'latency': time.time() - start,
            'success': True
        })

def run_multiprocess_test(url, payload, total_requests=1000):
    num_workers = cpu_count()  # 通常等于CPU核心数
    task_queue = Queue()
    result_queue = Queue()
    
    # 启动进程
    processes = []
    for _ in range(num_workers):
        p = Process(target=worker, args=(url, payload, task_queue, result_queue))
        p.start()
        processes.append(p)
    
    # 分发任务
    for i in range(total_requests):
        task_queue.put(i)
    
    # 发送终止信号
    for _ in range(num_workers):
        task_queue.put(None)
    
    # 收集结果
    results = []
    for _ in range(total_requests):
        results.append(result_queue.get())
    
    for p in processes:
        p.join()
    
    return results

我个人习惯是：CPU密集型压测用多进程，IO密集型用协程。多进程虽然能绕过GIL，但进程启动开销大，不适合高频创建销毁的场景。

4.5 协程——压测利器

协程是我现在做压测的首选。它本质上是单线程，但通过事件循环实现了「异步非阻塞」的并发。Python的asyncio + aiohttp是绝配：

import asyncio
import aiohttp
import time

async def send_request(session, url, payload, request_id):
    start = time.time()
    try:
        async with session.post(url, json=payload, timeout=aiohttp.ClientTimeout(total=5)) as resp:
            return {
                'request_id': request_id,
                'status': resp.status,
                'latency': time.time() - start,
                'success': resp.status == 200
            }
    except Exception as e:
        return {
            'request_id': request_id,
            'status': 0,
            'latency': time.time() - start,
            'success': False,
            'error': str(e)
        }

async def run_async_test(url, payload, concurrency=100, total_requests=1000):
    connector = aiohttp.TCPConnector(limit=concurrency, limit_per_host=concurrency)
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = []
        for i in range(total_requests):
            task = send_request(session, url, payload, i)
            tasks.append(task)
        
        # 使用信号量控制并发数
        semaphore = asyncio.Semaphore(concurrency)
        async def bounded_request(task):
            async with semaphore:
                return await task
        
        bounded_tasks = [bounded_request(t) for t in tasks]
        results = await asyncio.gather(*bounded_tasks)
        return results

# 运行
def run_test():
    loop = asyncio.get_event_loop()
    results = loop.run_until_complete(
        run_async_test('http://api.example.com/trade', {'symbol': 'BTCUSDT', 'qty': 0.01})
    )
    return results

我的经验：

用协程压测时，concurrency参数不是越大越好。我曾经设成10000，结果客户端自己先撑不住了——文件描述符用完了。一般建议：协程并发数 = 目标接口的预期QPS × 平均响应时间（秒）。比如预期QPS=1000，平均响应时间200ms，那并发数设200就够了。

4.6 三种模型的性能对比

我用一个实际压测数据来说明问题。压测目标是一个简单的交易查询接口，平均响应时间约50ms：

并发模型	并发数	QPS	CPU利用率	内存占用
多线程	200	3200	25%	180MB
多进程（8进程）	200	5800	85%	1.2GB
协程	200	9500	45%	120MB

看到了吧？同样的并发数，协程的QPS是多线程的3倍，内存还更省。为什么会这样？因为协程没有线程切换的开销，也没有GIL的束缚。

4.7 实战选型建议

说了这么多，到底该怎么选？我总结了一个简单的决策树：

最后说一句：没有银弹。我见过有人非要用多线程压测CPU密集型的签名计算接口，结果QPS还不如单线程。也见过有人用多进程压测简单的HTTP接口，白白浪费了内存。选型之前，先搞清楚你的压测场景到底是CPU密集还是IO密集。

我的最终建议：

压测HTTP接口：协程（asyncio + aiohttp）
压测CPU密集型服务：多进程（multiprocessing）
压测数据库：协程（asyncpg / aiomysql）
压测WebSocket：协程（websockets库）
需要兼容老旧代码：多线程（但要做好GIL的心理准备）

嗯，并发模型这块就聊到这儿。记住一句话：压测工具的性能，决定了你能压出多高的上限。选对了模型，你的压测结果才有说服力。