怎样用脚本分析Webhook延迟？

wen 实用脚本 2026-07-22 1

本文目录导读：

怎样用脚本分析Webhook延迟？

接收端记录法（推荐，最精准）
发送端记录法（适合控制发送方）
网络层分析（适合排查网络问题）
批量分析脚本（定期统计）
注意事项
推荐组合方案

分析Webhook延迟通常需要从发送端、接收端和网络链路三个维度入手，以下是几种实用的脚本分析方法,涵盖不同场景：

接收端记录法（推荐，最精准）

在接收Webhook的服务器上，编写脚本记录收到请求的时间戳，并与Webhook头部的X-Request-Id或自定义的时间戳进行对比。

Python示例（Flask）

from flask import Flask, request, jsonify
import time
import json
import logging
app = Flask(__name__)
@app.route('/webhook', methods=['POST'])
def webhook_handler():
    # 记录接收时刻（服务器时间）
    receive_time = time.time()
    # 获取发送端携带的时间戳（假设在Header或Body中）
    # 常见格式：X-Timestamp / X-Delivery-Time
    sent_timestamp = request.headers.get('X-Timestamp')
    body = request.get_json()
    # 如果是JSON Body中的时间戳
    body_timestamp = body.get('timestamp') if body else None
    # 计算延迟
    delays = {
        'receive_time': receive_time,
        'sent_timestamp': sent_timestamp,
        'body_timestamp': body_timestamp,
        'latency_from_header': (receive_time - float(sent_timestamp)) if sent_timestamp else None,
        'latency_from_body': (receive_time - float(body_timestamp)) if body_timestamp else None
    }
    # 记录日志（用于后续分析）
    logging.info(json.dumps(delays))
    return jsonify({'status': 'ok'}), 200
if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO, filename='webhook_latency.log')
    app.run(port=8080)

Node.js示例（Express）

const express = require('express');
const app = express();
app.use(express.json());
app.post('/webhook', (req, res) => {
    const receiveTime = Date.now();
    const sentTime = req.headers['x-timestamp'] || req.body?.timestamp;
    const latency = sentTime ? (receiveTime - parseInt(sentTime)) : 'N/A';
    console.log(JSON.stringify({
        receiveTime,
        sentTime,
        latency_ms: latency
    }));
    // 可将数据写入文件或数据库
    res.json({ status: 'ok' });
});
app.listen(8080);

发送端记录法（适合控制发送方）

如果你能控制Webhook的发送端（例如自己搭建的推送服务），可以在发送前记录时间戳,并在接收端返回确认后计算总耗时。

Bash脚本示例

#!/bin/bash
# 发送Webhook并计算耗时
URL="https://your-receiver.com/webhook"
PAYLOAD='{"event":"test","timestamp":'$(date +%s%N)'}'
# 记录开始时间
START=$(date +%s%N)
# 发送请求
curl -s -X POST "$URL" \
     -H "Content-Type: application/json" \
     -d "$PAYLOAD"
# 计算总耗时
END=$(date +%s%N)
LATENCY_MS=$(( (END - START) / 1000000 ))
echo "Total latency: ${LATENCY_MS} ms"
echo "$(date '+%Y-%m-%d %H:%M:%S'),${LATENCY_MS}" >> latency_log.csv

网络层分析（适合排查网络问题）

使用tcpdump + tshark或curl配合-w参数，分析TCP握手、TLS、传输等各阶段耗时。

使用curl分析详细耗时

#!/bin/bash
# 使用curl的 -w 参数提取各阶段耗时
URL="https://your-receiver.com/webhook"
PAYLOAD='{"test":"data"}'
curl -w "\n
HTTP_CODE: %{http_code}\n
DNS_TIME: %{time_namelookup}s\n
TCP_TIME: %{time_connect}s\n
SSL_TIME: %{time_appconnect}s\n
TOTAL_TIME: %{time_total}s\n
" \
-s -o /dev/null \
-X POST "$URL" \
-H "Content-Type: application/json" \
-d "$PAYLOAD"

Python版（使用socket模拟）

import socket
import ssl
import time
def analyze_network_latency(host, port=443, path='/webhook'):
    """分析TCP+TLS+HTTP各阶段耗时"""
    results = {}
    # DNS解析
    start = time.time()
    addr = socket.getaddrinfo(host, port)[0][4][0]
    results['dns'] = time.time() - start
    # TCP连接
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    start = time.time()
    sock.connect((addr, port))
    results['tcp'] = time.time() - start
    # TLS握手（如果有）
    context = ssl.create_default_context()
    ssl_sock = context.wrap_socket(sock, server_hostname=host)
    start = time.time()
    ssl_sock.do_handshake()
    results['tls'] = time.time() - start
    # HTTP请求
    request = f"POST {path} HTTP/1.1\r\nHost: {host}\r\nContent-Length: 20\r\n\r\n{{\"test\":\"data\"}}"
    start = time.time()
    ssl_sock.sendall(request.encode())
    response = ssl_sock.recv(4096)
    results['http_transfer'] = time.time() - start
    ssl_sock.close()
    return results
# 使用示例
print(analyze_network_latency('your-webhook-receiver.com'))

批量分析脚本（定期统计）

将采集到的延迟数据汇总，计算分位数、平均值等。

Python统计脚本

import json
import statistics
from collections import Counter
# 假设日志文件每行是一个JSON对象
with open('webhook_latency.log', 'r') as f:
    latencies = []
    for line in f:
        record = json.loads(line.strip())
        lat = record.get('latency_from_header')
        if lat and 0 < lat < 60:  # 过滤异常值（超过60秒）
            latencies.append(lat)
if latencies:
    print(f"样本数: {len(latencies)}")
    print(f"平均延迟: {statistics.mean(latencies):.2f}s")
    print(f"中位数: {statistics.median(latencies):.2f}s")
    print(f"P95: {sorted(latencies)[int(len(latencies)*0.95)]:.2f}s")
    print(f"P99: {sorted(latencies)[int(len(latencies)*0.99)]:.2f}s")
    print(f"最大延迟: {max(latencies):.2f}s")

注意事项

时间同步是关键：发送端和接收端的系统时间必须同步（使用NTP），否则测出的延迟可能不准（甚至出现负值）。
业务逻辑处理时间：上述方法测量的是网络传输 + 接收端入站延迟，如果接收端有复杂的业务处理（如数据库写入）,应单独计时。
重试与去重：Webhook可能因失败重发,日志中需要加入唯一ID去重。
工具选择：
- 需要长期监控 → 接收端记录法 + 日志聚合（如ELK）。
- 只需临时排查 → 发送端脚本 + curl -w。
- 需要深入网络 → tcpdump或Wireshark抓包。