Python 多线程批量检查 URL 状态码：ThreadPoolExecutor + Session

一个真实场景：data/*.json，每个 JSON 是数组，数组元素里有 imgHttpUrl。想批量检查这些 URL 是不是 403（图片失效），是的话把整个 JSON 文件删掉。文件几千个，单线程要跑一天，加多线程。

单线程版本先跑通

import json
from pathlib import Path
import requests

for json_file in Path("data").glob("*.json"):
    try:
        data = json.loads(json_file.read_text(encoding="utf-8"))
        if not data:
            continue

        url = data[0].get("imgHttpUrl")
        if not url:
            continue

        r = requests.get(url, stream=True, timeout=10)
        if r.status_code == 403:
            print(f"删除 {json_file}")
            json_file.unlink()

    except Exception as e:
        print(json_file, e)

要点：

• 只检查数组第一个元素（假设整个文件的图片来自同一源，一个 403 全体 403）
• stream=True — 只拿响应头，body 不下载，快
• 用 pathlib.Path.unlink() 删文件，比 os.remove 干净

加多线程

IO 密集，直接 ThreadPoolExecutor 就行，不用协程：

import json
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests

DATA_DIR = Path("data")
MAX_WORKERS = 100

session = requests.Session()

def check_file(json_file: Path):
    try:
        data = json.loads(json_file.read_text(encoding="utf-8"))
        if not data:
            return None

        url = data[0].get("imgHttpUrl")
        if not url:
            return None

        r = session.get(url, stream=True, timeout=10, allow_redirects=True)
        if r.status_code == 403:
            json_file.unlink()
            return json_file

    except Exception as e:
        print(f"[错误] {json_file}: {e}")
    return None


files = list(DATA_DIR.glob("*.json"))
deleted = []

with ThreadPoolExecutor(max_workers=MAX_WORKERS) as ex:
    futures = {ex.submit(check_file, f): f for f in files}
    for fut in as_completed(futures):
        result = fut.result()
        if result:
            deleted.append(result)

print(f"完成，共删除 {len(deleted)} 个文件")

Session 为什么关键

用 requests.Session() 而不是每次 requests.get()：

• 复用 TCP 连接 + TLS 握手，快 5~10 倍
• 连接池自动管理
• 全局默认 headers / cookies

批量请求同一个域名，Session 是必配的。

HEAD 还是 GET？

只判状态码，不需要 body，理论上 HEAD 更快：

r = session.head(url, timeout=10, allow_redirects=True)

但：

• CDN / 图片服务器可能不支持 HEAD，返回 405 Method Not Allowed（常见于阿里 OSS 未开该权限、部分反爬策略）
• 有些服务器 HEAD 返回和 GET 不同的状态码

保险起见先 HEAD、失败降级 GET：

def check(url):
    try:
        r = session.head(url, timeout=10, allow_redirects=True)
        if r.status_code == 405:
            r = session.get(url, stream=True, timeout=10, allow_redirects=True)
        return r.status_code
    except requests.RequestException:
        return None

线程数怎么调

• CPU 密集：os.cpu_count() 一般顶
• IO 密集（网络请求）：可以开几十甚至几百
• 目标服务器扛得住多少：看它——同源打 100 并发容易被 WAF ban，不同源随便开

试起来：从 20 开始，看服务器响应正常、CPU 也没打满、就慢慢加。100 是 Python requests 常见上限。

更快：httpx + asyncio

千 URL 级别 ThreadPoolExecutor 够用。上万级建议改 asyncio：

import asyncio, httpx, json
from pathlib import Path

async def check(client, json_file):
    try:
        data = json.loads(json_file.read_text(encoding="utf-8"))
        if not data:
            return
        url = data[0].get("imgHttpUrl")
        if not url:
            return
        r = await client.head(url, follow_redirects=True, timeout=10)
        if r.status_code == 403:
            json_file.unlink()
            print(f"删除 {json_file}")
    except Exception as e:
        print(json_file, e)

async def main():
    files = list(Path("data").glob("*.json"))
    limits = httpx.Limits(max_connections=200)
    async with httpx.AsyncClient(limits=limits) as client:
        await asyncio.gather(*(check(client, f) for f in files))

asyncio.run(main())

asyncio + httpx 可以轻松跑到 1000+ 并发，机器压力更小。

一句话总结

IO 密集批量请求 = ThreadPoolExecutor + Session。开 50-100 线程，先 HEAD 再降级 GET。上万量级换 asyncio + httpx。