一次请求，多次安全：深度解析幂等性 API 的设计与 Python 实战

一次请求，多次安全：深度解析幂等性 API 的设计与 Python 实战

“在分布式系统中，'请求是否已被处理’比’如何处理请求’更难回答。”

一、为什么幂等性是 API 设计的必修课？

2021 年，某头部电商平台在双十一期间出现了一个令人冷汗的 Bug：用户点击"立即付款"后，因网络抖动导致前端超时，自动重试机制触发了第二次请求，结果同一订单被扣款两次。这个 Bug 在上线后 47 分钟内造成了数千笔重复扣款，客服电话瞬间打爆。

根本原因：支付接口没有做幂等设计。

在微服务、分布式系统盛行的今天，网络不可靠、客户端超时重试、消息队列 at-least-once 投递……这些都是家常便饭。幂等性（Idempotency）不是可选项，而是健壮 API 的基础设施。

什么是幂等性？

数学定义：f(f(x)) = f(x)

API 语义：对同一请求执行一次与执行多次，系统状态和响应结果完全一致。

幂等操作： GET /orders/123 → 查询，多次相同
DELETE /orders/123 → 删除，删了还是删了
PUT /user/name=Alice → 设置，设置两次结果一样

非幂等操作：POST /orders → 创建，每次都新增一条
POST /wallet/deduct → 扣款，多扣一次就是 Bug

二、HTTP 方法的幂等性基础

先从协议层理解，RFC 7231 对各 HTTP 方法的幂等性有明确定义：

但协议层的幂等性只是"语义约定"，真正的业务幂等需要我们在代码层面主动实现。PUT 理论上幂等，但如果你的实现里有 update_count += 1，它就不幂等了。

三、幂等性的核心设计模式

模式一：幂等键（Idempotency Key）

这是最通用、最强大的幂等方案。客户端在请求头中携带一个唯一标识符，服务端以此作为去重依据。

流程图：
客户端生成 UUID → 携带在请求头 X-Idempotency-Key
↓
服务端检查该 Key 是否已处理
↓
已处理 → 返回缓存的原始响应
未处理 → 执行业务逻辑 → 存储结果 → 返回响应

Stripe、PayPal 等支付巨头都采用这种方案。下面用 Python + FastAPI 实现一个完整的幂等键中间件：

import uuid
import json
import hashlib
from datetime import timedelta
from functools import wraps
from typing import Optional, Callable

import redis
from fastapi import FastAPI, Request, Response, HTTPException
from fastapi.responses import JSONResponse

app = FastAPI()
redis_client = redis.Redis(host="localhost", port=6379, decode_responses=True)

# 幂等键的存储结构
IDEMPOTENCY_PREFIX = "idempotency:"
IDEMPOTENCY_TTL = timedelta(hours=24) # 24小时内的重复请求均返回缓存

class IdempotencyRecord:
"""幂等记录的数据结构"""

def __init__(self, status: str, response_body: dict = None,
status_code: int = 200):
self.status = status # "processing" | "completed"
self.response_body = response_body
self.status_code = status_code

def to_dict(self):
return {
"status": self.status,
"response_body": self.response_body,
"status_code": self.status_code
}

@classmethod
def from_dict(cls, data: dict):
return cls(**data)

def idempotent(ttl_seconds: int = 86400):
"""
幂等性装饰器
用法：在路由函数上添加 @idempotent()
"""
def decorator(func: Callable):
@wraps(func)
async def wrapper(request: Request, *args, **kwargs):
# 1. 提取幂等键
idempotency_key = request.headers.get("X-Idempotency-Key")

if not idempotency_key:
# 没有幂等键：直接执行（适合非关键操作）
return await func(request, *args, **kwargs)

# 2. 构造 Redis Key（加入用户标识，防止 Key 碰撞攻击）
user_id = request.headers.get("X-User-Id", "anonymous")
cache_key = f"{IDEMPOTENCY_PREFIX}{user_id}:{idempotency_key}"

# 3. 检查是否已有处理记录
cached = redis_client.get(cache_key)

if cached:
record = IdempotencyRecord.from_dict(json.loads(cached))

if record.status == "processing":
# 并发场景：另一个请求正在处理中
raise HTTPException(
status_code=409,
detail="请求正在处理中，请勿重复提交"
)

# 已完成：直接返回缓存结果
print(f"🔄 幂等命中：{idempotency_key}，返回缓存响应")
return JSONResponse(
content=record.response_body,
status_code=record.status_code,
headers={"X-Idempotency-Replayed": "true"}
)

# 4. 加锁标记"处理中"（防并发重复执行）
processing_record = IdempotencyRecord(status="processing")
redis_client.setex(
cache_key,
30, # 处理锁 30 秒超时，防死锁
json.dumps(processing_record.to_dict())
)

try:
# 5. 执行实际业务逻辑
response = await func(request, *args, **kwargs)

# 6. 提取响应内容并缓存
if isinstance(response, JSONResponse):
body = json.loads(response.body)
status_code = response.status_code
else:
body = response
status_code = 200

completed_record = IdempotencyRecord(
status="completed",
response_body=body,
status_code=status_code
)
redis_client.setex(
cache_key,
ttl_seconds,
json.dumps(completed_record.to_dict())
)

return response

except Exception as e:
# 7. 业务异常：删除处理锁，允许客户端重试
redis_client.delete(cache_key)
raise

return wrapper
return decorator

将这个装饰器应用到支付接口：

from pydantic import BaseModel
from decimal import Decimal

class PaymentRequest(BaseModel):
order_id: str
amount: Decimal
currency: str = "CNY"

@app.post("/payments")
@idempotent(ttl_seconds=86400)
async def create_payment(request: Request, body: PaymentRequest):
"""
创建支付订单

客户端调用示例：
POST /payments
X-Idempotency-Key: 550e8400-e29b-41d4-a716-446655440000
X-User-Id: user_123

{
"order_id": "ORD-2024-001",
"amount": "99.99",
"currency": "CNY"
}
"""
# 模拟支付处理逻辑
payment_result = await process_payment(
order_id=body.order_id,
amount=body.amount
)

return JSONResponse(content={
"payment_id": payment_result["id"],
"status": "success",
"amount": str(body.amount)
})

async def process_payment(order_id: str, amount: Decimal):
"""实际的支付处理逻辑"""
import asyncio
await asyncio.sleep(0.1) # 模拟处理延迟
return {"id": f"PAY-{uuid.uuid4().hex[:8].upper()}"}

模式二：数据库唯一约束兜底

幂等键方案依赖缓存层，如果 Redis 宕机，会有漏网之鱼。唯一约束作为最后一道防线，确保数据层绝对不重复：

from sqlalchemy import Column, String, Numeric, UniqueConstraint
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.exc import IntegrityError

Base = declarative_base()

class Payment(Base):
__tablename__ = "payments"

id = Column(String, primary_key=True)
order_id = Column(String, nullable=False)
idempotency_key = Column(String, nullable=False)
amount = Column(Numeric(10, 2))
status = Column(String, default="pending")

# 核心：唯一约束确保同一 idempotency_key 只能插入一次
__table_args__ = (
UniqueConstraint('idempotency_key', name='uq_payments_idem_key'),
UniqueConstraint('order_id', name='uq_payments_order_id'),
)

async def create_payment_with_db_guard(
db_session, order_id: str, amount: Decimal, idempotency_key: str
):
"""双重保障：缓存层 + 数据库唯一约束"""

# 先查询是否已存在
existing = db_session.query(Payment).filter_by(
idempotency_key=idempotency_key
).first()

if existing:
print(f"⚡ 数据库层幂等命中：{idempotency_key}")
return existing

try:
payment = Payment(
id=str(uuid.uuid4()),
order_id=order_id,
idempotency_key=idempotency_key,
amount=amount,
status="completed"
)
db_session.add(payment)
db_session.commit()
return payment

except IntegrityError:
# 并发场景下的唯一约束冲突
db_session.rollback()
# 重新查询并返回已存在的记录
return db_session.query(Payment).filter_by(
idempotency_key=idempotency_key
).first()

模式三：状态机防止重复操作

对于有明确状态流转的业务（订单、工单、审批流），状态机天然具备幂等性：

from enum import Enum
from dataclasses import dataclass, field
from typing import Dict, Set

class OrderStatus(Enum):
CREATED = "created"
PAID = "paid"
SHIPPED = "shipped"
COMPLETED = "completed"
CANCELLED = "cancelled"

# 合法的状态转换图
VALID_TRANSITIONS: Dict[OrderStatus, Set[OrderStatus]] = {
OrderStatus.CREATED: {OrderStatus.PAID, OrderStatus.CANCELLED},
OrderStatus.PAID: {OrderStatus.SHIPPED, OrderStatus.CANCELLED},
OrderStatus.SHIPPED: {OrderStatus.COMPLETED},
OrderStatus.COMPLETED: set(), # 终态，不可再转换
OrderStatus.CANCELLED: set(), # 终态
}

@dataclass
class Order:
id: str
status: OrderStatus = OrderStatus.CREATED

def transition_to(self, new_status: OrderStatus) –> bool:
"""
幂等的状态转换
– 目标状态 == 当前状态：幂等成功，返回 True
– 合法转换：执行转换，返回 True
– 非法转换：抛出异常
"""
if self.status == new_status:
# 重复操作，幂等处理
print(f"ℹ️ 订单 {self.id} 已处于 {new_status.value} 状态，忽略重复操作")
return True

if new_status not in VALID_TRANSITIONS[self.status]:
raise ValueError(
f"非法状态转换：{self.status.value} → {new_status.value}"
)

print(f"✅ 订单 {self.id}：{self.status.value} → {new_status.value}")
self.status = new_status
return True

# 测试幂等状态机
order = Order(id="ORD-001")

order.transition_to(OrderStatus.PAID) # ✅ 正常转换
order.transition_to(OrderStatus.PAID) # ℹ️ 幂等忽略，不报错
order.transition_to(OrderStatus.SHIPPED) # ✅ 正常转换

try:
order.transition_to(OrderStatus.CREATED) # ❌ 非法转换，抛异常
except ValueError as e:
print(f"拦截非法操作：{e}")

模式四：消息队列的幂等消费

消息队列的 at-least-once 投递保证消息不丢，但会产生重复消息。消费者必须实现幂等：

import hashlib
from typing import Dict, Any

class IdempotentMessageConsumer:
"""幂等消息消费者"""

def __init__(self, redis_client, ttl: int = 3600):
self.redis = redis_client
self.ttl = ttl
self.processed_prefix = "msg:processed:"

def _message_fingerprint(self, message: Dict[str, Any]) –> str:
"""生成消息指纹（基于业务关键字段，非消息ID）"""
# 使用业务幂等键，而非消息队列的 message_id
# 因为消息可能被不同的 message_id 重复投递
key_fields = {
"event_type": message.get("event_type"),
"business_id": message.get("business_id"),
"action": message.get("action"),
}
content = json.dumps(key_fields, sort_keys=True)
return hashlib.md5(content.encode()).hexdigest()

def process(self, message: Dict[str, Any], handler: Callable):
"""幂等处理消息"""
fingerprint = self._message_fingerprint(message)
cache_key = f"{self.processed_prefix}{fingerprint}"

# SET NX：原子操作，只有第一次设置成功
is_new = self.redis.set(cache_key, "1", ex=self.ttl, nx=True)

if not is_new:
print(f"🔁 重复消息，跳过处理：{fingerprint}")
return

try:
handler(message)
print(f"✅ 消息处理成功：{fingerprint}")
except Exception as e:
# 处理失败，删除标记，允许重试
self.redis.delete(cache_key)
raise

# 使用示例
consumer = IdempotentMessageConsumer(redis_client)

def handle_payment_event(msg):
print(f"处理支付事件：{msg['business_id']}")
# … 业务逻辑

# 即使同一消息被投递两次，handler 也只执行一次
consumer.process(
{"event_type": "payment_success", "business_id": "PAY-001", "action": "notify"},
handle_payment_event
)

四、客户端的正确重试姿势

幂等 API 设计好了，客户端也要配合正确使用：

import httpx
import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

class IdempotentHttpClient:
"""内置幂等重试的 HTTP 客户端"""

def __init__(self, base_url: str):
self.base_url = base_url
self.client = httpx.AsyncClient(base_url=base_url, timeout=10.0)

@retry(
# 指数退避：1s, 2s, 4s, 最多重试3次
wait=wait_exponential(multiplier=1, min=1, max=8),
stop=stop_after_attempt(3),
# 只对网络错误和 5xx 重试，不对 4xx 重试
retry=retry_if_exception_type((httpx.NetworkError, httpx.TimeoutException))
)
async def post_idempotent(self, path: str, body: dict,
idempotency_key: str = None,
user_id: str = None):
"""发送幂等 POST 请求"""

# 幂等键：由调用方生成并保存，重试时使用相同的 Key
if idempotency_key is None:
idempotency_key = str(uuid.uuid4())

headers = {
"X-Idempotency-Key": idempotency_key,
"Content-Type": "application/json"
}
if user_id:
headers["X-User-Id"] = user_id

response = await self.client.post(path, json=body, headers=headers)

# 检查是否是重放响应
if response.headers.get("X-Idempotency-Replayed") == "true":
print("📦 服务端返回幂等缓存结果")

# 对 5xx 错误主动触发重试
if response.status_code >= 500:
raise httpx.NetworkError(f"服务端错误：{response.status_code}")

return response

# 使用方式
async def main():
client = IdempotentHttpClient("http://api.example.com")

# 生成并保存幂等键（重试时复用）
idem_key = str(uuid.uuid4())

# 即使网络超时自动重试，支付也只会执行一次
response = await client.post_idempotent(
"/payments",
body={"order_id": "ORD-001", "amount": "99.99"},
idempotency_key=idem_key,
user_id="user_123"
)
print(f"支付结果：{response.json()}")

五、幂等设计的常见陷阱

陷阱一：幂等键范围太宽

错误做法：全局共享同一个幂等键命名空间，不同用户的 Key 可能碰撞。

正确做法：幂等键 = 用户 ID + 业务类型 + 客户端生成的 UUID，形成完整的命名空间隔离。

陷阱二：只做了请求去重，没做响应幂等

错误做法：第一次请求返回 {"payment_id": "PAY-001"}，重复请求返回 {"message": "already processed"}，客户端无法区分成功与失败。

正确做法：重复请求应返回与原始请求完全相同的响应体，客户端不需要任何特殊处理。

陷阱三：TTL 设置不合理

幂等键 TTL 需要覆盖客户端最长重试窗口。如果客户端可能在 1 小时内重试，TTL 就不能设为 30 分钟。通常建议 24 小时，支付场景甚至建议 7 天。

陷阱四：忽略并发竞争

两个相同 Key 的请求同时到达，都通过了"是否已处理"的检查，然后都开始执行业务。解决方案：使用 Redis SET NX（原子操作）加锁，确保同一时刻只有一个请求在处理。

六、完整测试用例

import pytest
from httpx import AsyncClient
import asyncio

@pytest.mark.asyncio
async def test_payment_idempotency():
"""测试支付接口的幂等性"""

async with AsyncClient(app=app, base_url="http://test") as client:
idempotency_key = str(uuid.uuid4())
headers = {
"X-Idempotency-Key": idempotency_key,
"X-User-Id": "test_user"
}
payload = {"order_id": "ORD-TEST-001", "amount": "100.00"}

# 第一次请求
resp1 = await client.post("/payments", json=payload, headers=headers)
assert resp1.status_code == 200
payment_id_1 = resp1.json()["payment_id"]

# 第二次请求（模拟重试）
resp2 = await client.post("/payments", json=payload, headers=headers)
assert resp2.status_code == 200
payment_id_2 = resp2.json()["payment_id"]

# 核心断言：两次响应完全一致
assert payment_id_1 == payment_id_2
assert resp2.headers.get("X-Idempotency-Replayed") == "true"

print(f"✅ 幂等测试通过：两次请求返回相同 payment_id={payment_id_1}")

@pytest.mark.asyncio
async def test_concurrent_idempotency():
"""测试并发场景下的幂等性"""

async with AsyncClient(app=app, base_url="http://test") as client:
idempotency_key = str(uuid.uuid4())
headers = {"X-Idempotency-Key": idempotency_key, "X-User-Id": "user_concurrent"}
payload = {"order_id": "ORD-CONCURRENT-001", "amount": "200.00"}

# 模拟 5 个并发请求
tasks = [
client.post("/payments", json=payload, headers=headers)
for _ in range(5)
]
responses = await asyncio.gather(*tasks, return_exceptions=True)

# 过滤成功响应
success_responses = [r for r in responses if hasattr(r, 'status_code')
and r.status_code == 200]

# 所有成功响应的 payment_id 必须一致
payment_ids = {r.json()["payment_id"] for r in success_responses}
assert len(payment_ids) == 1, f"发现多个不同的 payment_id：{payment_ids}"
print(f"✅ 并发幂等测试通过：{len(success_responses)} 个响应均返回相同结果")

七、总结与选型指南

幂等性设计没有银弹，需要根据场景组合使用多种模式：

场景 推荐方案
─────────────────────────────────────────────
支付/转账 幂等键（Redis）+ 数据库唯一约束
订单状态流转 状态机
消息队列消费 消息指纹去重
普通查询/更新 HTTP 方法语义（GET/PUT）
批量导入 业务主键唯一约束

核心三原则记住就够了：

① 客户端负责生成并保存幂等键，重试时携带相同的键

② 服务端用原子操作（Redis SET NX）加锁，缓存完整响应

③ 数据库唯一约束作为最后防线，不依赖单一缓存层

幂等性的本质，是让系统对"不确定的网络世界"给出"确定的业务答案"。把这个问题解决好，你的 API 才算真正成熟可靠。

互动讨论： 你的系统中遇到过因非幂等设计导致的数据问题吗？你是如何发现并修复的？在高并发场景下，幂等键的存储和清理策略有哪些值得分享的实践经验？欢迎在评论区交流！

参考资料：

• Stripe 幂等性设计文档
• FastAPI 官方文档
• Tenacity 重试库
• Redis SET NX 文档
• 《数据密集型应用系统设计》第 9 章——一致性与共识