Webhooks 详解：签名负载与重试机制

Backend2026-10-24TryQuickToolBox

你已经构建了一个 API，现在需要在事件发生时通知其他服务。你设置了一个 webhook 端点，但很快意识到可靠地传递 HTTP 请求比看起来要困难得多。网络故障、恶意行为者和重复事件都可能造成混乱。本指南解释了如何通过签名负载和重试机制来实现 webhooks，从而构建一个健壮的集成。

什么是 Webhooks？

Webhooks 是用户定义的 HTTP 回调。当你的应用程序中发生事件时，你会向订阅者提供的 URL 发送一个 HTTP POST 请求。这允许服务之间进行实时通信，而无需轮询。

例如，支付处理器可能会在扣款成功时向你的应用发送一个 webhook。然后你的应用可以立即更新订单状态。

为什么要对 Webhook 负载进行签名？

Webhook 端点是公开可访问的，因此任何人都可以发送虚假请求。如果没有验证，你的应用可能会处理欺诈性事件。对负载进行签名可以确保请求源自你的服务，并且未被篡改。

签名通常使用发送方和接收方之间共享的密钥。发送方基于负载计算一个基于哈希的消息认证码（HMAC），并将其包含在头部中。接收方重新计算 HMAC 并进行比较。

如何生成和验证签名

大多数实现使用 HMAC-SHA256。以下是一个 Python 中的简单示例：

import hmac
import hashlib
import json

def generate_signature(payload, secret):
    body = json.dumps(payload, separators=(',', ':')).encode('utf-8')
    signature = hmac.new(secret.encode('utf-8'), body, hashlib.sha256).hexdigest()
    return signature

def verify_signature(payload, signature, secret):
    expected = generate_signature(payload, secret)
    return hmac.compare_digest(expected, signature)

发送 webhook 时，将签名包含在类似 X-Signature 的头部中。接收方提取原始请求体，重新计算 HMAC，并进行比较。始终使用恒定时间比较函数以防止时序攻击。

实现重试以确保可靠交付

Webhooks 通过互联网发送，因此失败是不可避免的。你的服务器可能宕机、网络可能缓慢，或者订阅者可能返回错误。带指数退避的重试至关重要。

重试策略

一种常见的方法是使用递增的延迟进行重试：1 分钟、5 分钟、30 分钟、2 小时等。在达到最大尝试次数（例如 10 次）或总时间窗口（例如 24 小时）后停止。

仅在特定条件下重试：网络错误、5xx 响应或 429（请求过多）。不要对 4xx 错误（429 除外）进行重试，因为它们表示客户端错误，重试无法修复。

重试逻辑示例

以下是使用 requests 库的 Python 中的简化重试机制：

import requests
import time
from requests.exceptions import RequestException

def send_webhook(url, payload, secret, max_retries=5):
    headers = {'Content-Type': 'application/json'}
    signature = generate_signature(payload, secret)
    headers['X-Signature'] = signature
    
    for attempt in range(max_retries):
        try:
            response = requests.post(url, json=payload, headers=headers, timeout=10)
            if response.status_code < 500 and response.status_code != 429:
                return response.status_code
        except RequestException:
            pass
        
        # 带抖动的指数退避
        delay = (2 ** attempt) + (random.random() * 0.1)
        time.sleep(delay)
    
    return None  # 重试后失败

在生产环境中，你会将失败的尝试存储在队列中并异步处理。Celery 或云队列（AWS SQS、Google Cloud Tasks）等工具可以提供帮助。

Webhook 安全最佳实践

使用 HTTPS：始终通过 TLS 发送 webhooks 以防止窃听。
包含时间戳：在负载中添加时间戳，并拒绝超过几分钟的请求以防止重放攻击。
轮换密钥：定期更改签名密钥，并在轮换期间支持多个活动密钥。
验证 URL：注册 webhook 时，确保 URL 可访问并对测试请求返回 2xx。
速率限制：通过限制每个订阅者的请求数量来保护你的 webhook 发送方不被滥用。

处理重复和幂等性

重试可能导致重复交付。在负载中包含唯一的事件 ID，并让接收方跟踪已处理的 ID。如果之前见过某个事件 ID，则忽略它。

或者，将事件处理程序设计为幂等的。例如，多次使用相同数据更新记录应与执行一次具有相同的效果。

测试 Webhooks

使用 ngrok 等工具暴露你的本地服务器并测试 webhook 交付。通过返回错误代码模拟失败，并验证重试按预期发生。

常见问题

什么是签名的 webhook 负载？

签名负载包含一个加密签名（通常是 HMAC），允许接收方验证请求来自可信来源且未被篡改。

我应该重试失败的 webhook 多少次？

这取决于你的可靠性需求。常见的做法是在 24 小时内进行 5-10 次带指数退避的重试。始终允许订阅者手动触发重试。

我可以不使用签名来使用 webhooks 吗？

可以，但风险很高。没有签名，任何人都可以发送虚假事件。在生产环境中始终对 webhooks 进行签名以防止欺诈。