异步爬虫生产线：httpx + Playwright 构建高并发抖音数据采集管线

为什么同步爬虫不够用？

用 requests 写一个简单的搜索接口调用，代码大概长这样：

import requests

def search(keyword):
    for page in range(10):
        resp = requests.get(f"https://api.example.com/search?q={keyword}&page={page}")
        data = resp.json()
        process(data)
        time.sleep(2)

可以工作，但有两个致命问题：

串行阻塞：10页数据，每页等2秒，最少20秒。加上网络延迟，轻松破30秒
资源浪费：等待响应的时间，CPU 完全闲置

如果只是搜一个关键词，30秒还能忍。但要搜5个品类、每个品类的每个达人都要扫10条作品——那就是几百次请求，串行跑下来没有十分钟根本完不成。

异步改造：从 requests 到 httpx

基础并发模型

httpx 提供了和 requests 几乎一致的 API，但底层支持 asyncio：

import httpx
import asyncio

async def search_page(client: httpx.AsyncClient, keyword: str, page: int):
    resp = await client.get(
        "https://api.example.com/search",
        params={"q": keyword, "page": page}
    )
    return resp.json()

async def search_all(keyword: str, total_pages: int = 10):
    async with httpx.AsyncClient(timeout=30) as client:
        tasks = [search_page(client, keyword, p) for p in range(total_pages)]
        results = await asyncio.gather(*tasks)
        return results

10个请求同时发出，总耗时约等于最慢那一个请求的时间。从30秒压缩到3秒以内。

连接池复用

httpx.AsyncClient 内部维护了一个连接池。相比于每次请求都新建 TCP 连接，连接池复用的优势非常明显：

# 反模式：每次创建新 client
async def bad_search():
    for url in urls:
        async with httpx.AsyncClient() as client:
            resp = await client.get(url)

# 推荐：复用同一个 client
async def good_search(urls):
    async with httpx.AsyncClient(
        limits=httpx.Limits(max_connections=20, max_keepalive_connections=10)
    ) as client:
        tasks = [client.get(url) for url in urls]
        return await asyncio.gather(*tasks)

并发控制：信号量 + 速率限制

并发不是越高越好。抖音的限流机制对高频请求非常敏感——HTTP 444（无响应）或直接返回空结果。

信号量控制并发上限

class RateLimitedCrawler:
    def __init__(self, max_concurrency: int = 4):
        self.semaphore = asyncio.Semaphore(max_concurrency)
        self._last_request = 0.0
        self._min_interval = 1.0

    async def fetch(self, client: httpx.AsyncClient, url: str):
        async with self.semaphore:
            elapsed = time.monotonic() - self._last_request
            if elapsed < self._min_interval:
                await asyncio.sleep(self._min_interval - elapsed)
            resp = await client.get(url)
            self._last_request = time.monotonic()
            return resp

三档速度模式

将速率策略抽象为配置，而非硬编码：

SPEED_PROFILES = {
    "safe": {
        "search_interval": (3.0, 5.0),
        "author_interval": (3.0, 8.0),
        "post_interval":   (1.0, 3.0),
        "max_concurrency": 4,
    },
    "fast": {
        "search_interval": (1.5, 3.0),
        "author_interval": (2.0, 4.0),
        "post_interval":   (0.8, 1.5),
        "max_concurrency": 6,
    },
    "turbo": {
        "search_interval": (0.8, 1.5),
        "author_interval": (1.0, 2.5),
        "post_interval":   (0.3, 0.8),
        "max_concurrency": 8,
    },
}

def get_interval(profile: str, stage: str) -> float:
    low, high = SPEED_PROFILES[profile][f"{stage}_interval"]
    return random.uniform(low, high)

关键设计：间隔用随机范围而非固定值。固定间隔会被反爬系统识别为机器行为；随机抖动模拟了人类操作的不规律性。

签名破解：a_bogus 算法

抖音 Web 端的搜索接口会对请求参数进行签名，a_bogus 是其中最关键的一个字段。签名使用 SM3（国密哈希）对参数串进行运算：

from gmssl import sm3

def generate_a_bogus(params: dict, user_agent: str) -> str:
    param_str = "&".join(
        f"{k}={v}" for k, v in sorted(params.items())
    )
    payload = f"{param_str}{user_agent}{SALT_CONSTANT}"
    hash_bytes = sm3.sm3_hash(payload.encode())
    return extract_signature(hash_bytes)

注意：完整的签名逻辑涉及更多细节和逆向分析。上述代码为简化示意，实际实现请参考项目源码。

Cookie 是采集管线的”钥匙”。没有有效的登录态，搜索接口直接返回空。设计了四种获取策略，互为备份：

策略	实现	优点	缺点
Playwright 登录提取	启动 Chromium → 用户扫码 → 自动抓取 Cookie	最可靠，傻瓜式操作	需要 Playwright 环境
rookiepy	直接读取本地浏览器数据库	无需另开浏览器	需要管理员权限
CDP 调试端口	连接已打开的 Chrome DevTools	不影响浏览器状态	需要事先开启调试端口
剪贴板 cURL 解析	解析 F12 → Copy as cURL 的内容	最精确，完全手动控制	操作步骤稍多

async def extract_cookie(strategy: str = "playwright") -> str:
    strategies = {
        "playwright": extract_via_playwright,
        "rookiepy":   extract_via_rookiepy,
        "cdp":        extract_via_cdp,
        "clipboard":  extract_via_clipboard,
    }
    extractor = strategies.get(strategy)
    if not extractor:
        raise ValueError(f"Unknown strategy: {strategy}")
    cookie = await extractor()
    if not any(k in cookie for k in ("sessionid", "sid_guard")):
        raise CookieInvalidError("Cookie missing required fields")
    return cookie

容错与重试

生产环境的网络请求永远不可能 100% 成功。一个健壮的采集管线需要处理：

async def fetch_with_retry(
    client: httpx.AsyncClient,
    url: str,
    max_retries: int = 3,
    backoff_base: float = 2.0
) -> httpx.Response:
    for attempt in range(max_retries):
        try:
            resp = await client.get(url)
            if resp.status_code == 444:
                wait = backoff_base ** attempt
                logger.warning(f"Rate limited, waiting {wait}s...")
                await asyncio.sleep(wait)
                continue
            resp.raise_for_status()
            return resp
        except (httpx.TimeoutException, httpx.HTTPStatusError) as e:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(backoff_base ** attempt)
    raise RateLimitError(f"still rate limited after {max_retries} retries")

效果对比

指标	同步版 (requests)	异步版 (httpx)
搜索 200 条视频	~90s	~8s
扫描 50 位达人 (每人10条)	~300s	~45s (并发4)
总流程耗时	~6.5 min	~1 min
资源利用	单线程阻塞	异步并发 + 连接池复用

从 6.5 分钟到 1 分钟，异步改造带来了近 7 倍的效率提升。而这还没有启用极速模式——极速模式下总流程可以压到 30 秒以内。

对我自己的启发

这个项目让我重新审视了”速度”这件事。最初觉得爬虫就是发请求、等响应，没什么技术含量。但当真要把时间从几分钟压缩到几十秒时，才发现延迟不在网络上，而在设计上——同步还是异步、串行还是并发、固定间隔还是随机抖动，每一个决策都直接影响最终效率。

更深一层的体会是：异步编程的价值不在于让代码”看起来更高级”，而是在有限的资源下（同一台电脑、同一个 IP、同一个账号），系统性榨出每一次网络请求之间的闲置时间。这个原则同样适用于后端服务、消息队列、Agent 调度——凡是涉及等待的场景，都值得用异步重构。

项目的下一步方向：请求失败的自动降级（极速触发限流后自动切安全模式）、分布式多账号轮换、将采集结果写入 SQLite 支持增量更新。

丽·芙