用 Idempotency-Key 与数据库唯一约束实现订单接口幂等并缓存响应
用 Idempotency-Key 与数据库唯一约束实现订单接口幂等并缓存响应
问题场景
用户点击“提交订单”后，客户端超时重试、用户重复点击、网关重发，都可能让同一个订单请求到达服务端两次。只在客户端禁用按钮不可靠：请求可能已经到达服务端，只是响应丢失。若服务端不做幂等，就会创建两笔订单。
常见做法是让客户端为每次“逻辑提交”生成一个唯一的 Idempotency-Key 请求头。服务端第一次看到该键时执行下单，并把响应缓存下来；之后同一个键再次到达时，不再重复执行，而是直接返回首次响应。数据库层的唯一约束作为最后防线，防止幂等记录异常时仍然写入重复订单。
方案设计
服务端维护一张幂等表，主键为 Idempotency-Key。每条记录保存请求指纹、处理状态、首次响应的状态码与响应体。
请求进入后：
- 计算请求指纹，用于判断同一个键是否对应同一个请求体。
- 尝试插入幂等记录，状态为
in_progress。 - 插入成功，说明是首次请求：执行业务写入，缓存响应，把状态更新为
completed，提交事务。 - 插入冲突，说明键已存在：读取已有记录。指纹不一致返回 422；状态未完成返回 409；已完成则原样重放缓存的响应。
- 订单表的
idem_key唯一约束确保即使幂等逻辑被绕过，也不会产生重复订单。
示例把幂等记录插入、订单写入、响应缓存放在同一个数据库事务中。这样不会出现“订单成功但幂等记录失败”的中间状态；代价是同一个键的并发请求会短暂等待写锁。
最小可运行示例
使用 Flask + SQLite，仅用于演示思路。
import hashlib
import json
import sqlite3
import time
from flask import Flask, g, jsonify, request
app = Flask(__name__)
DB_PATH = "idempotency_demo.db"
def get_db():
if "db" not in g:
g.db = sqlite3.connect(DB_PATH, isolation_level=None)
g.db.row_factory = sqlite3.Row
g.db.execute("PRAGMA busy_timeout=5000")
return g.db
@app.teardown_appcontext
def close_db(_exc):
db = g.pop("db", None)
if db is not None:
db.close()
def init_db():
db = sqlite3.connect(DB_PATH)
db.executescript(
"""
CREATE TABLE IF NOT EXISTS idempotency_keys (
idem_key TEXT PRIMARY KEY,
fingerprint TEXT NOT NULL,
status TEXT NOT NULL CHECK (status IN ('in_progress', 'completed')),
response_status INTEGER,
response_body TEXT,
created_at INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS orders (
id INTEGER PRIMARY KEY AUTOINCREMENT,
idem_key TEXT NOT NULL UNIQUE,
user_id TEXT NOT NULL,
sku TEXT NOT NULL,
quantity INTEGER NOT NULL,
created_at INTEGER NOT NULL
);
"""
)
db.commit()
db.close()
def make_fingerprint(user_id, payload):
normalized = json.dumps(
{"user_id": user_id, "payload": payload},
sort_keys=True,
ensure_ascii=False,
separators=(",", ":"),
)
return hashlib.sha256(normalized.encode("utf-8")).hexdigest()
@app.route("/orders", methods=["POST"])
def create_order():
idem_key = request.headers.get("Idempotency-Key", "").strip()
if not idem_key or len(idem_key) > 128:
return jsonify(error="缺少或非法的 Idempotency-Key"), 400
payload = request.get_json(silent=True)
if not isinstance(payload, dict):
return jsonify(error="请求体必须是 JSON 对象"), 400
# 演示代码从请求体取用户。生产环境应从鉴权上下文获取。
user_id = str(payload.get("user_id", "anonymous"))
sku = str(payload.get("sku", ""))
try:
quantity = int(payload.get("quantity", 1))
except (TypeError, ValueError):
return jsonify(error="quantity 必须是整数"), 400
if not sku or quantity <= 0:
return jsonify(error="sku 和 quantity 不合法"), 400
fingerprint = make_fingerprint(user_id, payload)
db = get_db()
now = int(time.time())
try:
db.execute("BEGIN IMMEDIATE")
db.execute(
"INSERT INTO idempotency_keys(idem_key, fingerprint, status, created_at) "
"VALUES(?, ?, 'in_progress', ?)",
(idem_key, fingerprint, now),
)
except sqlite3.IntegrityError:
db.execute("ROLLBACK")
row = db.execute(
"SELECT fingerprint, status, response_status, response_body "
"FROM idempotency_keys WHERE idem_key = ?",
(idem_key,),
).fetchone()
if row is None:
return jsonify(error="幂等键状态异常，请重试"), 409
if row["fingerprint"] != fingerprint:
return jsonify(error="Idempotency-Key 已用于不同的请求体"), 422
if row["status"] != "completed":
return jsonify(error="相同请求正在处理中，请稍后重试"), 409
return app.response_class(
response=row["response_body"],
status=row["response_status"],
mimetype="application/json",
headers={"Idempotent-Replay": "true"},
)
except sqlite3.OperationalError:
return jsonify(error="数据库繁忙，请稍后重试"), 503
try:
cur = db.execute(
"INSERT INTO orders(idem_key, user_id, sku, quantity, created_at) "
"VALUES(?, ?, ?, ?, ?)",
(idem_key, user_id, sku, quantity, now),
)
body = json.dumps(
{"order_id": cur.lastrowid, "status": "created"},
ensure_ascii=False,
separators=(",", ":"),
)
db.execute(
"UPDATE idempotency_keys "
"SET status = 'completed', response_status = ?, response_body = ? "
"WHERE idem_key = ?",
(201, body, idem_key),
)
db.execute("COMMIT")
return app.response_class(response=body, status=201, mimetype="application/json")
except Exception:
db.execute("ROLLBACK")
raise
if __name__ == "__main__":
init_db()
app.run(host="127.0.0.1", port=5000, debug=True)
运行方法
安装 Flask 并启动：
python -m pip install flask
python app.py
第一次提交：
curl -i -X POST http://127.0.0.1:5000/orders -H "Content-Type: application/json" -H "Idempotency-Key: demo-key-0001" -d '{"user_id":"u_1001","sku":"SKU-A","quantity":2}'
会得到 201 和类似 {"order_id":1,"status":"created"} 的响应。重复执行同一条命令，会得到相同的响应体，并多出 Idempotent-Replay: true 头，数据库不会新增订单。
换一个请求体但复用同一个键：
curl -i -X POST http://127.0.0.1:5000/orders -H "Content-Type: application/json" -H "Idempotency-Key: demo-key-0001" -d '{"user_id":"u_1001","sku":"SKU-B","quantity":1}'
会返回 422，因为同一个键不能对应不同的请求内容。
边界与限制
幂等键的作用域：示例用 idem_key 作为全局主键，简单但不够安全。多租户系统应使用 UNIQUE(user_id, idem_key)，并在读取幂等记录时带上用户条件，避免跨用户猜测键读到别人的响应。用户标识必须来自鉴权上下文，不能来自请求体。
请求指纹：只应包含影响业务语义的字段。若把时间戳、随机数等无关字段纳入指纹，正常重试会被误判为不同请求。指纹不一致时返回 422，提示客户端同一键不能用于不同请求体。
并发与等待：单事务方案下，同一个键的并发请求会等待写锁。超过 busy_timeout 会返回 503。若业务处理时间较长，可以拆成两阶段：先提交 in_progress 记录，再执行业务，最后更新为 completed。两阶段需要额外处理进程崩溃后 in_progress 记录的过期清理或接管，复杂度更高。
响应缓存：缓存的状态码和响应体应设置保留期，定期清理，避免幂等表无限增长。不要缓存超大响应或包含敏感信息的响应。
数据库唯一约束是最后防线：即使幂等表因异常丢失记录，orders.idem_key UNIQUE 仍能阻止同一键写入第二笔订单。但要注意，示例中订单插入冲突会回滚并抛出异常，生产代码应捕获该情况并返回可预期的错误或查询已有订单。
幂等键不能替代其他机制：它不替代用户鉴权、库存扣减的原子性、支付回调验签、限流等。客户端应为每个新的逻辑请求生成新键，重试同一请求时复用旧键。
- 点赞
- 收藏
- 关注作者
评论（0）