调微信 API 调得猛了,两种情况会让你半夜起来救火:接口返回频率限制,或者账号被风控临时冻住。两种都是"触发了阈值",但处理思路不一样——限流是接口层面的硬限制,防封是账号层面的行为合理性。这篇把两层放在一起讲。
一、限流:接口层面的硬限制
接口限流是平台给的明确规则——比如每秒最多 5 次、每小时最多 200 次,超出直接返回错误码。处理方式就是令牌桶加队列:令牌按固定速度生成,请求先拿令牌再调接口,拿不到就排队等。
批量场景要给不同类型的请求不同速率——查询类可以快一点,发送类慢一点,加好友类最慢。统一出口统一控速,别让各业务各自为政。
二、防封:账号层面的软边界
防封不是"别调太猛"这么简单。微信对账号有行为画像——正常用户不会在 1 分钟内加 30 个好友、不会每 5 秒发一条完全相同的消息、不会凌晨 3 点还在批量操作。防封的核心是"让行为看起来像个人"。
几个简单有效的做法:随机化(两次操作之间随机等 3-10 秒)、打散(批量操作分成小批次在不同时间段做)、降峰(高峰时段降低操作频率)。这些在项目记忆里的硬约束有提到——发送间隔最低 3 秒、批量 ≥0.3 秒、文本 ≤500 字。
三、两层结合:限流桶加行为画像
限流桶在接口层做刚性控制,行为画像在业务层做柔性控制。两层都做才稳——光限流但行为模式可疑照样会被封,光画像但速率超高接口照样拒。
实际做法:给每个账号一个行为画像,记录最近操作类型、间隔、批量规模、时间段,每次操作前先过画像检查,过了再去拿令牌调接口。
四、限流触发后的正确动作
别重试。限流返回的错误码是明确告诉你"这段时间别来了",重试只会让等待时间更长。正确做法是按限流返回的 retry_after 字段等,或者指数退避。
防封与限流对照
层次 | 控制对象 | 核心手段 | 触发后动作 |
|---|---|---|---|
接口限流 | 请求速率 | 令牌桶+队列 | 按 retry_after 等 |
行为防封 | 账号画像 | 随机化+打散+降峰 | 主动降频 |
账号降级 | 账号状态 | 触发阈值自动下线 | 人工介入 |
双层控制实现
import time, random, threading from collections import deque class ThrottleBucket: def __init__(self, rate_per_sec): self.capacity = rate_per_sec * 2 self.tokens = self.capacity self.last = time.time() self.lock = threading.Lock() def take(self, n=1): with self.lock: now = time.time() elapsed = now - self.last self.tokens = min(self.capacity, self.tokens + elapsed * self.rate_per_sec) self.last = now if self.tokens >= n: self.tokens -= n return True return False class BehaviorGuard: # 行为画像:按账号记录近 N 次操作 def __init__(self, account, window=60, max_per_min=30): self.account = account self.window = window self.max_per_min = max_per_min self.ops = deque() # (timestamp, op_type) def before_op(self, op_type): now = time.time() # 1. 滑动窗口内同类操作不超阈值 self.ops = deque((t, o) for t, o in self.ops if now - t < self.window) same = sum(1 for _, o in self.ops if o == op_type) if same >= self.max_per_min: return False # 2. 操作间隔随机化:最小 3 秒,再加 0-7 秒随机 gap = 3 + random.uniform(0, 7) if self.ops and now - self.ops[-1][0] < gap: return False self.ops.append((now, op_type)) return True class SafeCaller: def __init__(self, api, bucket, guard): self.api = api self.bucket = bucket self.guard = guard def call(self, op_type, path, data): # 1. 先过行为画像检查 if not self.guard.before_op(op_type): time.sleep(random.uniform(2, 6)) return self.call(op_type, path, data) # 等会再试 # 2. 再拿令牌调接口 while not self.bucket.take(): time.sleep(0.1) return self.api.post(path, data)落地建议
限流和防封是两层不同的事——接口限流是硬规则用令牌桶控速,防封是账号行为画像用随机化和打散让操作看起来像个人。两层都做,光做一层不稳。限流触发后别重试,按 retry_after 等或退避。批量操作必须打散到不同时间段,高峰降频。微信侧的接口限制和风控由Eyun这类个人微信API平台 提供,双层控制逻辑在自建服务实现,具体字段以平台开发文档为准。