Go API实战：如何实现一个生产级的接口幂等性

2023 年我加入了一家快速增长的电商平台负责支付系统的稳定性。我们的业务正在高速扩张每天处理数万笔交易。一切看起来都在正轨上——直到一个周六的凌晨。当时我们接入了三家支付渠道其中一家东南亚本地钱包的 API 响应特别不稳定。用户在 App 内完成支付后我们的服务向渠道发送扣款请求。渠道扣款成功但由于网络超时响应没有及时返回。移动端等待几秒后没有收到确认自动触发了重试机制。结果就是一笔订单三次扣款四封客服投诉邮件。那次事故让我在凌晨四点和渠道方的值班工程师对账花了整整一个周末梳理日志。从那时起我真正理解了幂等性——不是作为一个理论概念而是作为一种在分布式系统中保障数据一致性的最后防线。幂等性在实践中的含义如果一个操作执行多次与执行一次产生的结果相同那么它就是幂等的。GET /users/123天然是幂等的。但POST /payments不是——除非你主动设计成那样。实现方式很简单客户端为每个逻辑操作生成一个唯一标识并在每次请求包括重试中携带它。服务端利用这个标识来判断“这个请求我是否已经处理过”如果处理过则直接返回之前的结果而不是再次执行业务逻辑。客户端 服务端 支付渠道 | | | |-- 支付请求 | | | Idempotency-Key: abc-123 -- | | | |-- 检查幂等存储 | | | (首次未命中) | | |-- 发起扣款 ----------------------| | |-- 扣款成功 ----------------------| | |-- 缓存处理结果 | |-- 返回成功响应 --------------------| | | | | | [网络超时客户端自动重试] | | | | | |-- 重试请求 | | | Idempotency-Key: abc-123 -- | | | |-- 检查幂等存储 | | | (命中直接返回缓存) | |-- 返回相同的成功响应 --------------| |支付渠道永远不会被重复调用。用户看到的是同样的成功结果。系统状态保持一致。核心数据结构在开始写代码之前我们需要定义幂等记录的存储结构packageidempotencyimport(contexttime)// IdempotencyRecord 存储幂等键及其对应的响应typeIdempotencyRecordstruct{Keystringjson:keyStatusCodeintjson:status_codeHeadersmap[string]stringjson:headersBody[]bytejson:bodyCreatedAt time.Timejson:created_atExpiresAt time.Timejson:expires_atRequestHashstringjson:request_hash// 用于检测同一Key下请求体是否变化InFlightbooljson:in_flight// 标记请求正在处理中防止并发问题}// IdempotencyStore 定义存储接口方便切换实现typeIdempotencyStoreinterface{Get(ctx context.Context,keystring)(*IdempotencyRecord,error)SetInFlight(ctx context.Context,keystring,requestHashstring,ttl time.Duration)(bool,error)Finalize(ctx context.Context,record*IdempotencyRecord)errorDelete(ctx context.Context,keystring)error}InFlight字段是防止并发重复的关键。如果没有它两个使用相同键的并发请求可能同时发现缓存未命中然后同时执行业务逻辑导致重复操作。方案一基于 Redis 的实现在分布式系统中Redis 是首选方案。它通过 Lua 脚本提供原子操作并内置 TTL 支持packageidempotencyimport(contextcrypto/sha256encoding/jsonfmttimegithub.com/redis/go-redis/v9)const(keyPrefixidemp:inFlightTTL30*time.Second defaultTTL24*time.Hour)typeRedisStorestruct{client*redis.Client}func(s*RedisStore)redisKey(keystring)string{returnkeyPrefixkey}// SetInFlight 使用 Lua 脚本实现原子性的检查并设置// 只有当键不存在时才会设置返回 true 表示成功抢占varsetInFlightScriptredis.NewScript( local key KEYS[1] local value ARGV[1] local ttl tonumber(ARGV[2]) local result redis.call(SET, key, value, NX, EX, ttl) if result then return 1 else return 0 end )func(s*RedisStore)SetInFlight(ctx context.Context,keystring,requestHashstring,ttl time.Duration)(bool,error){record:IdempotencyRecord{Key:key,InFlight:true,RequestHash:requestHash,CreatedAt:time.Now(),ExpiresAt:time.Now().Add(ttl),}data,_:json.Marshal(record)result,err:setInFlightScript.Run(ctx,s.client,[]string{s.redisKey(key)},string(data),int(ttl.Seconds()),).Int()iferr!nil{returnfalse,fmt.Errorf(set in-flight failed: %w,err)}returnresult1,nil}func(s*RedisStore)Finalize(ctx context.Context,record*IdempotencyRecord)error{record.InFlightfalsedata,_:json.Marshal(record)ttl:time.Until(record.ExpiresAt)ifttl0{ttldefaultTTL}returns.client.Set(ctx,s.redisKey(record.Key),data,ttl).Err()}func(s*RedisStore)Get(ctx context.Context,keystring)(*IdempotencyRecord,error){data,err:s.client.Get(ctx,s.redisKey(key)).Bytes()iferrredis.Nil{returnnil,nil}iferr!nil{returnnil,err}varrecord IdempotencyRecordiferr:json.Unmarshal(data,record);err!nil{returnnil,err}returnrecord,nil}func(s*RedisStore)Delete(ctx context.Context,keystring)error{returns.client.Del(ctx,s.redisKey(key)).Err()}// HashRequest 生成请求体的哈希值用于检测同一Key下的请求是否一致funcHashRequest(body[]byte)string{h:sha256.Sum256(body)returnfmt.Sprintf(%x,h[:8])}一个常见的误区不设置 TTL 就存储所有响应。这样 Redis 会被历史数据撑爆。务必设置过期时间。对于支付24 小时足够对于订单创建可能需要 7 天。方案二基于 PostgreSQL 的实现当 Redis 不可用或者你需要幂等记录与业务操作在同一个 ACID 事务中时PostgreSQL 是更好的选择packageidempotencyimport(contextdatabase/sqlencoding/jsonfmttime_github.com/lib/pq)// 表结构// CREATE TABLE idempotency_records (// key TEXT PRIMARY KEY,// request_hash TEXT NOT NULL,// status_code INTEGER,// headers JSONB,// body BYTEA,// in_flight BOOLEAN NOT NULL DEFAULT TRUE,// created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),// expires_at TIMESTAMPTZ NOT NULL// );// CREATE INDEX idx_idempotency_expires ON idempotency_records(expires_at);typePostgresStorestruct{db*sql.DB}func(s*PostgresStore)SetInFlight(ctx context.Context,keystring,requestHashstring,ttl time.Duration)(bool,error){result,err:s.db.ExecContext(ctx, INSERT INTO idempotency_records (key, request_hash, in_flight, expires_at) VALUES ($1, $2, TRUE, $3) ON CONFLICT (key) DO NOTHING ,key,requestHash,time.Now().Add(ttl))iferr!nil{returnfalse,err}rows,_:result.RowsAffected()returnrows1,nil}func(s*PostgresStore)Finalize(ctx context.Context,record*IdempotencyRecord)error{headersJSON,_:json.Marshal(record.Headers)_,err:s.db.ExecContext(ctx, UPDATE idempotency_records SET status_code $2, headers $3, body $4, in_flight FALSE, expires_at $5 WHERE key $1 ,record.Key,record.StatusCode,headersJSON,record.Body,record.ExpiresAt)returnerr}func(s*PostgresStore)Get(ctx context.Context,keystring)(*IdempotencyRecord,error){row:s.db.QueryRowContext(ctx, SELECT key, request_hash, status_code, headers, body, in_flight, created_at, expires_at FROM idempotency_records WHERE key $1 AND expires_at NOW() ,key)// ... 扫描逻辑returnrecord,nil}func(s*PostgresStore)Delete(ctx context.Context,keystring)error{_,err:s.db.ExecContext(ctx,DELETE FROM idempotency_records WHERE key $1,key)returnerr}使用 PostgreSQL 的优势在于你可以将SetInFlight和业务逻辑放在同一个数据库事务中。如果业务操作失败并回滚幂等记录也会一并回滚不会产生脏数据。HTTP 中间件实现下面是将幂等逻辑封装为 HTTP 中间件的完整实现packageidempotencyimport(bytesionet/httpstringstime)constHeaderIdempotencyKeyIdempotency-KeytypeMiddlewarestruct{store IdempotencyStore ttl time.Duration}typeresponseCapturestruct{http.ResponseWriter statusCodeintbody bytes.Buffer headers http.Header}func(m*Middleware)Handler(next http.Handler)http.Handler{returnhttp.HandlerFunc(func(w http.ResponseWriter,r*http.Request){key:strings.TrimSpace(r.Header.Get(HeaderIdempotencyKey))ifkey{next.ServeHTTP(w,r)return}ctx:r.Context()bodyBytes,_:io.ReadAll(r.Body)r.Bodyio.NopCloser(bytes.NewReader(bodyBytes))requestHash:HashRequest(bodyBytes)// 检查是否已存在existing,err:m.store.Get(ctx,key)iferrnilexisting!nil{// 同一Key下请求体不同 → 客户端使用错误ifexisting.RequestHash!requestHash{http.Error(w,idempotency key reused with different request,http.StatusUnprocessableEntity)return}// 有其他请求正在处理 → 返回冲突引导重试ifexisting.InFlight{w.Header().Set(Retry-After,1)http.Error(w,request in progress,http.StatusConflict)return}// 缓存命中重放响应fork,v:rangeexisting.Headers{w.Header().Set(k,v)}w.Header().Set(X-Idempotent-Replayed,true)w.WriteHeader(existing.StatusCode)w.Write(existing.Body)return}// 首次请求尝试抢占处理权claimed,err:m.store.SetInFlight(ctx,key,requestHash,m.ttl)iferr!nil||!claimed{w.Header().Set(Retry-After,1)http.Error(w,request in progress,http.StatusConflict)return}// 执行业务逻辑capture:responseCapture{ResponseWriter:w,headers:make(http.Header)}next.ServeHTTP(capture,r)// 5xx 错误不缓存应该让客户端重试ifcapture.statusCode500{m.store.Delete(ctx,key)return}// 缓存成功响应record:IdempotencyRecord{Key:key,RequestHash:requestHash,StatusCode:capture.statusCode,Headers:capture.headers,Body:capture.body.Bytes(),CreatedAt:time.Now(),ExpiresAt:time.Now().Add(m.ttl),}m.store.Finalize(ctx,record)})}statusCode 500的判断是关键——如果支付渠道返回 503我们绝不能缓存这个错误否则用户重试时只会拿到错误响应无法恢复正常。集成示例packagemainimport(encoding/jsonlognet/httptimegithub.com/redis/go-redis/v9yourapp/idempotency)funchandlePayment(w http.ResponseWriter,r*http.Request){varreqstruct{Amountintjson:amountCurrencystringjson:currency}json.NewDecoder(r.Body).Decode(req)// 实际支付逻辑...resp:map[string]interface{}{transaction_id:txn_generateID(),status:success,}json.NewEncoder(w).Encode(resp)}funcmain(){rdb:redis.NewClient(redis.Options{Addr:localhost:6379})store:idempotency.NewRedisStore(rdb)middleware:idempotency.NewMiddleware(store,24*time.Hour)mux:http.NewServeMux()mux.Handle(/payments,middleware.Handler(http.HandlerFunc(handlePayment)))log.Fatal(http.ListenAndServe(:8080,mux))}不同场景的 TTL 建议端点类型建议 TTL理由支付24小时用户支付失败后通常立即重试订单提交7天用户可能几天后回来确认状态表单提交1小时短期防重复足够异步任务永久配合清理任务任务可能长时间未完成总结那次重复扣款的事故让我明白了一个道理在分布式系统中幂等性不是锦上添花而是基础设施的一部分。当你部署第二个服务实例时就会立即面临“两个 Pod 同时处理同一请求怎么办”的问题。幂等性正是解决这个问题的标准方案。这里的模式——Lua 脚本实现原子操作、PostgreSQL 的INSERT ... ON CONFLICT DO NOTHING、InFlight状态机——都不是什么高深技巧。它们是经过验证的、可预测的解决方案能让你在凌晨两点安心入睡。不要等到事故发生了才去实现它。