使用 Python 处理 webhook：FastAPI + Redis

使用 FastAPI 和 Redis 队列实现可靠的 Webhook 处理

支付服务的 Webhook 要求快速响应——通常在 5 秒内完成。如果在端点中采用同步处理，当数据库或外部服务延迟时，就会导致超时。一个简单的实现方式如下：

@app.post("/webhook")
async def webhook(request: dict):
    update_database(request)
    send_email_to_user(request)
    return {"status": "ok"}

这种做法非常脆弱：处理过程中服务器崩溃会导致数据丢失，重复通知则会因数据库唯一约束而引发错误。解决方案？通过任务队列将接收与处理解耦。

使用任务队列的架构

API 端点接收 Webhook，验证签名，确保幂等性，并将任务加入 Redis 队列。一个独立的工作者异步执行业务逻辑。优势包括：

• HTTP 响应即时（低于 50 毫秒）。
• 即使工作者崩溃，任务也能持久保存。
• 可横向扩展（多个工作者可同时运行）。

组件构成：

• FastAPI 作为 API 层。
• Redis 作为消息队列（使用 LPUSH 和 BRPOP）。
• 工作者在隔离进程中运行。

实现 API 接收器

端点专注于安全性和速度。它检查 HMAC 签名，通过 transaction_id 防止重复，并将任务添加到队列中。

from fastapi import FastAPI, HTTPException, Header
from pydantic import BaseModel
import redis
import json
import hashlib
import os

app = FastAPI()
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
SECRET = os.getenv('WEBHOOK_SECRET', 'test_secret')

def check_signature(body: str, sign: str) -> bool:
    my_sign = hashlib.sha256(f"{body}{SECRET}".encode()).hexdigest()
    return my_sign == sign

@app.post("/api/payment")
async def payment_webhook(request: dict, x_sign: str = Header(None)):
    if not check_signature(json.dumps(request), x_sign):
        raise HTTPException(status_code=403, detail="Bad sign")
    
    task_id = request.get('transaction_id')
    if r.exists(f"processed:{task_id}"):
        return {"status": "duplicate"}
    
    task = {
        "id": task_id,
        "amount": request.get('amount'),
        "user_id": request.get('user_id')
    }
    r.lpush("payment_queue", json.dumps(task))
    
    return {"status": "accepted"}

关键实践：

• HMAC 签名防止伪造请求。
• Redis 键 processed:{id} 确保幂等性。
• LPUSH 高效地将任务加入队列。

后台任务处理器

工作者在循环中运行，使用 BRPOP（5 秒超时）拉取任务，执行逻辑，并标记为已处理（设置 1 天过期时间）。失败时，延迟后重试。

import redis
import json
import time
import logging

logging.basicConfig(level=logging.INFO)
log = logging.getLogger(__name__)

def process_task(data):
    log.info(f"Processing payment {data['id']}")
    time.sleep(1) 
    return True

def main():
    r = redis.Redis(host='localhost', port=6379)
    log.info("Worker started...")
    
    while True:
        task = r.brpop("payment_queue", timeout=5)
        if not task:
            continue
        _, raw_data = task
        data = json.loads(raw_data)
        
        try:
            success = process_task(data)
            if success:
                r.setex(f"processed:{data['id']}", 86400, "1")
                log.info(f"Task {data['id']} completed")
        except Exception as e:
            log.error(f"Error: {e}")
            r.lpush("payment_queue", raw_data)
            time.sleep(5)

if __name__ == "__main__":
    main()

常见问题及解决方案

• 缺少签名验证：易受伪造 POST 攻击。修复方法：使用带密钥的 HMAC。
• Redis 数据易失：重启后数据丢失。解决方案：启用 RDB/AOF 持久化，或改用 PostgreSQL 作为队列。
• 工作者重试循环：任务陷入无限重试。修复方法：记录尝试次数，使用死信队列，实施指数退避。

其他风险：

• 外部服务超时：在 process_task 中设置超时。
• 扩展挑战：监控队列长度（LLEN），按需扩展工作者数量。

核心要点

• 幂等性：如 processed:{id} 这类键可防止重复处理。
• 安全性：所有传入 Webhook 必须验证 HMAC 签名。
• 可靠性：队列可在故障期间保留任务。
• 可扩展性：独立工作者支持并行处理。
• 可观测性：记录队列指标和错误日志以提升监控能力。

— Editorial Team