回调收不到消息,排查起来最糟心的是"我不知道哪里坏了"——是平台没推、是中间网络断了、还是自己的服务没接住。这篇给一个定位三步法,外加三个最常见的坑。
一、先确认"平台有没有推"
这一步自己没法直接做,要找平台侧有没有回调记录。如果平台显示已经推了但你没收到,问题在你这边;如果平台显示根本没推,先看配置有没有生效、账号有没有在线。
有些平台提供了回调测试接口——手动触发一条测试事件,看能不能到。这是最快的验证方法。
二、中间链路:你的地址能不能被访问
回调地址必须是公网可达的 HTTPS 地址。本地地址不行、内网穿透偶尔断、证书过期都收不到。快速验证:用 curl 从外部服务器拉一下你的回调地址,能通就过了。
有些服务器开了防火墙或白名单,只允许特定 IP 访问回调端口——如果平台 IP 没在白名单里,同样收不到。
三、服务端:验签没过或响应太慢
验签是最容易踩的坑。参数排序错了、算法选错了、密钥写错了,签名就永远对不上。建议先跳过验签(临时注释),确认能收到原始请求体,再加回来。
响应超时是第二个坑。回调必须在规定时间内返回(通常 3-5 秒),超时平台判失败重推,甚至连续超时后停止推送。回调里只做验签、去重、入队,业务全放异步。
三步定位对照
步骤 | 检查什么 | 验证方式 | 常见结果 |
|---|---|---|---|
平台侧 | 有没有推 | 查平台回调日志/测试接口 | 推了/没推 |
中间链路 | 地址可达 | curl 外部访问/防火墙 | 通/不通 |
服务端 | 验签+响应时间 | 临时跳验签/看日志 | 验签错/超时 |
三个最常见的坑
验签参数排序——很多平台要求按 ASCII 码排序拼接,不是按字段定义顺序,排错签名永远对不上。
响应超时——回调里调数据库或调模型,很容易超 5 秒。正确做法是收到就 ACK,业务放异步。
IP 白名单——服务器只允许特定 IP 访问,平台 IP 没在白名单里收不到,这种情况看起来像是"平台没推"。
快速排查骨架
import hashlib, time from flask import Flask, request app = Flask(__name__) SECRET = "your_secret" processed = set() @app.route("/hook", methods=["POST"]) def hook(): start = time.time() raw = request.get_data(as_text=True) # 1. 先打印原始请求体,确认能收到 print("[原始请求]", raw, "耗时:", round(time.time() - start, 3), "s") data = request.get_json(force=True) or {} # 2. 验签:临时注释掉先确认收得到,再加回来 # if not verify(data): # print("[验签失败]", data) # return {"code": -1}, 403 mid = data.get("msg_id") if mid in processed: print("[重复] 已处理过", mid) return {"code": 0} processed.add(mid) # 3. 立即返回,不做业务 # 业务放异步队列里处理 # asyncio.create_task(process(data)) # 异步任务 print("[接收成功]", mid, "总耗时:", round(time.time() - start, 3), "s") return {"code": 0} def verify(data): sign = data.get("signature", "") payload = "".join(f"{k}{v}" for k, v in sorted(data.items()) if k != "signature") payload += SECRET expect = hashlib.md5(payload.encode()).hexdigest() return sign == expect落地建议
回调收不到按三步查:先看平台有没有推(查日志/用测试接口)、再测地址公网可达(curl 外部访问+防火墙)、最后验签和响应时间。验签跳过测试是最快定位方法——注释掉先确认能收到,再加回来。回调里绝对不做耗时业务,只做验签、去重、入队,必须在时限内返回。微信侧的回调推送由Eyun这类个人微信API平台 提供,排查和接收逻辑在自建服务实现,具体字段以平台开发文档为准。