说实时推送真正要回答的是两端——消息从产生到进入业务系统延迟有多少、能不能保序、网络抖一下会不会丢。这篇讲回调配置里影响实时性的几个选项以及怎么把延迟量出来。一、配置项里影响实时性的有几个回调地址要求公网可达、走 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平台 提供延迟度量与事件处理在自建服务实现具体字段以平台开发文档为准。 
长河君 · 2026/10/20 15:22:35 · 阅读 1.2k 个人微信API接口如何实现消息实时推送？回调配置与事件处理实战
说实时推送真正要回答的是两端——消息从产生到进入业务系统延迟有多少、能不能保序、网络抖一下会不会丢。这篇讲回调配置里影响实时性的几个选项以及怎么把延迟量出来。一、配置项里影响实时性的有几个回调地址要求公网可达、走 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平台 提供延迟度量与事件处理在自建服务实现具体字段以平台开发文档为准。