Rnote API Rnote API
返回博客

小红书 API 大规模采集最佳实践：限流、重试、分页与去重

Rnote API 团队 · · 1264 次阅读 · English
小红书数据 最佳实践 限流分页

把小红书（RedNote / Xiaohongshu）数据采集从"能跑"做到"稳定规模化"，关键在几个工程细节：限流、重试、分页、去重和成本控制。这篇总结一套可以直接照搬的最佳实践，配 Rnote API 使用。

1. 限流与并发

给请求设并发上限与节流，别一上来就打满。遇到 429（限流）时退避稍等再试，平稳的请求曲线比突发洪峰更稳定、更省心。

2. 重试与退避

只对网络错误与 5xx 做重试，用指数退避（1s、2s、4s…）。好消息是 Rnote API 仅成功请求（HTTP 2xx）扣费，失败重试不额外计费，所以你可以放心加重试逻辑。

3. 分页：不同端点游标不同

端点 翻页方式
search/notes page 递增
user/posted 用上一页返回的 cursor
note/comments 游标对象 cursor + index
topic/feed cursor_score + last_note_id + last_note_ct

翻页时务必把上一次响应里的游标字段原样带入下一次请求，保证连续性。

4. 去重

跨页、跨排序、跨任务都可能拿到重复数据。统一按 note_id / user_id 去重（用集合或数据库唯一索引），既省额度也保证数据干净。

5. 成本与额度控制

  • 仅成功请求扣费，失败不花钱。
  • 先用小批量验证采集逻辑与字段，再规模化放量。
  • 给任务设预算上限，避免脚本异常时跑飞。

一个稳健的分页采集骨架（Python）

import time, requests

API = "https://rnote.dev/api/v2/crawler"
H = {"X-API-Key": "YOUR_API_KEY"}

def get(path, params, retries=3):
    for i in range(retries):
        r = requests.get(f"{API}/{path}", params=params, headers=H, timeout=30)
        if r.status_code == 200:
            return r.json()
        if r.status_code == 429:
            time.sleep(2 ** i)          # 限流退避
        elif r.status_code >= 500:
            time.sleep(2 ** i)          # 服务端错误退避
        else:
            r.raise_for_status()        # 4xx 不重试
    raise RuntimeError("retries exhausted")

def search_all(keyword, max_pages=10):
    seen = set()
    for page in range(1, max_pages + 1):
        data = get("search/notes", {"keyword": keyword, "page": page})
        items = extract_items(data)     # 按响应结构取列表
        if not items:
            break
        for it in items:
            nid = it["note_id"]
            if nid not in seen:         # 去重
                seen.add(nid)
                yield it

6. 错误分类：哪些该重试，哪些不该

上面讲了怎么退避，但退避的前提是这个错误值得重试。分错类的代价不对称：该重试的不重试只是丢数据，不该重试的一直重试会把额度和日志一起烧掉。

类型 状态码 处理
可重试 429、5xx、连接超时 指数退避，重试上限 3~5 次
不可重试 400、404、422 记进失败清单，跳过
必须停机 401、402 立刻停整个任务并告警

第三类最容易写错。把 402（余额不足）当普通失败重试是最常见的一种——余额不会因为重试而变多，结果是几万条无效请求刷满日志，真正的问题反而被埋掉。这两个状态码应该 raise，不是 continue。

7. 幂等与断点续跑

批量任务一定会中断：网络、部署、手滑 Ctrl+C。让它可以从中断处接着跑，只需要一条原则——把「已完成」落盘，而不是留在内存里：

import pathlib

state = pathlib.Path("done.txt")
done = set(state.read_text().split()) if state.exists() else set()

with state.open("a") as f:
    for note_id in all_ids:
        if note_id in done:      # 已完成，跳过，不再花钱
            continue
        if not process(note_id):  # 失败的留给第二轮，不写入 done
            continue
        f.write(note_id + "\n")
        f.flush()                 # 关键：不 flush，被 kill 时缓冲区里的几百条一起丢

那个 flush() 是最常被省掉、也最常在需要它的时候失效的一行。

配套的做法是两轮制：第一轮全量跑、失败只记不重试；第二轮只跑失败清单。第一轮不会被少数难缠的条目拖住，第二轮往往因为限速窗口已过、上游恢复，成功率反而很高。两轮之后还失败的，多半是真的取不到（笔记删了、账号注销），该进"确认不可得"清单，而不是留在队列里无限重试。

8. 给不同用途分开 Key

一把 Key 跑所有环境，是团队接入后最贵的习惯：测试脚本一次死循环，把生产的额度和限速一起吃光。限速是按 Key 算的、互不影响，所以按环境拆开本身就是一层保险丝：

  • prod — 生产服务，限速给足
  • staging — 预发布，一半
  • dev-<姓名> / batch — 开发与批量任务，最低

出事那天才见效果：某把 Key 泄漏或被误用，停用它就行，其余环境照常跑。详见《API Key 怎么管》。

开始使用

把这套实践套进你的采集任务，几乎不用改就能稳定跑量。免费注册获取 API Key，配合 Python / Go 教程上手，详见接口文档与定价。

返回博客