never-primp 2.4.2

基于原primp 用rust重构调整的python请求库 - The fastest python HTTP client that can impersonate web browsers

NEVER_PRIMP

基于 Rust + wreq 的高性能 Python HTTP 客户端 专为网络爬虫、浏览器指纹伪装与风控绕过设计

安装 · 快速开始 · 性能原理 · Cookie 管理 · 浏览器伪装 · API 参考

---

为什么选择 NEVER_PRIMP？

功能	NEVER_PRIMP	requests	httpx	curl-cffi
浏览器 TLS/JA3/JA4 指纹	✅ 100+ 配置	❌	❌	✅ 有限
HTTP/2 指纹（AKAMAI）	✅	❌	❌	✅
请求头顺序精确控制	✅	❌	❌	❌
Cookie 分割（HTTP/2 风格）	✅	❌	❌	❌
跨子域名 Cookie 共享	✅ RFC 6265	✅	✅	❌
Cookie 跨会话持久化	✅	❌	❌	❌
高并发无锁 Client 共享	✅	❌	✅	❌
GIL 释放（真实并行）	✅	❌	✅	❌
重试策略	✅ 内置预算	❌	❌	❌
文件上传（multipart）	✅	✅	✅	✅

性能基准（测试 URL: https://www.baidu.com）

	requests	httpx	curl-cffi	never_primp
单次请求	646 ms	90 ms	122 ms	86 ms
串行 10 次	655 ms	20 ms	47 ms	19 ms
并发 100 任务	697 ms	23 ms	56 ms	20 ms

---

安装

pip install -U never-primp

平台支持：Linux (x86_64/aarch64) · Windows (x86_64) · macOS (x86_64/ARM64)

从源码构建：

pip install maturin
maturin develop --release

---

快速开始

import never_primp

# 最简单的用法
r = never_primp.get("https://httpbin.org/get")
print(r.status_code, r.json())

# 带浏览器指纹的 Client
client = never_primp.Client(
    impersonate="chrome_143",
    impersonate_os="windows",
    timeout=30.0,
)
r = client.get("https://httpbin.org/headers")
print(r.json())

# 上下文管理器
with never_primp.Client(impersonate="firefox_147") as client:
    r = client.post("https://httpbin.org/post", json={"key": "value"})
    print(r.json())

---

性能优化原理深度解析

这一节详细解释 never_primp 每一项性能设计的底层原理。

1. 去除 Arc<Mutex<Client>>：消灭并发瓶颈

旧版问题

// 旧版：所有线程争同一把锁
pub struct RClient {
    client: Arc<Mutex<wreq::Client>>,  // 问题所在
}

// 每次发请求：
let resp = client.lock().unwrap().request(...).send().await;
//                ^^^^^^^^^^^
//                持锁期间，其他线程全部阻塞等待！

在 Python 的 ThreadPoolExecutor 场景下，20 个线程各自调用 client.get()：

线程1  ──[lock]──── request ──── [unlock]──
线程2             [等待] ────── request ──── [等待]...
线程3                       [等待] ...
                             串行化！

新版方案

wreq::Client 内部已用 Arc 包裹所有状态（连接池、Cookie Jar、配置），它实现了 Clone + Send + Sync：

// 新版：直接存储，clone 是零成本的 Arc 引用计数+1
pub struct RClient {
    client: wreq::Client,   // 内部已是 Arc<Inner>
}

// 请求时 clone 传入 async 块
let client = self.client.clone();  // 仅增加引用计数，O(1)，无锁
let future = async move {
    client.request(method, url).send().await
};
py.detach(|| RUNTIME.block_on(future));  // GIL 释放后真正并行

20 个线程同时发请求：

线程1  ── clone(O1) ─── request ───────── 并行！
线程2  ── clone(O1) ─── request ───────── 并行！
线程3  ── clone(O1) ─── request ───────── 并行！
           ↑ 每个 clone 只是原子计数+1，互不干扰

性能影响：高并发场景下吞吐量从串行变为真正并行，延迟从 O(N×T) 降为 O(T)。

---

2. GIL 释放：Python 多线程的正确姿势

Python 的 GIL（Global Interpreter Lock）保证同一时刻只有一个线程执行 Python 字节码，但 IO 操作期间可以释放。

never_primp 的实现：

// 进入 IO 之前显式释放 GIL
let wreq_response = py.detach(|| {
    RUNTIME.block_on(future)  // 整个 HTTP 请求期间 GIL 释放
});
// GIL 在这里自动重新获取

py.detach() 等价于在 C 扩展中调用 Py_BEGIN_ALLOW_THREADS，让其他 Python 线程在等待 HTTP 响应期间正常运行。

实际效果：

GIL 持有时间线（旧版无释放）：
Thread1: [GIL][send HTTP][wait response][GIL] ...
Thread2:            [等待GIL..............][GIL][send HTTP]...
结果：几乎串行

GIL 释放时间线（新版）：
Thread1: [GIL][send HTTP─────────]→[wait resp in Tokio]→[GIL][return]
Thread2:      [GIL][send HTTP─────]→[wait resp in Tokio]→[GIL][return]
Thread3:           [GIL][send HTTP]→[wait resp in Tokio]→[GIL][return]
结果：真正并行

GIL 只在构建请求和处理响应这两个极短的 CPU 阶段被持有，网络 IO 等待期间全部并行。

---

3. Tokio 异步运行时：4 Worker 线程的选择

// src/runtime.rs
pub static RUNTIME: LazyLock<Runtime> = LazyLock::new(|| {
    tokio::runtime::Builder::new_multi_thread()
        .worker_threads(4)       // 4 个 worker 线程
        .thread_name("never-primp-worker")
        .enable_all()
        .build()
        .unwrap()
});

为什么是 4 而不是更多？

Tokio 的 worker 线程处理的是 IO 事件（socket 可读/可写的通知），而不是实际等待数据。HTTP 请求的主要时间消耗在网络 RTT，不是 CPU 计算：

典型 HTTP 请求生命周期：
  建立连接: ~5ms  (CPU: <1ms)
  发送请求: ~0.1ms (CPU: <0.1ms)
  等待响应: ~50ms  (CPU: 0，纯等待)
  读取响应: ~1ms   (CPU: <0.5ms)

4 个 worker 可以同时管理数千个 "等待响应" 状态的 socket，
因为 epoll/IOCP 一次系统调用可以轮询所有就绪事件。

对于 CPU 密集的任务应该用 spawn_blocking，对于 IO 密集的请求任务，4 个 worker 通常已经饱和。增加到 16 个 worker 对 IO bound 场景几乎无提升，反而增加线程切换开销。

---

4. 连接池：TCP 复用的关键参数

client_builder
    .pool_max_idle_per_host(512)       // 每主机保留 512 条空闲 TCP 连接
    .pool_max_size(2048)               // 全局上限：2048 条
    .pool_idle_timeout(Duration::from_secs(90))  // 90s 内未使用则关闭
    .tcp_keepalive(Duration::from_secs(15))       // SO_KEEPALIVE
    .tcp_keepalive_interval(Duration::from_secs(5))
    .tcp_keepalive_retries(3u32)
    .tcp_nodelay(true);                // 禁用 Nagle 算法

为什么这些参数很重要？

TCP 连接复用（最大收益）：

建立一条 TCP+TLS 连接的开销：

TCP 三次握手:  ~10ms (1 RTT)
TLS 1.3 握手: ~15ms (1 RTT)
合计:          ~25ms

如果每个请求都新建连接（pool 为 0），100 个请求就浪费 2500ms 在握手上。连接池让同一主机的请求复用已建立的连接，第 2 次请求开始几乎没有建连开销。

为什么每主机 512 空闲连接？

对于爬虫场景，可能同时对同一主机发起大量并发请求（比如爬取一个网站的 100 个页面）。池大小决定了"瞬间并发"能复用多少连接：

• 太小（如 10）：超出部分需重新建连，浪费 25ms/请求
• 512：几乎不会出现需要新建连接的情况

TCP Keepalive（连接稳定性）：

没有 Keepalive 的问题：
  服务器/NAT/防火墙在 60-120s 内无流量会 silently drop 连接
  下次使用这条"死连接"时：RST 或超时，请求失败

SO_KEEPALIVE 工作原理：
  每 15s 发一个 TCP ACK 探测包（几乎无流量）
  服务器响应 → 连接确认活跃，重置超时计时器
  服务器无响应 → 5s 后重试，最多 3 次，才判定连接断开

效果：空闲连接保持真正可用，避免"僵尸连接"导致的请求失败

TCP_NODELAY（降低小包延迟）：

Nagle 算法会把小数据包积攒到 MSS（约 1460 字节）再发送，降低延迟对 HTTP 不友好。tcp_nodelay(true) 禁用它，每次 write() 立即发送，降低请求延迟约 10-40ms。

---

5. Cookie Jar 的 RFC 6265 实现

wreq 的 Jar 内部结构：

HashMap<domain, HashMap<path, CookieJar>>
   ↑ 按 (domain, path) 二级索引存储

子域名 Cookie 共享（domain_match）：

fn domain_match(host: &str, domain: &str) -> bool {
    host == domain                          // 完全匹配
    || (host.len() > domain.len()
        && host.ends_with(domain)
        && host.as_bytes()[host.len() - domain.len() - 1] == b'.')
    //  ↑ 确保是真子域名：api.example.com 匹配 example.com
    //    但 notexample.com 不匹配 example.com
}

查询 api.example.com 的 cookie 时，遍历所有 domain key 做 domain_match：

• example.com → 匹配，返回其 cookie
• api.example.com → 完全匹配，也返回

读写锁而非互斥锁：

store: Arc<RwLock<HashMap<...>>>
//         ↑ 读写锁

// 查询时（大多数操作）：多个线程并发读，互不阻塞
store.read().get(host)...

// 修改时（Set-Cookie 响应）：独占写锁
store.write().entry(domain).or_default()...

爬虫场景下读（发请求带 Cookie）远多于写（收到 Set-Cookie），RwLock 比 Mutex 在并发读时效率高得多。

---

6. 重试预算机制

// 内置令牌桶重试预算
let policy = RetryPolicy::default()
    .max_retries_per_request(2);  // 每请求最多重试 2 次

wreq 的重试策略内置了预算限制（默认 20% 额外负载）：

假设发出 1000 个请求：
  正常请求: 1000
  允许的重试: 1000 × 20% = 200

如果某段时间重试过多（>200），
预算耗尽，后续请求即使失败也不再重试，
避免雪崩效应（retry storm）导致服务器更不稳定。

这比简单的 for _ in range(3): try: request() 更智能，在大并发场景下保护目标服务器。

---

7. 头部顺序控制（anti-bot 核心）

现代风控系统（Cloudflare、Akamai、PerimeterX）会检测请求头的顺序作为 bot 指纹：

真实 Chrome 143 的头部顺序：
  :method: GET
  :authority: example.com
  :scheme: https
  :path: /
  sec-ch-ua: ...
  sec-ch-ua-mobile: ?0
  sec-ch-ua-platform: "Windows"
  upgrade-insecure-requests: 1
  user-agent: Mozilla/5.0...
  accept: text/html,...
  sec-fetch-site: none
  sec-fetch-mode: navigate
  sec-fetch-user: ?1
  sec-fetch-dest: document
  accept-encoding: gzip, deflate, br, zstd
  accept-language: zh-CN,...

never_primp 用 OrigHeaderMap 记录头部插入顺序：

// 普通 HeaderMap 只保证 O(1) 查找，不保证顺序
// OrigHeaderMap 是有序列表，精确控制发送顺序
let mut orig_headers = OrigHeaderMap::new();
for (key, _) in client_headers.iter() {
    orig_headers.insert(key.clone());  // 保留插入顺序
}
request_builder = request_builder.orig_headers(orig_headers);

这确保了发出去的 TCP 字节流中头部顺序与真实浏览器完全一致。

---

Cookie 管理

自动跨子域名共享

client = never_primp.Client(impersonate="chrome_143")

# 登录后，服务器设置 domain=.example.com 的 Cookie
client.get("https://example.com/login")  # 自动存储 Set-Cookie

# 访问子域名时，这些 Cookie 自动发送（RFC 6265）
client.get("https://api.example.com/data")   # 自动带 Cookie
client.get("https://cdn.example.com/asset")  # 自动带 Cookie

手动 Cookie 管理

# 设置带域名属性的 Cookie（对所有子域名生效）
client.set_cookie("auth_token", "eyJhbGci...",
                  url="https://example.com",
                  domain="example.com",
                  path="/")

# 查询特定域名（含子域名匹配）
cookies = client.get_cookies_for_domain("api.example.com")

# 批量获取（返回完整元数据）
for name, value, domain, path in client.get_all_cookies():
    print(f"[{domain}{path}] {name}={value}")

跨会话 Cookie 持久化

import json

# 会话一：登录
client = never_primp.Client(impersonate="chrome_143")
client.get("https://example.com/login")

# 导出并保存
cookies = client.export_cookies()
with open("session.json", "w") as f:
    json.dump(cookies, f)

# ─── 程序重启 ───

# 会话二：恢复登录状态
client2 = never_primp.Client(impersonate="chrome_143")
with open("session.json") as f:
    client2.import_cookies([tuple(c) for c in json.load(f)])

# 直接访问需要登录的页面
r = client2.get("https://example.com/dashboard")

---

浏览器指纹伪装

支持的浏览器（100+ 配置）

浏览器	版本范围	别名
Chrome	100–145	"chrome" → 最新
Edge	101–145	"edge" → 最新
Firefox	109–147	"firefox" → 最新
Safari macOS	15.3–26.2	"safari" → 最新
Safari iOS	16.5–26.2	"safari_ios" → 最新
Safari iPad	18–26.2	"safari_ipad" → 最新
Opera	116–119	"opera" → 最新
OkHttp	3.9–5	"okhttp" → 最新

伪装内容

client = never_primp.Client(
    impersonate="chrome_143",
    impersonate_os="windows",   # windows / macos / linux / android / ios
)

每个配置包含：

• TLS 指纹：cipher suites 顺序、TLS extensions、椭圆曲线、签名算法（JA3/JA4）
• HTTP/2 指纹：SETTINGS 帧参数、WINDOW_UPDATE、HEADERS 帧顺序（AKAMAI 指纹）
• 请求头集合：与该浏览器版本完全一致的默认头部
• 请求头顺序：精确匹配浏览器的头部发送顺序

Cookie 分割（HTTP/2 浏览器行为）

# HTTP/2 中浏览器将每个 Cookie 作为独立的 Header Frame 字段发送
client = never_primp.Client(
    impersonate="chrome_143",
    split_cookies=True,     # 默认 True（Python 层）
)

# 发出的 HTTP/2 HEADERS 帧：
# cookie: session=abc
# cookie: user_id=123
# cookie: csrf_token=xyz

---

API 参考

Client 构造参数

client = never_primp.Client(
    # 认证
    auth=("username", "password"),  # Basic Auth
    auth_bearer="token",            # Bearer Auth

    # 网络
    proxy="socks5://127.0.0.1:1080",
    timeout=30.0,                   # 总超时（秒）
    verify=True,                    # SSL 证书验证
    ca_cert_file="/path/to/ca.pem", # 自定义 CA 证书

    # 浏览器伪装
    impersonate="chrome_143",
    impersonate_os="windows",       # windows/macos/linux/android/ios

    # HTTP 协议
    http1_only=False,               # 强制 HTTP/1.1
    http2_only=False,               # 强制 HTTP/2
    https_only=False,               # 拒绝 HTTP 请求
    follow_redirects=True,
    max_redirects=20,

    # Cookie
    cookie_store=True,              # 启用持久 Cookie
    split_cookies=True,             # HTTP/2 风格 Cookie 头

    # 重试
    max_retries=2,                  # 网络错误时的最大重试次数

    # 默认值（所有请求共享）
    headers={"X-Custom": "value"},
    params={"version": "2"},
)

请求方法

# 所有方法支持相同的参数集
r = client.get(url,
    params={"q": "python"},
    headers={"Accept": "application/json"},
    cookies={"session": "abc"},
    timeout=10.0,
    proxy="http://127.0.0.1:8080",
    verify=False,
    # 请求级别临时覆盖
    impersonate="firefox_147",
    impersonate_os="linux",
)

r = client.post(url,
    json={"key": "value"},            # JSON 请求体
    # data={"key": "value"},          # Form 表单
    # content=b"raw bytes",           # 原始字节
    # files={"file": "/path/to/file"},# 文件上传（multipart）
)

Response 对象

r = client.get("https://httpbin.org/get")

r.status_code      # int: 200
r.url              # str: 最终 URL（重定向后）
r.headers          # dict[str, str]: 响应头
r.cookies          # dict[str, str]: Set-Cookie 解析结果
r.content          # bytes: 原始响应体
r.text             # str: 自动编码检测后的文本
r.encoding         # str: 检测到的编码
r.json()           # Any: JSON 解析

Cookie 管理方法

# 单个 Cookie
client.get_cookie(name, url)                          # → str | None
client.set_cookie(name, value, url,
                  domain=None, path=None)             # → None
client.remove_cookie(name, url)                       # → None

# 批量操作
client.get_cookies(url)                               # → dict[str, str]
client.set_cookies(url, {"k": "v"})                   # → None
client.clear_cookies()                                # → None

# 全局查询
client.get_all_cookies()    # → list[tuple[name, value, domain, path]]
client.get_cookies_for_domain(domain)  # → dict[str, str]（含子域名匹配）

# 跨会话持久化
client.export_cookies()                                   # → list[tuple[...]]
client.import_cookies([(name, value, domain, path)])      # → None

请求头管理方法

client.headers = {"User-Agent": "bot"}       # 设置（替换全部）
client.headers                               # 读取
client.set_header("X-Custom", "value")       # 设置单个
client.get_header("X-Custom")               # 读取单个 → str | None
client.headers_update({"Accept": "*/*"})     # 合并更新
client.delete_header("X-Custom")             # 删除单个
client.clear_headers()                       # 清空全部

便利函数（无需创建 Client）

import never_primp

r = never_primp.get("https://httpbin.org/get")
r = never_primp.post("https://httpbin.org/post", json={"a": 1})
r = never_primp.put(url, data={"k": "v"})
r = never_primp.delete(url)
r = never_primp.patch(url, json={})
r = never_primp.head(url)
r = never_primp.options(url)

AsyncClient（异步接口）

import asyncio
import never_primp

async def main():
    async with never_primp.AsyncClient(
        impersonate="chrome_143",
        max_retries=2,
    ) as client:
        r = await client.get("https://httpbin.org/get")
        print(r.json())

asyncio.run(main())

---

开发

# 环境依赖
pip install maturin

# 开发构建（快速迭代）
maturin develop

# 发布构建（完整优化）
maturin develop --release

# 代码检查
cargo check
cargo clippy
cargo fmt

# 运行示例
python example/concurrent_requests.py
python example/cookie_management.py
python example/browser_impersonation.py

---

架构概览

Python 调用  client.get(url)
      ↓
Client.__init__.py   # Python 封装层（ergonomics）
      ↓
RClient.request()   # PyO3 Rust 类（src/client.rs）
      ↓
py.detach()         # 释放 Python GIL
      ↓
RUNTIME.block_on()  # 进入 Tokio 异步运行时（4 workers）
      ↓
wreq::Client        # Rust HTTP 客户端（内部 Arc，无锁 clone）
      ↓
wreq 连接池          # TCP 复用 + TLS 会话复用
      ↓
目标服务器
      ↓
Response::from_wreq_response()  # 懒加载转换（src/response.rs）
      ↓
返回 Python，GIL 重新获取

---

License

MIT