说"实时推送"，真正要回答的是两端——消息从产生到进入业务系统，延迟有多少、能不能保序、网络抖一下会不会丢。这篇讲回调配置里影响实时性的几个选项，以及怎么把延迟量出来。
一、配置项里影响实时性的有几个
回调地址要求公网可达、走 HTTPS。除此之外，超时时间和重试次数影响最直接——超时设太短，正常处理也可能被判失败；重试太频繁，网络抖动时会涌来一批重复消息。
配置完成后先做一次连通性验证：手动触发一条测试事件，确认能到达并打印出来，再接真实流量。
二、延迟怎么量：给两端打时间戳
实时性不能靠感觉，得量。做法是在消息产生时记录一个时间戳，回调到达时记录另一个，两个相减就是端到端延迟。
把延迟按分钟统计出来（平均值、P95），才能知道高峰期是不是变慢了。没有这个数，"实时"只是一句感觉。
三、顺序与去重
同一会话的消息要保序——处理时按会话串行，不同会话可以并发，互不影响。
去重还是按事件 ID。网络抖动触发的重试会带来重复事件，去重要放在处理之前，而不是处理之后。
实时性关注点对照
关注点 | 配置/做法 | 目标 | 常见问题 |
|---|---|---|---|
地址 | 公网 HTTPS | 可达 | 内网地址不通 |
超时 | 合理放宽 | 正常处理不被判失败 | 设太短误判 |
重试 | 控制次数 | 抖动不放大 | 重推风暴 |
延迟 | 两端打时间戳 | 可量化 | 只凭感觉 |
顺序 | 按会话串行 | 保序 | 并发乱序 |
延迟度量与保序实现
import time import threading class RealtimeMonitor: def __init__(self): self.samples = [] # 延迟样本（秒） def on_event(self, event): # 端到端延迟：事件产生时间 -> 到达时间 produced = event.get("create_time", 0) / 1000.0 self.samples.append(max(0.0, time.time() - produced)) if len(self.samples) > 1000: # 只留最近一千条 self.samples.pop(0) return event def report(self): if not self.samples: return {} data = sorted(self.samples) n = len(data) return { "count": n, "avg": round(sum(data) / n, 2), "p95": round(data[int(n * 0.95) - 1], 2), "max": round(data[-1], 2), } class OrderedWorker: def __init__(self): self.locks = {} def handle(self, event): # 同一会话串行，保证顺序；不同会话可并发 key = event.get("chat_id") or event.get("from") lock = self.locks.setdefault(key, threading.Lock()) with lock: return self.process(event)落地建议
回调地址必须公网可达并走 HTTPS，配好先做连通性验证。超时别设太短，否则正常处理也会被判失败触发重推。重试次数控制住，防止抖动放大成重推风暴。实时性要拿数据说话，在消息产生和到达两端打时间戳，统计平均和 P95。同一会话按串行处理保序，去重放在处理之前。微信侧的回调推送由Eyun这类个人微信API平台 提供，延迟度量与事件处理在自建服务实现，具体字段以平台开发文档为准。
