做过一个电商客服系统,上线第一周就出事:客户发一条消息,机器人回了三条一样的。客户截图发朋友圈吐槽"这商家机器人是不是傻"。

排查发现是 Webhook 回调没做幂等。平台重推了两次(网络波动),系统每次都当新消息处理,于是回复了三遍。这事儿让我明白:Webhook 设计不是接个口那么简单,幂等、重试、安全一个都不能少。

这篇把 Webhook 设计的三个关键点讲清楚,全是血泪经验。

一、Webhook 在实际业务里到底多重要

先说说为什么 Webhook 这么关键。我做的几个项目,都离不开它:

1. 电商客服机器人

客户发消息问"我的订单到哪了",机器人要能收到这条消息才能回复。Webhook 就是接收客户消息的通道,挂了机器人就聋了。

2. 社群关键词监控

运营同学要在 50 个群里监控关键词(比如"差评""投诉"),一旦出现立刻人工介入。Webhook 把群消息实时推过来,系统才能及时响应。

3. 订单状态变更通知

客户在微信里发"已付款",Webhook 推给订单系统,系统自动更新订单状态。如果这条消息丢了,订单就一直停在"待付款",客户以为没成功又付一次,那就是事故。

4. 自动加好友

有人加你微信,Webhook 推送好友请求事件,系统自动通过并发送欢迎语。私域加粉引流全靠这个。

这些场景的共同点:消息不能丢,也不能重复处理。这就是 Webhook 设计要解决的核心问题。

配置回调的接口见 设置 HTTP 回调地址,事件类型见 Webhook 事件索引。

二、Webhook 的工作机制

先理清机制。Webhook 是平台主动把消息推给你的服务端,流程是:

1. 微信收到新消息

2. 平台 POST 到你配置的回调 URL

3. 你的服务处理消息,返回 code = "1000"

4. 如果平台收不到成功响应(超时、5xx),会重推

关键点:平台只保证"至少推送一次",不保证"只推一次"。所以幂等必须自己做。

三、幂等设计(最重要的一个)

1. 用消息 ID 去重

每条消息都有唯一的 msgId,用 Redis 做去重:

import redis

r = redis.Redis(host="localhost", decode_responses=True)

def handle_webhook(data):
    msg_id = data.get("msgId")
    if not msg_id:
        return {"code": "1000"}
    
    # setnx 原子操作,设置成功说明是第一次收到
    key = f"wechat:msg:{msg_id}"
    if not r.set(key, "1", nx=True, ex=600):  # 10分钟过期
        print(f"重复消息,丢弃: {msg_id}")
        return {"code": "1000"}
    
    process_message(data)
    return {"code": "1000"}

2. 过期时间设置

建议 5-10 分钟。太短可能平台延迟重推时还没处理完,太长浪费内存。

3. 幂等的边界

幂等只保证"同一条消息不处理两次",不保证"业务逻辑幂等"。比如收到消息后要给对方回一条,即使消息去重了,回复逻辑也要考虑重复触发。

四、快速响应(别让回调卡住)

平台对回调响应有时间要求(通常 5 秒),超时会认为你挂了,重推一次。所以业务处理要异步:

from flask import Flask, request, jsonify
from celery import Celery

app = Flask(__name__)
celery = Celery("tasks", broker="redis://localhost:6379")

@app.route("/webhook", methods=["POST"])
def webhook():
    data = request.json
    # 立即丢到异步队列,快速返回
    process_message.delay(data)
    return jsonify({"code": "1000", "message": "success"})

@celery.task
def process_message(data):
    # 耗时处理放这里
    msg_id = data.get("msgId")
    # 去重检查 + 业务逻辑

原则:回调入口只做接收和入队,业务处理全部异步。这样即使业务逻辑卡住,也不影响回调接收。

我之前栽过这个坑:客服系统收到消息后查客户信息、查历史记录、生成回复,整个流程 8 秒。平台等不及,重推了 3 次。改异步后正常。

五、安全防护(别让人钻空子)

Webhook 是公网可访问的,别人知道你的 URL 就能伪造请求。

1. 来源校验

import hashlib
import hmac

def verify_signature(payload, signature, secret):
    """校验平台签名"""
    expected = hmac.new(
        secret.encode(), payload.encode(), hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)

@app.route("/webhook", methods=["POST"])
def webhook():
    signature = request.headers.get("X-Signature", "")
    if not verify_signature(request.get_data(as_text=True), signature, WEBHOOK_SECRET):
        return jsonify({"code": "1001", "message": "invalid signature"}), 403
    # 正常处理

2. IP 白名单

如果平台提供出口 IP 列表,在 Nginx 或应用层做 IP 白名单,比签名校验更简单。

3. HTTPS 强制

回调地址必须用 HTTPS,明文 HTTP 传输消息内容有泄露风险。

六、失败补偿(兜底机制)

即使做了上述所有措施,还是可能丢消息。需要补偿:

1. 对账机制

定时任务每小时跑一次,拉取平台消息列表跟本地比对,补处理缺失的消息。

2. 死信队列

异步处理失败的消息进死信队列,人工介入或定时重试:

@celery.task(bind=True, max_retries=3)
def process_message(self, data):
    try:
        do_something(data)
    except Exception as e:
        # 重试 3 次还失败,进死信队列
        raise self.retry(exc=e, countdown=60)

七、实战:电商客服机器人 Webhook 完整示例

把上面的东西整合起来,一个可用的客服机器人 Webhook:

from flask import Flask, request, jsonify
import redis
import hmac
import hashlib

app = Flask(__name__)
r = redis.Redis(decode_responses=True)
WEBHOOK_SECRET = "your_secret"

@app.route("/webhook", methods=["POST"])
def webhook():
    # 1. 签名校验
    signature = request.headers.get("X-Signature", "")
    if not verify_signature(request.get_data(as_text=True), signature, WEBHOOK_SECRET):
        return jsonify({"code": "1001"}), 403
    
    data = request.json
    msg_id = data.get("msgId")
    
    # 2. 幂等去重
    if msg_id and not r.set(f"wechat:msg:{msg_id}", "1", nx=True, ex=600):
        return jsonify({"code": "1000"})  # 重复消息,直接返回成功
    
    # 3. 异步处理,快速返回
    handle_customer_message.delay(data)
    return jsonify({"code": "1000", "message": "success"})

def verify_signature(payload, signature, secret):
    expected = hmac.new(secret.encode(), payload.encode(), hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature.strip())

@celery.task(bind=True, max_retries=3)
def handle_customer_message(self, data):
    try:
        from_user = data.get("fromUser")
        content = data.get("content", "")
        
        # 关键词匹配
        if "退款" in content:
            client.send_text(w_id, from_user, "退款申请已收到,客服将处理")
        elif "发货" in content:
            client.send_text(w_id, from_user, "请提供订单号,我帮您查询")
        else:
            # 转人工
            client.send_text(w_id, from_user, "人工客服稍后回复")
    except Exception as e:
        raise self.retry(exc=e, countdown=60)

八、踩坑记录

坑1:幂等 key 用错

一开始用 fromUser + content 做幂等 key,结果客户连续发了两条一样的消息,第二条被丢了。后来改成 msgId,平台保证 msgId 唯一。

坑2:回调处理太慢

前面说过,8 秒处理导致重推 3 次。异步后正常。

坑3:签名校验漏了空格

签名用 hmac.compare_digest 时,字符串前后的空格会导致校验失败。务必用 strip() 处理。

坑4:HTTPS 证书过期

回调地址用 HTTPS,证书过期了没注意,平台全部推送失败,客服系统聋了一整天。后来加了证书到期监控。

九、小结

Webhook 设计的核心是三个字:快、稳、安全。快速响应(异步处理)、稳定投递(幂等 + 补偿)、安全防护(签名 + 白名单)。这三点做到位,回调链路基本不会出大问题。

更多细节参考官方的 Webhook 可靠性 文档。

• Eyun平台