轮询写了一下午还没跑通？Webhook回调让我把半天活压进15分钟

一个真实的时间账

前阵子我要接一个长文本摘要的异步任务。第一版我用了最朴素的办法：客户端起个循环，每3秒调一次「查结果」接口，超时30次就报错。写代码花了20分钟，看起来还行。真正的时间黑洞在后面：

• 任务平均耗时40秒，轮询间隔设3秒，一次成功要空转十几轮请求；
• 任务失败时循环逻辑写得含糊，偶发卡死，我花了一个多小时排查；
• 用户量大一点之后，轮询请求把接口日志刷得没法看，排查一次线上问题又搭进去40分钟。

一个下午就这样没了。后来我改成 Webhook 回调的方式，同样的功能，核心代码不到30行，写完跑通只用了15分钟。这篇文章把这套「异步任务 + 回调」的接法完整过一遍，以 ThisToken.AI 的接口为例，你可以直接复制代码开始跑。

为什么 Webhook 比轮询省

轮询的本质是「我不知道你好了没有，我反复问」。它的成本结构是：

1. 请求数量浪费：任务实际处理只需1次调用，轮询却带来10~20次查询请求；
2. 延迟被间隔锁死：任务2秒就完成，你还是要等到下一个轮询点才知道；
3. 失败路径复杂：循环、超时、退避这些逻辑都要自己写，写错就是事故。

Webhook 把模型反过来：你告诉服务方「好了叫我」，然后你该干什么干什么。一次任务，一次回调，没有中间的空转。对独立开发者来说，最直观的收益是——你不用再写轮询循环和它的超时分支，少写的代码就是少踩的坑。

第一步：注册并拿到 API Key

1. 打开 https://api.thistoken.ai/register ，用邮箱注册一个账号（流程很短，邮箱验证即可）；
2. 登录后进入控制台，在「API Keys」页面点击创建，给 Key 起个名字（比如 webhook-demo）；
3. 复制 Key 保存好。它只在创建时完整展示一次，丢了只能重建。

Key 的调用额度与计费方式，以官网价格页为准，不要轻信任何第三方转述的数字。

第二步：准备一个能被回调的地址

Webhook 需要一个公网可达的 HTTPS 地址接收回调。本地开发阶段不用买服务器，用内网穿透工具即可，比如：

# 以常见的内网穿透工具为例，本地起在 8000 端口
ngrok http 8000

它会给你一个类似 https://xxxx.ngrok-free.app 的临时地址。记下它，回调地址就是 https://xxxx.ngrok-free.app/webhook。

第三步：跑通第一段代码

下面这段 Python 代码做了三件事：起一个接收回调的 HTTP 服务、提交一个异步任务、在回调里取出结果。

import time
import hashlib
import hmac
import threading
import requests
from flask import Flask, request, jsonify

API_KEY = "sk-你的APIKey"
BASE_URL = "https://api.thistoken.ai/v1"
WEBHOOK_URL = "https://xxxx.ngrok-free.app/webhook"  # 换成你的穿透地址
SECRET = "my-webhook-secret"  # 用于校验回调签名

app = Flask(__name__)
result_box = {}

@app.route("/webhook", methods=["POST"])
def webhook():
    # 1. 校验签名，防止伪造回调
    raw = request.get_data()
    received = request.headers.get("X-Signature", "")
    expected = hmac.new(
        SECRET.encode(), raw, hashlib.sha256
    ).hexdigest()
    if not hmac.compare_digest(expected, received):
        return jsonify({"error": "invalid signature"}), 401

    # 2. 取出任务结果
    payload = request.get_json()
    task_id = payload.get("task_id")
    result_box[task_id] = payload.get("result")
    print(f"收到回调: task_id={task_id}, status={payload.get('status')}")
    return jsonify({"received": True}), 200

def submit_task():
    """提交异步任务，并登记回调地址"""
    resp = requests.post(
        f"{BASE_URL}/tasks",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={
            "type": "summarize",
            "input": "这里放一段需要摘要的长文本……",
            "callback_url": WEBHOOK_URL,
            "callback_secret": SECRET,
        },
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json()["task_id"]

if __name__ == "__main__":
    task_id = submit_task()
    print(f"任务已提交: {task_id}，等待回调……")

    # 本地起服务接收回调（生产环境请放到独立进程/机器上）
    threading.Thread(
        target=lambda: app.run(port=8000),
        daemon=True,
    ).start()

    for _ in range(60):
        if task_id in result_box:
            print("任务结果:", result_box[task_id])
            break
        time.sleep(1)
    else:
        print("超时未收到回调，请检查穿透地址是否可达")

运行它，正常情况下十几秒后你就能在终端看到回调结果。注意几个细节：

• 签名校验必须做。任何拿到你回调地址的人都能给你发假结果，HMAC 校验能挡掉伪造请求；
• 回调要快速返回 200。回调处理逻辑要轻，收到后先落库/入队，耗时处理放到后台，否则可能触发服务方的重试；
• 回调端点要幂等。服务方通常会重试未确认的回调，同一个 task_id 可能收到多次，按 task_id 去重。

效率对比，算一笔小账

把两种方式放在一起看：

我自己的体感是：同样的异步功能，从「写完调通一个下午」变成「15分钟跑通，剩下的时间喝咖啡」。如果按上面的对照做粗略估算，一个每天跑几千次任务的小服务，改用回调后查询请求量能减少九成以上——这部分省下的不只是服务器开销，还有你盯日志的时间。

收尾

异步任务和回调是很多 AI 能力接入的标准姿势，第一次跑通之后你会发现模式是通用的。如果这篇文章对你有帮助，去 https://api.thistoken.ai/register 注册一个账号，把上面那段代码跑起来——15分钟后，你的第一个回调就会到。

---

想直接跑通示例？访问 https://api.thistoken.ai/register 注册 ThisToken.AI，获取 API Key 后即可开始。