books

《后端架构设计·架构基础与边界》7.3 幂等责任归属：API、领域、消息层的契约

把一次重复下单拆到 API 幂等键中间件、领域支付守卫和消息 Inbox 三层，明确每层的去重键、承诺与不负责的部分；给出可重放的 HTTP 中间件、按 ProviderRef 幂等的 MarkPaid、消费者去重实现，以及四组重复请求测试的真实输出。

Leeting Yan 6 分钟阅读 2595 字

重复从哪里来

一次“创建订单”可能被重复执行很多次：用户双击、客户端超时重试、网关重放、支付回调重复到达、Kafka 至少一次投递把同一条消息再送一遍。站内 幂等设计与消息可靠性 讲过幂等的通用手段，本节只回答组织问题：这些重复分别该由哪一层拦住，每层又明确不负责什么。真正的闭环不是多加一个 Redis key，而是把去重键、承诺和失败边界写进三份契约。

三层各管什么

层去重键拦住的重复明确不负责
APIIdempotency-Key + 请求指纹同一客户端请求的超时重试、网关重放绕过 API 的回调、消息重投
领域业务自然键 + 状态机同一业务事实从不同入口重复提交传输层重投的消息执行
消息消息 ID / inbox 记录broker 至少一次投递造成的重投内容相同但 ID 不同的新消息

API 层承诺“同一个 key 加同一个请求体，返回同一个响应”，它不理解订单状态；领域层承诺“同一笔支付凭证只生效一次”，它不关心请求从 HTTP 还是 webhook 进来；消息层承诺“同一条消息只执行一次副作用”，它不知道消息里的业务是否已经做过。三层键不同、失败模式不同，所以不能互相替代。

API 层：把同一请求变成同一响应

中间件先读 Idempotency-Key，再用 method + path + body 的 SHA-256 作为请求指纹。key 已存在且指纹不同，返回 409；key 存在但首个请求还在执行，同样返回 409，让客户端稍后重试；命中已完成记录，直接重放状态码和响应体，并标记 Idempotent-Replayed。5xx 不缓存：服务端错误不是确定的业务结果，下一次重试应该真正执行。内存实现没有 TTL 与淘汰策略，key 作用域应由网关或租户层拼接。

// file: internal/platform/idempotency/middleware.go
// tier: L1
// verify: go test -race -v ./internal/platform/idempotency/
package idempotency

import (
	"bytes"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"io"
	"net/http"
	"sync"
)

var ErrFingerprintMismatch = errors.New("idempotency key reused with different request")
var ErrInFlight = errors.New("idempotent request already in flight")

type Response struct {
	Status int
	Body   []byte
}

type record struct {
	fingerprint string
	response    *Response
}

type MemoryStore struct {
	mu      sync.Mutex
	records map[string]*record
}

func NewMemoryStore() *MemoryStore { return &MemoryStore{records: make(map[string]*record)} }

func (s *MemoryStore) Begin(key, fingerprint string) (*Response, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	rec, ok := s.records[key]
	if !ok {
		s.records[key] = &record{fingerprint: fingerprint}
		return nil, nil
	}
	if rec.fingerprint != fingerprint {
		return nil, ErrFingerprintMismatch
	}
	if rec.response == nil {
		return nil, ErrInFlight
	}
	cp := Response{Status: rec.response.Status, Body: append([]byte(nil), rec.response.Body...)}
	return &cp, nil
}

func (s *MemoryStore) Finish(key string, resp *Response) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if resp == nil {
		delete(s.records, key)
		return
	}
	s.records[key].response = &Response{Status: resp.Status, Body: append([]byte(nil), resp.Body...)}
}

func Middleware(store *MemoryStore, next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		key := r.Header.Get("Idempotency-Key")
		if key == "" {
			next.ServeHTTP(w, r)
			return
		}
		body, err := io.ReadAll(r.Body)
		if err != nil {
			http.Error(w, "read request body", http.StatusBadRequest)
			return
		}
		r.Body = io.NopCloser(bytes.NewReader(body))
		sum := sha256.Sum256([]byte(r.Method + "\n" + r.URL.Path + "\n" + string(body)))
		cached, err := store.Begin(key, hex.EncodeToString(sum[:]))
		switch {
		case errors.Is(err, ErrFingerprintMismatch), errors.Is(err, ErrInFlight):
			http.Error(w, err.Error(), http.StatusConflict)
			return
		case err != nil:
			http.Error(w, "idempotency store unavailable", http.StatusServiceUnavailable)
			return
		case cached != nil:
			w.Header().Set("Idempotent-Replayed", "true")
			w.WriteHeader(cached.Status)
			_, _ = w.Write(cached.Body)
			return
		}

		capture := &captureWriter{ResponseWriter: w}
		next.ServeHTTP(capture, r)
		status := capture.status
		if status == 0 {
			status = http.StatusOK
		}
		if status >= 500 {
			store.Finish(key, nil)
		} else {
			store.Finish(key, &Response{Status: status, Body: capture.body.Bytes()})
		}
	})
}

type captureWriter struct {
	http.ResponseWriter
	status int
	body   bytes.Buffer
}

func (c *captureWriter) WriteHeader(status int) {
	if c.status == 0 {
		c.status = status
		c.ResponseWriter.WriteHeader(status)
	}
}

func (c *captureWriter) Write(p []byte) (int, error) {
	if c.status == 0 {
		c.WriteHeader(http.StatusOK)
	}
	c.body.Write(p)
	return c.ResponseWriter.Write(p)
}

领域层：同一业务事实只生效一次

支付回调不一定经过创建订单的 API，也可能由支付网关直接重放。此时 API 幂等键帮不上忙，必须由领域层按业务自然键守卫。MarkPaid 只在 INVENTORY_RESERVED 状态接受支付；已经是 PAID 且 ProviderRef 相同，返回 alreadyPaid=true，不追加第二笔支付；同一个订单换了支付凭证，则视为状态冲突。

// file: internal/order/pay.go
// tier: L1
// verify: go test -race -v ./internal/platform/idempotency/
package order

import "errors"

var ErrInvalidState = errors.New("invalid state")

const (
	StatusInventoryReserved = "INVENTORY_RESERVED"
	StatusPaid              = "PAID"
)

type Payment struct {
	AmountCents int64
	ProviderRef string
}

type Order struct {
	ID       string
	Status   string
	Payments []Payment
}

// MarkPaid 对同一个 ProviderRef 幂等；状态不对或换了支付凭证才报错。
func (o *Order) MarkPaid(p Payment) (alreadyPaid bool, err error) {
	switch o.Status {
	case StatusPaid:
		if len(o.Payments) > 0 && o.Payments[0].ProviderRef == p.ProviderRef {
			return true, nil
		}
		return false, ErrInvalidState
	case StatusInventoryReserved:
		o.Payments = append(o.Payments, p)
		o.Status = StatusPaid
		return false, nil
	default:
		return false, ErrInvalidState
	}
}

消息层：至少一次投递下的消费去重

消息层面对的是 broker 的重投，去重键是消息 ID。Inbox.HandleOnce 先查已处理集合，已处理则返回 delivered=false；处理成功后才写入标记，处理失败不标记，让下一次重投可以重试。

// file: internal/messaging/inbox.go
// tier: L1
// verify: go test -race -v ./internal/platform/idempotency/
package messaging

import (
	"context"
	"sync"
)

// Inbox 是消息层的 L1 去重；生产实现应把效果与标记放进同一事务。
type Inbox struct {
	mu   sync.Mutex
	seen map[string]struct{}
}

func NewInbox() *Inbox { return &Inbox{seen: make(map[string]struct{})} }

func (i *Inbox) HandleOnce(ctx context.Context, messageID string, handle func(context.Context) error) (delivered bool, err error) {
	i.mu.Lock()
	_, done := i.seen[messageID]
	i.mu.Unlock()
	if done {
		return false, nil
	}
	if err := handle(ctx); err != nil {
		return false, err
	}
	i.mu.Lock()
	i.seen[messageID] = struct{}{}
	i.mu.Unlock()
	return true, nil
}

这个 L1 实现的检查和写入不是原子的，两个并发重投可能同时通过；生产实现要把“业务效果 + 去重标记”放进同一事务，或依赖 inbox 表唯一约束。消息 ID 只能证明“这条投递来过”，不能证明“这笔业务没做过”。

组合起来才闭环

三层组合的价值在于覆盖不同的重复路径。客户端重试先被 API 缓存拦住；如果缓存过期、客户端换了 key，领域状态机仍然会拒绝第二次支付；如果支付结果由消息异步回填，消息 Inbox 拦住 broker 重投。任何一层单独存在，都有一条绕过去的路径。

// file: internal/platform/idempotency/composition_test.go
// tier: L1
// verify: go test -race -v ./internal/platform/idempotency/
package idempotency_test

import (
	"context"
	"errors"
	"io"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync/atomic"
	"testing"

	"archlab/internal/messaging"
	"archlab/internal/order"
	"archlab/internal/platform/idempotency"
)

func do(h http.Handler, key, body string) *httptest.ResponseRecorder {
	req := httptest.NewRequest(http.MethodPost, "/api/v1/orders", strings.NewReader(body))
	req.Header.Set("Idempotency-Key", key)
	rec := httptest.NewRecorder()
	h.ServeHTTP(rec, req)
	return rec
}

func TestDuplicateRequestRunsHandlerOnce(t *testing.T) {
	var calls atomic.Int32
	next := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		calls.Add(1)
		w.WriteHeader(http.StatusCreated)
		_, _ = io.WriteString(w, `{"id":"o-1"}`)
	})
	h := idempotency.Middleware(idempotency.NewMemoryStore(), next)

	first, second := do(h, "key-1", `{"sku":"a"}`), do(h, "key-1", `{"sku":"a"}`)
	if calls.Load() != 1 || second.Code != http.StatusCreated || second.Body.String() != first.Body.String() {
		t.Fatalf("calls=%d code=%d body=%q", calls.Load(), second.Code, second.Body.String())
	}
	if second.Header().Get("Idempotent-Replayed") != "true" {
		t.Fatal("missing replay marker")
	}
	if conflict := do(h, "key-1", `{"sku":"b"}`); conflict.Code != http.StatusConflict || calls.Load() != 1 {
		t.Fatalf("conflict=%d calls=%d", conflict.Code, calls.Load())
	}
}

func TestServerErrorIsNotCached(t *testing.T) {
	var calls atomic.Int32
	h := idempotency.Middleware(idempotency.NewMemoryStore(), http.HandlerFunc(
		func(w http.ResponseWriter, _ *http.Request) {
			if calls.Add(1) == 1 {
				http.Error(w, "boom", http.StatusInternalServerError)
				return
			}
			w.WriteHeader(http.StatusCreated)
		}))

	if first := do(h, "key-1", `{}`); first.Code != http.StatusInternalServerError {
		t.Fatalf("first = %d", first.Code)
	}
	if second := do(h, "key-1", `{}`); second.Code != http.StatusCreated || calls.Load() != 2 {
		t.Fatalf("second=%d calls=%d", second.Code, calls.Load())
	}
}

func TestDomainGuardMakesPaymentIdempotent(t *testing.T) {
	o := &order.Order{ID: "o-1", Status: order.StatusInventoryReserved}
	p := order.Payment{AmountCents: 1000, ProviderRef: "ref-1"}
	if already, err := o.MarkPaid(p); err != nil || already {
		t.Fatalf("first already=%v err=%v", already, err)
	}
	if already, err := o.MarkPaid(p); err != nil || !already || len(o.Payments) != 1 {
		t.Fatalf("repeat already=%v err=%v payments=%d", already, err, len(o.Payments))
	}
	other := p
	other.ProviderRef = "ref-2"
	if _, err := o.MarkPaid(other); !errors.Is(err, order.ErrInvalidState) {
		t.Fatalf("different ref err = %v", err)
	}
}

func TestInboxSkipsRedeliveryAndRetriesFailures(t *testing.T) {
	inbox := messaging.NewInbox()
	var calls atomic.Int32
	handle := func(context.Context) error {
		if calls.Add(1) == 1 {
			return errors.New("handler failed")
		}
		return nil
	}
	if delivered, err := inbox.HandleOnce(context.Background(), "m-1", handle); delivered || err == nil {
		t.Fatalf("failed delivery delivered=%v err=%v", delivered, err)
	}
	if delivered, err := inbox.HandleOnce(context.Background(), "m-1", handle); !delivered || err != nil {
		t.Fatalf("retry delivered=%v err=%v", delivered, err)
	}
	if delivered, _ := inbox.HandleOnce(context.Background(), "m-1", handle); delivered || calls.Load() != 2 {
		t.Fatalf("redelivery delivered=%v calls=%d", delivered, calls.Load())
	}
}

真实运行结果

以下输出来自本机 macOS（darwin/arm64，Go 1.26.0），在独立模块 .local/checks/ch07-be-arch-base-07-3-idempotency-responsibility 中执行。gofmt -l . 无输出，go vet ./... 通过。

$ go test -race -v ./internal/platform/idempotency/
--- PASS: TestDuplicateRequestRunsHandlerOnce (0.00s)
--- PASS: TestServerErrorIsNotCached (0.00s)
--- PASS: TestDomainGuardMakesPaymentIdempotent (0.00s)
--- PASS: TestInboxSkipsRedeliveryAndRetriesFailures (0.00s)
ok  	archlab/internal/platform/idempotency	1.277s

常见错误与边界

第一，只用 API 幂等键，不做领域守卫。回调、消息和管理后台都可能绕过 API；key 一旦变化，重复支付就没有第二道防线。用请求 ID 做领域自然键同样无效：它标识“这一次调用”，不标识“这一笔业务”。

第二，消费者按消息内容判重。两条内容相同但 ID 不同的消息可能是合法两次业务，也可能是一次重投；内容哈希无法区分，必须用消息 ID 加业务唯一键组合判断。把 5xx 也缓存同样危险：瞬时故障会被永久重放。领域守卫本身还应校验金额，并依赖唯一约束兜底并发。

小结

幂等不是某一层的属性，而是三层契约的组合：API 层用 Idempotency-Key 加请求指纹保证“同一请求同一响应”，领域层用业务自然键和状态机保证“同一事实只生效一次”，消息层用消息 ID 和 inbox 保证“同一投递只执行一次”。三层的键不同，所以谁也不能替代谁；三层的失败模式不同，所以缺一层就会留下一条绕行路径；压测与故障注入留给卷二。下一章从边界转向迁移：当一个遗留系统需要被替换时，如何用绞杀者模式增量切流，而不是一次性重写。

阅读导航：上一节：7.2 事件、命令与消息的语义差异 · 下一节：8.1 绞杀者模式与增量迁移 。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「books」更多文章

  1. 《互联网产品的 AI 架构·平台化与治理》附录 C 平台化评审清单
  2. 《互联网产品的 AI 架构·平台化与治理》附录 B 能力目录与配额字段速查
  3. 《互联网产品的 AI 架构·平台化与治理》附录 A 治理角色与职责速查