构建 Webhook 发送系统：2026 年可靠的回传

上次更新时间：24年2026月XNUMX日，发布者： 凯撒·菲克森

直接回答： 可靠的 webhook 或联盟回传交付服务需要持久队列、有限制的重试次数（带抖动）、每个端点的熔断机制、重复保护、签名验证以及遥测功能（用于在交付失败前显示交付速度是否变慢）。

此实现有意保持执行模型的精简：仅使用 Python 工作进程和 PostgreSQL 来存储队列、传递历史记录和端点健康状况。对于 B2B SaaS、iGaming 联盟营销运营以及任何需要向外部发送事件且不会将任何一次 HTTP 超时视为转化丢失的产品而言，这是一个实用的起点。

交付合同必须保证什么

通过积极争取让商标与其相匹配的域名优先注册来维护	为何重要	运行检查
持久事件记录	事件在工作进程重启后仍然存在。	每个已接收的包裹都有一个稳定的ID和状态。
已签署的请求	接收方可以验证发送方身份并检测机身是否被篡改。	使用带时间戳的签名，并拒绝过期的请求。
重复保护	否则，重试可能会导致重复转换或更新。	发送事件 ID 并使接收者幂等。
有限制的重试次数	瞬态故障可恢复，不会对性能下降的终端造成过大影响。	减少抖动；在达到记录的尝试次数限制后停止。
端点遥测	单凭队列深度无法发现运行缓慢或出现故障的合作伙伴。	跟踪每个端点的成功率、最早待处理事件和延迟百分位数。

安全性和重复控制优先于重试调整。

将出站请求体视为敏感的操作数据。对请求体进行签名，包含发送时间戳和不可更改的事件 ID，轮换签名密钥，并确保接收应用程序能够安全地忽略同一事件的重复发送。成功的 HTTP 响应并不能证明业务事件只被执行了一次；接收方必须自行判断。

对于联盟营销或 iGaming 回传工作流程，除非明确设置了访问控制和保留规则，否则请勿将点击 ID、转化 ID 和与支付相关的字段记录在日志中。下文提供的 Scaleo 示例仅作为联盟营销回传延迟的示例，不能替代您自身服务之间的交付协议文档。

有用的实现参考资料： Stripe webhook 签名验证, PostgreSQL SELECT 和 SKIP LOCKED和 AWS 关于指数退避和抖动的指导.

数据模型

一切都从两个表格开始：一个用于交付队列，一个用于延迟测量。

SQL

-- Pending and in-flight webhook deliveries
CREATE TABLE webhook_queue (
    id              SERIAL PRIMARY KEY,
    endpoint_id     INTEGER NOT NULL,
    endpoint_url    TEXT NOT NULL,
    payload         JSONB NOT NULL,
    
    -- Delivery state
    status          VARCHAR(20) NOT NULL DEFAULT 'pending',
        -- pending, in_flight, delivered, failed, dead_letter
    attempt_count   INTEGER NOT NULL DEFAULT 0,
    max_attempts    INTEGER NOT NULL DEFAULT 5,
    
    -- Timing
    created_at      TIMESTAMP NOT NULL DEFAULT NOW(),
    next_attempt_at TIMESTAMP NOT NULL DEFAULT NOW(),
    delivered_at    TIMESTAMP,
    last_error      TEXT,
    
    -- Response tracking
    last_status_code INTEGER,
    last_response_ms INTEGER,  -- response time in milliseconds
    
    INDEX idx_queue_next (status, next_attempt_at)
        WHERE status IN ('pending', 'failed')
);

-- Per-endpoint latency measurements (ring buffer)
CREATE TABLE endpoint_latency (
    id              SERIAL PRIMARY KEY,
    endpoint_id     INTEGER NOT NULL,
    response_ms     INTEGER NOT NULL,
    status_code     INTEGER,
    measured_at     TIMESTAMP NOT NULL DEFAULT NOW(),
    
    INDEX idx_latency_endpoint (endpoint_id, measured_at)
);

-- Endpoint health state (circuit breaker)
CREATE TABLE endpoint_health (
    endpoint_id         INTEGER PRIMARY KEY,
    endpoint_url        TEXT NOT NULL,
    state               VARCHAR(20) NOT NULL DEFAULT 'closed',
        -- closed (healthy), open (broken), half_open (testing)
    consecutive_failures INTEGER NOT NULL DEFAULT 0,
    failure_threshold   INTEGER NOT NULL DEFAULT 10,
    last_failure_at     TIMESTAMP,
    last_success_at     TIMESTAMP,
    opened_at           TIMESTAMP,  -- when circuit opened
    cooldown_seconds    INTEGER NOT NULL DEFAULT 300,  -- 5 min before half_open
    
    -- Latency stats (updated periodically)
    p50_ms              INTEGER,
    p95_ms              INTEGER,
    p99_ms              INTEGER,
    sample_count        INTEGER NOT NULL DEFAULT 0
);

关于这个模式，有三点需要注意。

首先， webhook_queue 表格使用 next_attempt_at 使用列而不是单独的调度机制。工作进程轮询满足特定条件的行。 status IN ('pending', 'failed') AND next_attempt_at <= NOW()这是一个简易的延迟队列，每分钟处理约 10,000 条消息时运行良好。超过这个数量，就需要换用专业的消息代理。

第二，该 endpoint_latency 表充当环形缓冲区。我会定期清除超过 24 小时的行。延迟百分位数 endpoint_health 这些数据是根据这个滚动窗口计算出来的——它们代表的是近期的行为，而不是历史平均值。

第三，该 endpoint_health 该表实现了断路器状态机。更多详情请见下文。

送货员

核心工作循环的设计刻意保持简洁。复杂性体现在重试逻辑和熔断机制上，而不是交付路径本身。

蟒蛇

import requests
import time
import psycopg2
from psycopg2.extras import RealDictCursor
from datetime import datetime, timedelta

DB_DSN = "postgresql://user:pass@localhost/webhooks"

def get_connection():
    return psycopg2.connect(DB_DSN)

def deliver_webhooks(batch_size=50):
    """
    Fetch pending webhooks and attempt delivery.
    Uses SELECT FOR UPDATE SKIP LOCKED for safe concurrent workers.
    """
    conn = get_connection()
    cur = conn.cursor(cursor_factory=RealDictCursor)
    
    try:
        cur.execute("""
            SELECT id, endpoint_id, endpoint_url, payload, 
                   attempt_count, max_attempts
            FROM webhook_queue
            WHERE status IN ('pending', 'failed')
              AND next_attempt_at <= NOW()
            ORDER BY next_attempt_at ASC
            LIMIT %s
            FOR UPDATE SKIP LOCKED
        """, (batch_size,))
        
        rows = cur.fetchall()
        
        for row in rows:
            # Check circuit breaker before attempting
            if is_circuit_open(cur, row['endpoint_id']):
                # Don't attempt delivery — reschedule for after cooldown
                reschedule_for_cooldown(cur, row['id'], row['endpoint_id'])
                continue
            
            # Attempt delivery and measure latency
            result = attempt_delivery(
                row['endpoint_url'], 
                row['payload']
            )
            
            # Record latency measurement regardless of success/failure
            record_latency(
                cur, 
                row['endpoint_id'], 
                result['response_ms'], 
                result['status_code']
            )
            
            if result['success']:
                mark_delivered(cur, row['id'], result)
                record_success(cur, row['endpoint_id'])
            else:
                handle_failure(
                    cur, row['id'], row['endpoint_id'],
                    row['attempt_count'], row['max_attempts'],
                    result
                )
        
        conn.commit()
    
    except Exception as e:
        conn.rollback()
        raise
    finally:
        cur.close()
        conn.close()


def attempt_delivery(url, payload):
    """
    Fire the webhook and measure response time.
    Returns dict with success, status_code, response_ms, error.
    """
    start = time.monotonic()
    
    try:
        response = requests.post(
            url,
            json=payload,
            timeout=15,          # 15 second hard timeout
            headers={
                'Content-Type': 'application/json',
                'User-Agent': 'WebhookDelivery/1.0',
                'X-Delivery-Timestamp': str(int(time.time()))
            }
        )
        
        elapsed_ms = int((time.monotonic() - start) * 1000)
        
        return {
            'success': 200 <= response.status_code < 300,
            'status_code': response.status_code,
            'response_ms': elapsed_ms,
            'error': None if response.ok else f"HTTP {response.status_code}"
        }
    
    except requests.Timeout:
        elapsed_ms = int((time.monotonic() - start) * 1000)
        return {
            'success': False,
            'status_code': None,
            'response_ms': elapsed_ms,
            'error': 'timeout_15s'
        }
    
    except requests.ConnectionError as e:
        elapsed_ms = int((time.monotonic() - start) * 1000)
        return {
            'success': False,
            'status_code': None,
            'response_ms': elapsed_ms,
            'error': f'connection_error: {str(e)[:200]}'
        }

此 SELECT FOR UPDATE SKIP LOCKED 该子句对于运行多个工作实例至关重要。 SKIP LOCKED两个工作进程会阻塞在同一行上。有了它，每个工作进程都会获取不同的待处理 Webhook 批次。这样，只需启动更多工作进程即可实现横向扩展。

此 time.monotonic() 打电话代替 time.time() 这是故意的。 time.time() 在 NTP 调整期间可能会出现倒退。 time.monotonic() 绝不会倒退，这在测量亚秒级延迟时非常重要。

使用指数退避和抖动的重试逻辑

当交付失败时，重试时间决定了您的系统是能够优雅地恢复，还是会产生一连串的请求，从而冲击正在承受压力的终端。

蟒蛇

import random

def calculate_next_attempt(attempt_count, base_delay=30, max_delay=3600):
    """
    Exponential backoff with full jitter.
    
    Attempt 1: 0-30s
    Attempt 2: 0-60s  
    Attempt 3: 0-120s
    Attempt 4: 0-240s
    Attempt 5: 0-480s (capped at max_delay)
    
    Full jitter prevents thundering herd when an endpoint
    recovers and hundreds of retries fire simultaneously.
    """
    exponential_delay = base_delay * (2 ** attempt_count)
    capped_delay = min(exponential_delay, max_delay)
    jittered_delay = random.uniform(0, capped_delay)
    
    return datetime.utcnow() + timedelta(seconds=jittered_delay)


def handle_failure(cur, webhook_id, endpoint_id, 
                   attempt_count, max_attempts, result):
    """
    Handle a failed delivery attempt.
    Either retry with backoff or move to dead letter queue.
    """
    new_attempt_count = attempt_count + 1
    
    if new_attempt_count >= max_attempts:
        # Exhausted retries — dead letter
        cur.execute("""
            UPDATE webhook_queue 
            SET status = 'dead_letter',
                attempt_count = %s,
                last_error = %s,
                last_status_code = %s,
                last_response_ms = %s
            WHERE id = %s
        """, (
            new_attempt_count, result['error'],
            result['status_code'], result['response_ms'],
            webhook_id
        ))
    else:
        # Schedule retry with backoff
        next_attempt = calculate_next_attempt(new_attempt_count)
        cur.execute("""
            UPDATE webhook_queue
            SET status = 'failed',
                attempt_count = %s,
                next_attempt_at = %s,
                last_error = %s,
                last_status_code = %s,
                last_response_ms = %s
            WHERE id = %s
        """, (
            new_attempt_count, next_attempt,
            result['error'], result['status_code'],
            result['response_ms'], webhook_id
        ))
    
    # Update circuit breaker
    record_failure(cur, endpoint_id)

为什么选择全抖动而不是去相关抖动或等抖动？AWS 发布了权威分析。全抖动（在 0 到指数上限之间随机化）能使所有客户端的总完成时间最短。等抖动（在上限的一半到上限之间随机化）更为保守，但清除重试积压的速度较慢。对于拥有多个独立端点的 Webhook 交付，全抖动是最佳选择，因为每个端点的重试都是独立的——无需进行协调。

断路器：停止敲击断裂的端点

熔断机制可以防止系统将资源浪费在持续故障的端点上。如果没有它，一个失效的端点会累积数百个待处理的重试请求，每个请求都会在 15 秒后超时——这会消耗你的工作资源，导致永远无法成功的请求被提交。

蟒蛇

def is_circuit_open(cur, endpoint_id):
    """
    Check if the circuit breaker is open (endpoint is broken).
    If open and cooldown has passed, transition to half_open.
    """
    cur.execute("""
        SELECT state, opened_at, cooldown_seconds
        FROM endpoint_health
        WHERE endpoint_id = %s
    """, (endpoint_id,))
    
    row = cur.fetchone()
    if not row:
        return False  # no health record = assume healthy
    
    if row['state'] == 'closed':
        return False
    
    if row['state'] == 'open':
        # Check if cooldown period has elapsed
        if row['opened_at'] and row['cooldown_seconds']:
            elapsed = (datetime.utcnow() - row['opened_at']).total_seconds()
            if elapsed >= row['cooldown_seconds']:
                # Transition to half_open — allow one probe
                cur.execute("""
                    UPDATE endpoint_health
                    SET state = 'half_open'
                    WHERE endpoint_id = %s
                """, (endpoint_id,))
                return False  # allow the probe delivery
        return True  # still in cooldown
    
    if row['state'] == 'half_open':
        return False  # allow probe delivery
    
    return False


def record_failure(cur, endpoint_id):
    """
    Record a delivery failure. Open circuit if threshold reached.
    """
    cur.execute("""
        UPDATE endpoint_health
        SET consecutive_failures = consecutive_failures + 1,
            last_failure_at = NOW()
        WHERE endpoint_id = %s
        RETURNING consecutive_failures, failure_threshold, state
    """, (endpoint_id,))
    
    row = cur.fetchone()
    if not row:
        # Create health record on first failure
        cur.execute("""
            INSERT INTO endpoint_health (endpoint_id, endpoint_url, 
                consecutive_failures, last_failure_at)
            VALUES (%s, '', 1, NOW())
            ON CONFLICT (endpoint_id) DO UPDATE
            SET consecutive_failures = endpoint_health.consecutive_failures + 1,
                last_failure_at = NOW()
        """, (endpoint_id,))
        return
    
    if row['state'] == 'half_open':
        # Probe failed — re-open circuit with longer cooldown
        cur.execute("""
            UPDATE endpoint_health
            SET state = 'open',
                opened_at = NOW(),
                cooldown_seconds = LEAST(cooldown_seconds * 2, 3600)
            WHERE endpoint_id = %s
        """, (endpoint_id,))
    
    elif (row['state'] == 'closed' and 
          row['consecutive_failures'] >= row['failure_threshold']):
        # Threshold reached — open circuit
        cur.execute("""
            UPDATE endpoint_health
            SET state = 'open',
                opened_at = NOW(),
                cooldown_seconds = 300  -- reset to 5 minutes
            WHERE endpoint_id = %s
        """, (endpoint_id,))


def record_success(cur, endpoint_id):
    """
    Record a delivery success. Close circuit if half_open.
    """
    cur.execute("""
        UPDATE endpoint_health
        SET consecutive_failures = 0,
            last_success_at = NOW(),
            state = 'closed'
        WHERE endpoint_id = %s
    """, (endpoint_id,))

大多数实现都忽略了半开状态到全开状态的升级机制，以及双倍冷却时间。如果在探测期间（半开状态）某个端点发生故障，你肯定不希望在 5 分钟后重试。因为该端点仍然处于故障状态。应该将冷却时间加倍至 10 分钟，然后是 20 分钟，最终上限为 1 小时。这样可以防止熔断机制变成周期性的频繁故障。

延迟百分位跟踪

平均值会骗人。一个平均响应时间为 200 毫秒的终端，可能 95% 的情况下响应时间为 50 毫秒，而剩下的 5% 情况下响应时间为 3,000 毫秒。平均值看起来似乎没问题。但 P95 指标揭示了一个影响每 20 次交付中 1 次的问题。

蟒蛇

def record_latency(cur, endpoint_id, response_ms, status_code):
    """
    Record a latency measurement and update percentile stats.
    """
    cur.execute("""
        INSERT INTO endpoint_latency 
            (endpoint_id, response_ms, status_code)
        VALUES (%s, %s, %s)
    """, (endpoint_id, response_ms, status_code))


def update_latency_percentiles(cur, endpoint_id, window_hours=24):
    """
    Calculate P50, P95, P99 from the rolling window.
    Uses PostgreSQL's percentile_cont for exact percentiles.
    """
    cur.execute("""
        SELECT 
            COUNT(*) as sample_count,
            percentile_cont(0.50) WITHIN GROUP 
                (ORDER BY response_ms) AS p50,
            percentile_cont(0.95) WITHIN GROUP 
                (ORDER BY response_ms) AS p95,
            percentile_cont(0.99) WITHIN GROUP 
                (ORDER BY response_ms) AS p99
        FROM endpoint_latency
        WHERE endpoint_id = %s
          AND measured_at >= NOW() - INTERVAL '%s hours'
    """, (endpoint_id, window_hours))
    
    row = cur.fetchone()
    
    if row and row['sample_count'] > 0:
        cur.execute("""
            UPDATE endpoint_health
            SET p50_ms = %s,
                p95_ms = %s,
                p99_ms = %s,
                sample_count = %s
            WHERE endpoint_id = %s
        """, (
            int(row['p50']), int(row['p95']), 
            int(row['p99']), row['sample_count'],
            endpoint_id
        ))
    
    return row


def get_slow_endpoints(cur, p95_threshold_ms=2000):
    """
    Find endpoints whose P95 latency exceeds the threshold.
    These are candidates for investigation or circuit opening.
    """
    cur.execute("""
        SELECT endpoint_id, endpoint_url, 
               p50_ms, p95_ms, p99_ms, sample_count,
               state, consecutive_failures
        FROM endpoint_health
        WHERE p95_ms > %s
          AND sample_count >= 20  -- need sufficient samples
        ORDER BY p95_ms DESC
    """, (p95_threshold_ms,))
    
    return cur.fetchall()

PostgreSQL的 percentile_cont 是一个有序集聚合函数，用于计算精确的百分位数。对于大型数据集，您应该切换到 percentile_disc （返回实际观测值而非插值）或使用 t-digest 近似值。对于 24 小时窗口的 webhook 发送，原始数据的精确百分位数在每个端点约 100,000 万次测量值以内都足够快。

此 get_slow_endpoints 我每 15 分钟运行一次此函数进行计划检查。P95 值超过 2 秒的端点会被标记出来进行调查。P95 值超过 5 秒的端点的熔断阈值会被降低——在熔断器打开之前，允许的连续失败次数会减少，因为每次失败的交付都会占用一个工作线程，直到超时为止。

监控 Webhook 交付健康状况

这是我每五分钟运行一次的监控查询。它会生成整个交付管道的单行运行状况摘要：

SQL

SELECT
    -- Queue depth
    COUNT(*) FILTER (WHERE status = 'pending') AS pending,
    COUNT(*) FILTER (WHERE status = 'failed') AS awaiting_retry,
    COUNT(*) FILTER (WHERE status = 'in_flight') AS in_flight,
    COUNT(*) FILTER (WHERE status = 'dead_letter') AS dead_letter,
    
    -- Delivery rate (last hour)
    COUNT(*) FILTER (
        WHERE status = 'delivered' 
        AND delivered_at >= NOW() - INTERVAL '1 hour'
    ) AS delivered_last_hour,
    
    -- Failure rate (last hour)
    COUNT(*) FILTER (
        WHERE status IN ('failed', 'dead_letter')
        AND created_at >= NOW() - INTERVAL '1 hour'
    ) AS failed_last_hour,
    
    -- Oldest undelivered
    MIN(created_at) FILTER (
        WHERE status IN ('pending', 'failed')
    ) AS oldest_pending,
    
    -- Average delivery latency (last hour, successful only)
    AVG(last_response_ms) FILTER (
        WHERE status = 'delivered'
        AND delivered_at >= NOW() - INTERVAL '1 hour'
    ) AS avg_delivery_ms_last_hour

FROM webhook_queue;

此 oldest_pending 在这个查询中，值是最重要的指标。如果它比最大重试窗口（所有退避延迟的总和）还要旧，则说明结构上存在问题——可能是工作进程卡住了，端点被黑洞吞噬了，或者队列增长速度超过了清空速度。

我针对以下三种情况发出警报：死信数量增加（端点永久失效且无人调查）、待处理队列超过 30 分钟（投递滞后）以及 每个端点的 P95 延迟超过阈值，表明交付可靠性下降第三个是预警信号——延迟增加先于故障发生。一个原本响应时间为 200 毫秒的端点，如果开始需要 3 秒才能响应，则说明它即将超时。

死信队列不仅仅是存储系统

大多数团队都会使用死信队列，将其作为一张表，用于存放失败的 Webhook 数据。他们会在事件响应期间偶尔检查一下。这是一种浪费。

死信队列是你最有价值的调试数据集。每一行都代表系统多次尝试投递但最终放弃的邮件。死信的模式能告诉你一些成功指标永远无法提供的信息。

蟒蛇

def analyze_dead_letters(cur, hours=24):
    """
    Analyze recent dead letter entries for patterns.
    Returns per-endpoint failure analysis.
    """
    cur.execute("""
        SELECT 
            endpoint_id,
            endpoint_url,
            COUNT(*) AS dead_count,
            
            -- Most common error
            MODE() WITHIN GROUP (ORDER BY last_error) AS primary_error,
            
            -- Most common status code
            MODE() WITHIN GROUP (ORDER BY last_status_code) 
                AS primary_status_code,
            
            -- Timing
            MIN(created_at) AS first_dead,
            MAX(created_at) AS last_dead,
            
            -- Average attempts before giving up
            AVG(attempt_count)::INTEGER AS avg_attempts
            
        FROM webhook_queue
        WHERE status = 'dead_letter'
          AND created_at >= NOW() - INTERVAL '%s hours'
        GROUP BY endpoint_id, endpoint_url
        ORDER BY dead_count DESC
        LIMIT 20
    """, (hours,))
    
    return cur.fetchall()

当我审查死信时，我会寻找三种模式。

集群故障： 同一终端在同一小时内出现 50 个死信，表示该终端已宕机且未在重试窗口期内恢复。措施：延长重试窗口或实施手动重新排队。

状态码模式： 401/403 死信数量激增表示端点已轮换凭据，但无人更新 webhook 配置。429（请求过多）错误数量激增表示您已超过其速率限制，需要进行限流。

逐步积累： 单个端点每天出现 2-3 个死信，均匀分布。这是最隐蔽的模式——端点大部分时间都能正常工作，但会间歇性地出现故障，随着时间的推移耗尽重试次数。修复方法通常是增加重试次数。 max_attempts 针对特定端点或减少超时时间。

运行 Worker

将所有内容串联起来的主回路：

蟒蛇

import signal
import sys

running = True

def shutdown_handler(signum, frame):
    global running
    running = False
    print(f"Received signal {signum}, shutting down gracefully...")

signal.signal(signal.SIGTERM, shutdown_handler)
signal.signal(signal.SIGINT, shutdown_handler)

def main():
    print("Webhook delivery worker starting...")
    
    while running:
        try:
            deliver_webhooks(batch_size=50)
        except Exception as e:
            print(f"Worker error: {e}")
            time.sleep(5)  # back off on errors
            continue
        
        # Update latency stats every 100 iterations
        # (cheap operation, doesn't need to run every loop)
        if int(time.time()) % 100 == 0:
            conn = get_connection()
            cur = conn.cursor(cursor_factory=RealDictCursor)
            try:
                cur.execute(
                    "SELECT DISTINCT endpoint_id FROM endpoint_health"
                )
                for row in cur.fetchall():
                    update_latency_percentiles(cur, row['endpoint_id'])
                conn.commit()
            finally:
                cur.close()
                conn.close()
        
        # Poll interval — 500ms keeps latency low without
        # hammering the database
        time.sleep(0.5)

    print("Worker shut down cleanly.")

if __name__ == '__main__':
    main()

此 SIGTERM 处理程序对于容器化环境中的正常关闭至关重要。当 Kubernetes 发送 SIGTERM 信号时，工作进程会完成当前批次的操作，提交事务并退出。如果没有它，数据行就会卡在容器中。 in_flight 状态为：没有工作进程正在处理它们。

这个系统无法做到的事情（以及当你需要更多功能时）

此实现方案在单个 PostgreSQL 实例上使用 2-3 个工作进程，每分钟最多可处理约 10,000 条订单。超过此限制，需要进行三项更改。

首先，将 PostgreSQL 队列替换为 Redis Streams 或 RabbitMQ。 SELECT FOR UPDATE SKIP LOCKED 这种模式在高吞吐量下会导致队列表写入争用。专用消息代理可以消除这种争用。

其次，添加基于端点的速率限制。某些接收端点有速率限制（例如每分钟 100 个请求，每小时 1,000 个请求）。如果没有客户端速率限制，您将超出其配额并收到 429 错误。为每个端点实现一个令牌桶。

第三，添加请求签名。有效负载上的 HMAC-SHA256 签名可以让接收端验证 webhook 是否来自您的系统，以及是否在传输过程中被篡改。对于任何发送财务数据的 webhook 系统来说，这都是基本要求。

本文介绍的系统是基础架构。它能够处理所有 Webhook 发送系统（无论规模大小）都需要解决的难题，例如重试逻辑、熔断机制、延迟测量和死信分析。您需要添加的具体组件（例如消息代理、速率限制器和请求签名）则取决于您的吞吐量和安全需求。

最重要的部分恰恰是大多数团队会忽略的部分：衡量交付系统本身的性能。如果你无法回答“过去 24 小时内 P95 交付到 X 端点的延迟是多少”，那么你就是在盲目操作。首先要构建监控系统，其他一切都会随之而来。

Webhook 交付常见问题解答

webhook 发送方是否应该承诺只发送一次？

通常情况下不需要。发送方应该让重试过程可见，并提供稳定的事件 ID；接收方应该保证处理过程的幂等性，以便同一个事件可以多次传递而不会造成业务结果的重复。

哪些失败需要重试？

仅对合约中归类为暂时性故障（例如网络错误、超时和特定服务器响应）进行重试。对于格式错误的请求、身份验证失败或其他永久性错误，如果没有明确的补救措施，请勿反复重试。

团队何时应该摆脱基于数据库的队列？

当衡量出的争用情况、积压队列的积压时间、吞吐量或运维恢复需求表明数据库队列不再满足交付合同时，就应该进行迁移。容量决策应基于观察到的工作负载，而不是笼统的请求速率声明。