Payment API幂等性设计：用Stripe、支付宝、微信支付和2C2P防止重复扣款

用户点击"立即支付"，请求超时，页面没有任何响应——系统接下来会发生什么？

如果Payment API没有实现幂等性，答案可能是：用户被重复扣款。订单被重复创建。Webhook触发后服务器崩溃，库存已经扣减但订单记录从未提交。

这些不是罕见的边缘案例，而是真实发生在双十一大促期间、支付网关响应慢时、移动端网络断连时，或者Kubernetes Pod在数据库写入和API响应之间被驱逐时的生产故障模式。

本文将系统讲解如何正确设计幂等性Payment API——提供FastAPI + PostgreSQL的完整实现代码，以及支付宝、微信支付、2C2P和Stripe的具体示例，覆盖出海电商的主流支付场景。

什么是Payment API中的幂等性

幂等操作（Idempotent operation）是指无论执行多少次，结果始终与执行一次相同的操作。

在Payment API的语境下，这意味着：无论"创建扣款"请求因网络超时、客户端重试或Webhook重复投递而被发送多少次，系统只会创建一笔扣款，金额固定，用户账户只被扣一次。

实现这一保证的机制是幂等键（Idempotency Key）——由客户端生成的唯一标识符，绑定到每次支付尝试，并在每次重试时随请求一并发送。服务端用这个Key来返回缓存结果，而不是重新处理请求。

核心要点：幂等键 = 每次支付意图对应的唯一ID，由客户端生成，每次重试都携带。服务端凭此识别重复请求，跳过处理，直接返回缓存的响应结果。

为什么幂等性对电商和跨境支付系统至关重要

分布式系统存在三种需要幂等性的故障模式：

Pre-server failure（请求未到达服务器）：请求从未到达服务器，这种情况下重试是安全的。
Mid-processing failure（处理中途失败）：请求到达服务器，开始处理，但在完成前失败。无幂等性的重试 = 重复扣款。
Post-processing failure（响应未到达客户端）：服务器成功处理了请求，但确认响应从未到达客户端。无幂等性的重试 = 重复扣款。

只有第一种场景可以安全地直接重试。第二和第三种——也是生产环境中最常见的——需要在支付网关侧和自身后端侧都建立幂等性层。

在中国市场及东南亚出海场景中，以下因素进一步放大了风险：

微信支付、支付宝的异步通知：支付结果通知（notify_url回调）采用at-least-once投递策略，同一笔交易的成功通知可能到达多次。
跨境收款场景：网络链路更长，超时率更高，客户端自动重试更为常见。
电子发票合规：重复的交易记录会导致增值税专用发票数据不一致，影响税务申报的准确性。
人民银行相关监管要求：金融科技系统的交易日志完整性具有合规意义，重复记录可能引发审计风险。

各支付网关的幂等性实现方式

支付宝（Alipay）幂等性

支付宝通过请求参数中的out_trade_no实现幂等性——这是商户侧的唯一交易号，功能与2C2P的invoiceNo类似。同一out_trade_no的重复请求会返回原始交易结果，不会产生新的扣款。

import hashlib
import uuid
from datetime import datetime

def build_alipay_payment_request(order_id: str, amount_cny: float) -> dict:
    out_trade_no = f"ORDER-{order_id}"  # 这就是支付宝的幂等键，必须全局唯一

    payload = {
        "app_id": "YOUR_APP_ID",
        "method": "alipay.trade.create",
        "charset": "utf-8",
        "sign_type": "RSA2",
        "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "version": "1.0",
        "biz_content": {
            "out_trade_no": out_trade_no,   # 幂等键
            "total_amount": f"{amount_cny:.2f}",
            "subject": f"Order {order_id}",
            "product_code": "FAST_INSTANT_TRADE_PAY",
        }
    }
    return payload

关键约束：out_trade_no必须在调用支付宝之前生成并持久化到数据库。在构建请求的函数内部动态生成会绕过幂等保护——重试时会产生新的ID，支付宝无法识别为重复请求。

微信支付（WeChat Pay）幂等性

微信支付使用out_trade_no字段，与支付宝的机制完全一致：相同的out_trade_no在有效期内只会产生一笔支付。

import hashlib
import time
import uuid

def build_wechatpay_payment_request(order_id: str, amount_fen: int) -> dict:
    out_trade_no = f"ORDER-{order_id}"  # 微信支付的幂等键

    payload = {
        "appid": "YOUR_APPID",
        "mchid": "YOUR_MCHID",
        "description": f"Order {order_id}",
        "out_trade_no": out_trade_no,       # 幂等键，同一笔订单保持不变
        "amount": {
            "total": amount_fen,             # 单位：分（人民币最小单位）
            "currency": "CNY"
        },
        "notify_url": "https://your-domain.com/webhooks/wechatpay",
    }
    return payload

微信支付Webhook注意事项：微信支付的notify_url回调同样采用at-least-once投递。系统需要对out_trade_no进行去重判断，避免重复处理同一笔支付成功通知——这正是本文后续讲解的Webhook去重模式的核心场景。

Stripe幂等键（Python）

Stripe通过Idempotency-Key HTTP Header实现幂等性，客户端生成UUID，随请求一并发送，Stripe缓存响应24小时。

import stripe
import uuid

stripe.api_key = "sk_live_..."

idempotency_key = str(uuid.uuid4())  # 每次支付尝试生成一次，由客户端生成

charge = stripe.PaymentIntent.create(
    amount=100,          # 1.00 USD（单位：分）
    currency="usd",
    payment_method="pm_xxxx",
    confirm=True,
    idempotency_key=idempotency_key,
)

Stripe的约束：使用相同的Idempotency-Key但不同的参数会返回400错误。同一个Key必须始终对应同一个支付意图——Key是一种承诺，而不仅仅是一个标签。Stripe同样会缓存失败响应，防止客户端和服务端之间的状态不一致。

2C2P幂等性：通过`invoiceNo`实现

2C2P通过请求payload中的invoiceNo字段实现幂等性——这是商户提供的每笔交易的唯一标识符。提交相同的invoiceNo两次，会返回第一次的处理结果。

import hashlib
import hmac
import json

def build_2c2p_payment_request(order_id: str, amount_usd: float, secret_key: str) -> dict:
    invoice_no = f"ORDER-{order_id}"  # 这就是2C2P的幂等键

    payload = {
        "merchantID": "YOUR_MERCHANT_ID",
        "invoiceNo": invoice_no,
        "description": f"Order {order_id}",
        "amount": f"{amount_usd:.2f}",
        "currencyCode": "840",         # USD ISO 4217
        "paymentChannel": ["CC"],
    }

    payload_str = json.dumps(payload, separators=(",", ":"))
    signature = hmac.new(
        secret_key.encode(), payload_str.encode(), hashlib.sha256
    ).hexdigest()
    payload["signature"] = signature

    return payload

仅靠网关侧的幂等性还不够

支付网关只能保护PSP层面的重复扣款，无法保护你自己的数据库免受以下问题影响：

超时后客户端重试导致订单记录重复创建
两个并发重试导致库存被扣减两次
Webhook重复投递导致发送两封订单确认邮件

要解决这些问题，必须在自己的服务端建立幂等性层。

PostgreSQL去重表设计

核心模式：在执行任何与支付相关的写操作之前，先检查这个幂等键是否已经处理过。如果是，返回存储的结果；如果否，获取锁、执行处理、存储结果。

CREATE TABLE idempotency_keys (
    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    key             TEXT NOT NULL UNIQUE,
    request_path    TEXT NOT NULL,
    request_hash    TEXT NOT NULL,      -- 请求体的SHA-256哈希
    response_status INT,
    response_body   JSONB,
    locked_at       TIMESTAMPTZ,        -- 处理开始时设置（分布式锁）
    completed_at    TIMESTAMPTZ,        -- 处理完成时设置
    created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
    expires_at      TIMESTAMPTZ NOT NULL DEFAULT now() + INTERVAL '24 hours'
);

CREATE INDEX ON idempotency_keys (key);
CREATE INDEX ON idempotency_keys (expires_at);  -- 供夜间清理任务使用

locked_at字段是关键所在：它防止两个使用相同Key的并发请求同时通过"尚未处理"检查。ON CONFLICT DO NOTHING插入加上锁检查的组合，构成了无需Redis的分布式互斥锁。

FastAPI实现：IdempotencyGuard Dependency

以下是可复用的FastAPI Dependency，用于在所有支付端点上强制执行幂等性。

核心函数

# app/dependencies/idempotency.py

import hashlib
import json
from typing import Optional

from fastapi import Depends, Header, HTTPException, Request, status
from fastapi.responses import JSONResponse
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import text

from app.db import get_db

async def get_idempotency_record(db: AsyncSession, key: str) -> Optional[dict]:
    result = await db.execute(
        text("""
            SELECT id, request_path, request_hash, response_status,
                   response_body, locked_at, completed_at
            FROM idempotency_keys
            WHERE key = :key AND expires_at > now()
        """),
        {"key": key},
    )
    return result.mappings().first()

async def acquire_idempotency_lock(
    db: AsyncSession, key: str, request_path: str, request_hash: str
):
    """
    INSERT ... ON CONFLICT DO NOTHING 确保对于给定的Key，
    同一时刻只有一个并发请求能够获取锁。
    """
    await db.execute(
        text("""
            INSERT INTO idempotency_keys (key, request_path, request_hash, locked_at)
            VALUES (:key, :path, :hash, now())
            ON CONFLICT (key) DO NOTHING
        """),
        {"key": key, "path": request_path, "hash": request_hash},
    )
    await db.commit()

async def complete_idempotency_key(
    db: AsyncSession, key: str, status_code: int, response_body: dict
):
    await db.execute(
        text("""
            UPDATE idempotency_keys
            SET response_status = :status,
                response_body   = :body,
                completed_at    = now()
            WHERE key = :key
        """),
        {"key": key, "status": status_code, "body": json.dumps(response_body)},
    )
    await db.commit()

class IdempotencyGuard:
    """
    FastAPI Dependency。检查Idempotency-Key Header，
    重放时返回缓存响应，处理中返回409，
    新请求则获取锁并继续处理。
    """

    def __init__(self, require: bool = True):
        self.require = require

    async def __call__(
        self,
        request: Request,
        idempotency_key: Optional[str] = Header(None, alias="Idempotency-Key"),
        db: AsyncSession = Depends(get_db),
    ):
        if not idempotency_key:
            if self.require:
                raise HTTPException(
                    status_code=status.HTTP_400_BAD_REQUEST,
                    detail="该支付端点需要提供Idempotency-Key Header。",
                )
            return None

        body = await request.body()
        request_hash = hashlib.sha256(body).hexdigest()

        record = await get_idempotency_record(db, idempotency_key)

        if record:
            if record["request_hash"] != request_hash:
                raise HTTPException(
                    status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
                    detail="该Idempotency-Key已与不同的请求体一起使用过。",
                )
            if record["completed_at"]:
                # 已处理完成——返回缓存的结果
                return JSONResponse(
                    content=record["response_body"],
                    status_code=record["response_status"],
                    headers={"Idempotency-Replayed": "true"},
                )
            if record["locked_at"] and not record["completed_at"]:
                # 另一个并发请求正在处理中
                raise HTTPException(
                    status_code=status.HTTP_409_CONFLICT,
                    detail="该Idempotency-Key的请求正在处理中，请稍后重试。",
                )

        # 新Key——获取锁并继续处理
        await acquire_idempotency_lock(db, idempotency_key, request.url.path, request_hash)
        request.state.idempotency_key = idempotency_key
        request.state.idempotency_db = db
        return idempotency_key

支付端点

# app/routers/payments.py

from decimal import Decimal
from fastapi import APIRouter, Depends, HTTPException, Request
from pydantic import BaseModel

from app.dependencies.idempotency import IdempotencyGuard, complete_idempotency_key
from app.services.alipay import charge_via_alipay
from app.services.orders import create_order_record
from app.db import get_db

router = APIRouter()
idempotency_guard = IdempotencyGuard(require=True)

class CreateChargeRequest(BaseModel):
    order_id: str
    amount_cny: Decimal
    payment_method: str  # "alipay" | "wechatpay" | "stripe"

@router.post("/payments/charge", dependencies=[Depends(idempotency_guard)])
async def create_charge(
    body: CreateChargeRequest,
    request: Request,
    db=Depends(get_db),
):
    idempotency_key = request.state.idempotency_key

    try:
        # out_trade_no直接复用order_id，保证支付宝侧也是幂等的
        charge = await charge_via_alipay(
            out_trade_no=body.order_id,
            amount_cny=float(body.amount_cny),
            idempotency_key=idempotency_key,
        )

        order = await create_order_record(db, body.order_id, charge["trade_no"])

        result = {
            "order_id": str(order.id),
            "trade_no": charge["trade_no"],
            "status": charge["trade_status"],
            "amount_cny": str(body.amount_cny),
        }

        await complete_idempotency_key(db, idempotency_key, 201, result)
        return result, 201

    except Exception as e:
        # 临时性错误（网络、PSP超时）→ 删除Key，允许客户端重试
        # 终端错误（余额不足、账户受限）→ 保留错误响应，防止重试循环
        await db.execute(
            text("DELETE FROM idempotency_keys WHERE key = :key"),
            {"key": idempotency_key},
        )
        await db.commit()
        raise HTTPException(status_code=502, detail=str(e))

Webhook去重：处理At-Least-Once投递

支付宝和微信支付的异步通知（notify_url回调）均采用at-least-once投递策略——同一笔交易的支付成功通知可能多次到达你的端点。Webhook Handler必须以Event ID（对于支付宝是out_trade_no，对于微信支付是transaction_id）作为去重键来实现幂等处理。

# app/routers/webhooks.py

import hashlib
from fastapi import APIRouter, Depends, HTTPException, Request
from sqlalchemy import text

from app.dependencies.idempotency import (
    acquire_idempotency_lock,
    complete_idempotency_key,
    get_idempotency_record,
)
from app.db import get_db

router = APIRouter()

@router.post("/webhooks/alipay")
async def alipay_webhook(request: Request, db=Depends(get_db)):
    form_data = await request.form()
    out_trade_no = form_data.get("out_trade_no")   # 商户订单号
    trade_status = form_data.get("trade_status")   # TRADE_SUCCESS | TRADE_FINISHED

    if not out_trade_no:
        raise HTTPException(status_code=400, detail="缺少out_trade_no字段")

    webhook_key = f"webhook:alipay:{out_trade_no}"
    record = await get_idempotency_record(db, webhook_key)

    if record and record["completed_at"]:
        # 支付宝要求对已处理的通知返回"success"字符串
        return "success"

    await acquire_idempotency_lock(
        db,
        key=webhook_key,
        request_path="/webhooks/alipay",
        request_hash=hashlib.sha256(str(dict(form_data)).encode()).hexdigest(),
    )

    if trade_status in ("TRADE_SUCCESS", "TRADE_FINISHED"):
        await mark_order_paid(db, out_trade_no)

    await complete_idempotency_key(db, webhook_key, 200, {"ok": True})

    # 支付宝要求返回纯文本"success"，否则会继续重试通知
    return "success"

@router.post("/webhooks/wechatpay")
async def wechatpay_webhook(request: Request, db=Depends(get_db)):
    payload = await request.json()
    transaction_id = payload.get("transaction_id")  # 微信支付交易单号

    if not transaction_id:
        raise HTTPException(status_code=400, detail="缺少transaction_id字段")

    webhook_key = f"webhook:wechatpay:{transaction_id}"
    record = await get_idempotency_record(db, webhook_key)

    if record and record["completed_at"]:
        return {"code": "SUCCESS", "message": "成功"}

    await acquire_idempotency_lock(
        db,
        key=webhook_key,
        request_path="/webhooks/wechatpay",
        request_hash=hashlib.sha256(str(payload).encode()).hexdigest(),
    )

    trade_state = payload.get("trade_state")
    if trade_state == "SUCCESS":
        out_trade_no = payload.get("out_trade_no")
        await mark_order_paid(db, out_trade_no)

    await complete_idempotency_key(db, webhook_key, 200, {"ok": True})
    return {"code": "SUCCESS", "message": "成功"}

注意：支付宝要求Webhook Handler在处理成功后返回纯文本字符串"success"，否则支付宝服务器会持续重试通知，直到第25次。微信支付则要求返回JSON格式的{"code": "SUCCESS"}。两者的去重逻辑完全相同，但响应格式不同。

关键设计决策

1. Key的有效期窗口

将有效期与网关的缓存窗口对齐。Stripe缓存24小时，支付宝和微信支付的out_trade_no有效期通常为90天，但你的服务端去重窗口设为24小时已足够覆盖重试场景。

expires_at TIMESTAMPTZ NOT NULL DEFAULT now() + INTERVAL '24 hours'

运行夜间清理任务防止表膨胀：

DELETE FROM idempotency_keys WHERE expires_at < now();

2. 失败时的策略：删除还是保留Key

错误类型	推荐策略	原因
网络超时 / PSP 5xx	删除Key	允许客户端用相同Key重试
余额不足（终端错误）	保留错误响应	防止不可恢复错误上的重试循环
无效参数 / 400	保留4xx响应	同上——重试没有意义

3. 并发请求处理

ON CONFLICT DO NOTHING确保同一时刻只有一个并发请求能获取锁。第二个请求看到locked_at IS NOT NULL, completed_at IS NULL后收到409，客户端应使用指数退避策略重试。

4. Key的生成职责

幂等键必须始终在客户端侧（前端或调用方微服务）生成——绝不能由服务端生成。若由服务端生成，超时后的重试每次都会产生新Key，完全绕过去重层。

对于支付宝和微信支付，out_trade_no实际上就是Order ID，应在创建Order记录时一并持久化，确保后续所有重试使用同一个值。

幂等键生命周期

Client                    Your API                 Payment Gateway (Alipay/WeChat/Stripe)
  |                           |                              |
  |-- POST /payments/charge   |                              |
  |   Idempotency-Key: abc123 |                              |
  |                           |-- INSERT idempotency_keys    |
  |                           |   (ON CONFLICT DO NOTHING)   |
  |                           |                              |
  |                           |-- POST charge                |
  |                           |   out_trade_no: ORDER-xxx -->|
  |                           |                              |-- 只处理一次
  |                           |<-- charge response           |
  |                           |                              |
  |                           |-- UPDATE completed_at        |
  |<-- 201 订单创建成功        |                              |
  |                           |                              |
  | [网络断开 -- 客户端重试]                                  |
  |                           |                              |
  |-- POST /payments/charge   |                              |
  |   Idempotency-Key: abc123 |                              |
  |                           |-- SELECT idempotency_keys    |
  |                           |   completed_at IS NOT NULL   |
  |<-- 201（缓存响应）         |                              |
  |   Idempotency-Replayed: true                             |

总结

层级	网关 / 工具	幂等性机制
PSP — 支付宝	`out_trade_no`请求参数	商户订单号去重，有效期90天
PSP — 微信支付	`out_trade_no`请求参数	商户订单号去重，有效期90天
PSP — Stripe	`Idempotency-Key` Header	24小时响应缓存
PSP — 2C2P	Payload中的`invoiceNo`	商户Invoice去重
自身API	FastAPI + PostgreSQL	`idempotency_keys`表 + `ON CONFLICT DO NOTHING`锁
Webhook	支付宝`out_trade_no` / 微信`transaction_id`	同一去重表的带前缀Key