【系统架构设计百科】幂等性设计：分布式环境下的安全重试

// Go：DELETE 处理示例
func handleDelete(w http.ResponseWriter, r *http.Request) {
    id := extractID(r)
    
    result, err := db.Exec("DELETE FROM orders WHERE id = ?", id)
    if err != nil {
        http.Error(w, "internal error", http.StatusInternalServerError)
        return
    }
    
    rowsAffected, _ := result.RowsAffected()
    if rowsAffected == 0 {
        // 资源已不存在，但操作仍是幂等的
        http.Error(w, "not found", http.StatusNotFound)
        return
    }
    
    w.WriteHeader(http.StatusNoContent)
}

// 幂等的 PATCH：设置绝对值
{ "status": "confirmed" }

// 非幂等的 PATCH：增量操作
{ "op": "increment", "path": "/balance", "value": 100 }

// Go：幂等键生成策略

// 方式一：UUID v4，全局唯一
import "github.com/google/uuid"
key := uuid.New().String()
// 例如：550e8400-e29b-41d4-a716-446655440000

// 方式二：业务语义键，可读性强且天然去重
key := fmt.Sprintf("order_%s_pay_%s", orderID, userID)
// 例如：order_20260413001_pay_user123

// 方式三：请求内容哈希，自动去重
import "crypto/sha256"
body := []byte(`{"amount":2000,"currency":"usd"}`)
hash := sha256.Sum256(body)
key := fmt.Sprintf("%x", hash)

// Java：参数一致性校验
public class IdempotencyValidator {

    public void validate(
            IdempotencyRecord record,
            String requestBody) throws IdempotencyConflictException {
        
        String storedHash = record.getRequestBodyHash();
        String currentHash = sha256(requestBody);
        
        if (!storedHash.equals(currentHash)) {
            throw new IdempotencyConflictException(
                "Idempotency key already used with different parameters. "
                + "Key: " + record.getIdempotencyKey()
            );
        }
    }
    
    private String sha256(String input) {
        try {
            MessageDigest digest = MessageDigest.getInstance("SHA-256");
            byte[] hash = digest.digest(input.getBytes(StandardCharsets.UTF_8));
            return bytesToHex(hash);
        } catch (NoSuchAlgorithmException e) {
            throw new RuntimeException(e);
        }
    }
    
    private String bytesToHex(byte[] bytes) {
        StringBuilder sb = new StringBuilder();
        for (byte b : bytes) {
            sb.append(String.format("%02x", b));
        }
        return sb.toString();
    }
}

CREATE TABLE idempotency_keys (
    id              BIGSERIAL PRIMARY KEY,
    key             VARCHAR(255) NOT NULL UNIQUE,
    request_method  VARCHAR(10) NOT NULL,
    request_path    VARCHAR(512) NOT NULL,
    request_hash    VARCHAR(64) NOT NULL,
    response_code   INT,
    response_body   TEXT,
    status          VARCHAR(20) NOT NULL DEFAULT 'processing',
    created_at      TIMESTAMP NOT NULL DEFAULT NOW(),
    expires_at      TIMESTAMP NOT NULL DEFAULT NOW() + INTERVAL '24 hours',
    
    CONSTRAINT chk_status CHECK (status IN ('processing', 'completed', 'failed'))
);

CREATE INDEX idx_idempotency_key ON idempotency_keys (key);
CREATE INDEX idx_idempotency_expires ON idempotency_keys (expires_at);

-- PostgreSQL：支付记录表，带幂等键唯一约束
CREATE TABLE payments (
    id              BIGSERIAL PRIMARY KEY,
    idempotency_key VARCHAR(255) NOT NULL,
    order_id        VARCHAR(64) NOT NULL,
    user_id         VARCHAR(64) NOT NULL,
    amount          DECIMAL(12, 2) NOT NULL,
    currency        VARCHAR(3) NOT NULL DEFAULT 'CNY',
    status          VARCHAR(20) NOT NULL DEFAULT 'pending',
    channel         VARCHAR(32),
    channel_tx_id   VARCHAR(128),
    created_at      TIMESTAMP NOT NULL DEFAULT NOW(),
    updated_at      TIMESTAMP NOT NULL DEFAULT NOW(),
    
    CONSTRAINT uk_idempotency_key UNIQUE (idempotency_key)
);

// Go：基于数据库唯一约束的幂等支付处理
package payment

import (
    "context"
    "database/sql"
    "fmt"
)

type PaymentService struct {
    db *sql.DB
}

func (s *PaymentService) CreatePayment(
    ctx context.Context,
    idempotencyKey, orderID, userID, currency string,
    amount float64,
) (*Payment, error) {

    query := `
        INSERT INTO payments (idempotency_key, order_id, user_id, amount, currency, status)
        VALUES ($1, $2, $3, $4, $5, 'pending')
        ON CONFLICT (idempotency_key) DO NOTHING
        RETURNING id, created_at
    `

    var payment Payment
    err := s.db.QueryRowContext(ctx, query,
        idempotencyKey, orderID, userID, amount, currency,
    ).Scan(&payment.ID, &payment.CreatedAt)

    if err == sql.ErrNoRows {
        // ON CONFLICT 命中，返回已有记录
        return s.getByIdempotencyKey(ctx, idempotencyKey)
    }
    if err != nil {
        return nil, fmt.Errorf("insert payment: %w", err)
    }

    // 新建记录，执行实际支付逻辑
    if err := s.processPayment(ctx, &payment); err != nil {
        return nil, fmt.Errorf("process payment: %w", err)
    }
    return &payment, nil
}

func (s *PaymentService) getByIdempotencyKey(
    ctx context.Context, key string,
) (*Payment, error) {
    var p Payment
    err := s.db.QueryRowContext(ctx,
        `SELECT id, idempotency_key, order_id, user_id, amount,
                currency, status, created_at
         FROM payments WHERE idempotency_key = $1`, key,
    ).Scan(&p.ID, &p.IdempotencyKey, &p.OrderID, &p.UserID,
        &p.Amount, &p.Currency, &p.Status, &p.CreatedAt)
    if err != nil {
        return nil, fmt.Errorf("query payment: %w", err)
    }
    return &p, nil
}

func (s *PaymentService) processPayment(
    ctx context.Context, payment *Payment,
) error {
    _, err := s.db.ExecContext(ctx,
        "UPDATE payments SET status = 'completed', updated_at = NOW() WHERE id = $1",
        payment.ID)
    return err
}

-- MySQL：等效写法
INSERT INTO payments (idempotency_key, order_id, user_id, amount, currency, status)
VALUES (?, ?, ?, ?, ?, 'pending')
ON DUPLICATE KEY UPDATE id = id;

-- 通过 ROW_COUNT() 判断是否为新插入
-- ROW_COUNT() = 1 表示新插入
-- ROW_COUNT() = 0 表示已存在（ON DUPLICATE KEY 未实际更新）

// Go：基于 Redis SETNX 的幂等性保障
package idempotency

import (
    "context"
    "encoding/json"
    "errors"
    "fmt"
    "time"

    "github.com/redis/go-redis/v9"
)

var (
    ErrRequestInProgress = errors.New("request is being processed")
    ErrDuplicateRequest  = errors.New("duplicate request")
)

const (
    keyPrefix     = "idempotency:"
    lockTTL       = 30 * time.Second  // 处理中锁定时长
    responseTTL   = 24 * time.Hour    // 响应缓存时长
)

type CachedResponse struct {
    StatusCode int    `json:"status_code"`
    Body       string `json:"body"`
    CreatedAt  int64  `json:"created_at"`
}

type IdempotencyGuard struct {
    rdb *redis.Client
}

func NewIdempotencyGuard(rdb *redis.Client) *IdempotencyGuard {
    return &IdempotencyGuard{rdb: rdb}
}

// TryAcquire 尝试获取幂等锁
// 返回值：
//   cached  - 如果请求已处理过，返回缓存的响应
//   acquired - 是否成功获取锁
//   err     - 错误信息
func (g *IdempotencyGuard) TryAcquire(
    ctx context.Context,
    idempotencyKey string,
) (cached *CachedResponse, acquired bool, err error) {

    lockKey := keyPrefix + "lock:" + idempotencyKey
    resultKey := keyPrefix + "result:" + idempotencyKey

    // 先检查是否已有处理结果
    resultData, err := g.rdb.Get(ctx, resultKey).Bytes()
    if err == nil {
        var resp CachedResponse
        if jsonErr := json.Unmarshal(resultData, &resp); jsonErr == nil {
            return &resp, false, nil
        }
    }
    if err != nil && !errors.Is(err, redis.Nil) {
        return nil, false, fmt.Errorf("check result: %w", err)
    }

    // 尝试加锁
    ok, err := g.rdb.SetNX(ctx, lockKey, "processing", lockTTL).Result()
    if err != nil {
        return nil, false, fmt.Errorf("acquire lock: %w", err)
    }
    if !ok {
        return nil, false, ErrRequestInProgress
    }

    return nil, true, nil
}

// Complete 标记请求已完成，缓存响应
func (g *IdempotencyGuard) Complete(
    ctx context.Context,
    idempotencyKey string,
    statusCode int,
    body string,
) error {

    lockKey := keyPrefix + "lock:" + idempotencyKey
    resultKey := keyPrefix + "result:" + idempotencyKey

    resp := CachedResponse{
        StatusCode: statusCode,
        Body:       body,
        CreatedAt:  time.Now().Unix(),
    }

    data, err := json.Marshal(resp)
    if err != nil {
        return fmt.Errorf("marshal response: %w", err)
    }

    // 使用 Pipeline 原子地存储结果并释放锁
    pipe := g.rdb.Pipeline()
    pipe.Set(ctx, resultKey, data, responseTTL)
    pipe.Del(ctx, lockKey)
    _, err = pipe.Exec(ctx)
    if err != nil {
        return fmt.Errorf("save result: %w", err)
    }

    return nil
}

// Rollback 业务失败时释放锁，允许客户端重试
func (g *IdempotencyGuard) Rollback(
    ctx context.Context,
    idempotencyKey string,
) error {

    lockKey := keyPrefix + "lock:" + idempotencyKey
    return g.rdb.Del(ctx, lockKey).Err()
}

// Go：看门狗锁续期
func (g *IdempotencyGuard) startWatchdog(
    ctx context.Context,
    lockKey string,
    interval time.Duration,
) context.CancelFunc {

    watchCtx, cancel := context.WithCancel(ctx)

    go func() {
        ticker := time.NewTicker(interval)
        defer ticker.Stop()

        for {
            select {
            case <-watchCtx.Done():
                return
            case <-ticker.C:
                // 每隔 interval 续期一次
                g.rdb.Expire(watchCtx, lockKey, lockTTL)
            }
        }
    }()

    return cancel
}

-- 消息去重表
CREATE TABLE message_dedup (
    message_id   VARCHAR(128) NOT NULL,
    topic        VARCHAR(64) NOT NULL,
    consumer     VARCHAR(64) NOT NULL,
    status       VARCHAR(20) NOT NULL DEFAULT 'processing',
    processed_at TIMESTAMP,
    created_at   TIMESTAMP NOT NULL DEFAULT NOW(),
    
    PRIMARY KEY (message_id, topic, consumer)
);

-- 定期清理已处理的历史记录
CREATE INDEX idx_dedup_created ON message_dedup (created_at);

// Java：消息去重消费者
import org.springframework.transaction.annotation.Transactional;

public class IdempotentConsumer {

    private final JdbcTemplate jdbcTemplate;
    private final OrderService orderService;

    public IdempotentConsumer(
            JdbcTemplate jdbcTemplate,
            OrderService orderService) {
        this.jdbcTemplate = jdbcTemplate;
        this.orderService = orderService;
    }

    @Transactional
    public void consume(Message message) {
        String messageId = message.getMessageId();
        String topic = message.getTopic();
        String consumer = "order-service";

        // 尝试插入去重记录
        int inserted = jdbcTemplate.update(
            "INSERT INTO message_dedup (message_id, topic, consumer, status) "
            + "VALUES (?, ?, ?, 'processing') "
            + "ON CONFLICT (message_id, topic, consumer) DO NOTHING",
            messageId, topic, consumer
        );

        if (inserted == 0) {
            // 消息已被处理或正在处理
            log.info("重复消息，跳过处理: messageId={}", messageId);
            return;
        }

        try {
            // 执行业务逻辑
            orderService.processOrder(message.getBody());

            // 更新去重记录状态
            jdbcTemplate.update(
                "UPDATE message_dedup SET status = 'completed', "
                + "processed_at = NOW() "
                + "WHERE message_id = ? AND topic = ? AND consumer = ?",
                messageId, topic, consumer
            );
        } catch (Exception e) {
            // 业务处理失败，删除去重记录，允许重试
            jdbcTemplate.update(
                "DELETE FROM message_dedup "
                + "WHERE message_id = ? AND topic = ? AND consumer = ?",
                messageId, topic, consumer
            );
            throw e;
        }
    }
}

-- 每天凌晨执行，清理 7 天前的已完成记录
DELETE FROM message_dedup
WHERE status = 'completed'
  AND created_at < NOW() - INTERVAL '7 days';

-- 清理超时的 processing 状态记录（可能是僵尸记录）
UPDATE message_dedup
SET status = 'timeout'
WHERE status = 'processing'
  AND created_at < NOW() - INTERVAL '1 hour';

CREATE TABLE payment_orders (
    id                  BIGSERIAL PRIMARY KEY,
    payment_no          VARCHAR(32) NOT NULL UNIQUE,
    merchant_id         VARCHAR(32) NOT NULL,
    merchant_order_no   VARCHAR(64) NOT NULL,
    amount              DECIMAL(12, 2) NOT NULL,
    currency            VARCHAR(3) NOT NULL,
    status              VARCHAR(20) NOT NULL,
    channel             VARCHAR(32),
    channel_trade_no    VARCHAR(128),
    created_at          TIMESTAMP NOT NULL DEFAULT NOW(),
    updated_at          TIMESTAMP NOT NULL DEFAULT NOW(),
    
    CONSTRAINT uk_merchant_order
        UNIQUE (merchant_id, merchant_order_no)
);

// Java：账务系统幂等记账
@Service
public class AccountService {

    private final AccountRepository accountRepo;
    private final TransactionRepository txRepo;

    @Transactional(isolation = Isolation.READ_COMMITTED)
    public TransactionResult debit(DebitRequest request) {
        String paymentNo = request.getPaymentNo();

        // 检查是否已记账
        Optional<Transaction> existing =
                txRepo.findByPaymentNo(paymentNo);
        if (existing.isPresent()) {
            Transaction tx = existing.get();
            log.info("重复记账请求，返回已有结果: paymentNo={}, txId={}",
                    paymentNo, tx.getId());
            return TransactionResult.fromExisting(tx);
        }

        // 执行记账
        Account account = accountRepo
                .findByAccountNoForUpdate(request.getAccountNo())
                .orElseThrow(() -> new AccountNotFoundException(
                        request.getAccountNo()));

        if (account.getBalance().compareTo(request.getAmount()) < 0) {
            throw new InsufficientBalanceException(
                    request.getAccountNo(), request.getAmount());
        }

        account.setBalance(
                account.getBalance().subtract(request.getAmount()));
        accountRepo.save(account);

        Transaction tx = new Transaction();
        tx.setPaymentNo(paymentNo);
        tx.setAccountNo(request.getAccountNo());
        tx.setAmount(request.getAmount().negate());
        tx.setType("DEBIT");
        tx.setBalanceAfter(account.getBalance());
        txRepo.save(tx);

        return TransactionResult.fromNew(tx);
    }
}

// Go：支付状态机
type PaymentStatus string

const (
    StatusPending   PaymentStatus = "pending"
    StatusPaying    PaymentStatus = "paying"
    StatusSuccess   PaymentStatus = "success"
    StatusFailed    PaymentStatus = "failed"
    StatusRefunded  PaymentStatus = "refunded"
    StatusClosed    PaymentStatus = "closed"
)

// 合法的状态转移
var validTransitions = map[PaymentStatus][]PaymentStatus{
    StatusPending:  {StatusPaying, StatusClosed},
    StatusPaying:   {StatusSuccess, StatusFailed},
    StatusSuccess:  {StatusRefunded},
    StatusFailed:   {StatusPending},   // 允许重新支付
    StatusRefunded: {},                 // 终态
    StatusClosed:   {},                 // 终态
}

func (s PaymentStatus) CanTransitTo(target PaymentStatus) bool {
    allowed, ok := validTransitions[s]
    if !ok {
        return false
    }
    for _, a := range allowed {
        if a == target {
            return true
        }
    }
    return false
}

// 使用乐观锁更新状态
func updatePaymentStatus(
    ctx context.Context,
    db *sql.DB,
    paymentNo string,
    fromStatus PaymentStatus,
    toStatus PaymentStatus,
) error {

    result, err := db.ExecContext(ctx,
        `UPDATE payment_orders 
         SET status = $1, updated_at = NOW() 
         WHERE payment_no = $2 AND status = $3`,
        toStatus, paymentNo, fromStatus,
    )
    if err != nil {
        return fmt.Errorf("update status: %w", err)
    }

    rows, _ := result.RowsAffected()
    if rows == 0 {
        return fmt.Errorf(
            "status transition rejected: %s -> %s for payment %s",
            fromStatus, toStatus, paymentNo,
        )
    }

    return nil
}

// Go：Token 预分配核心逻辑
package token

import (
    "context"
    "crypto/rand"
    "encoding/hex"
    "fmt"
    "time"

    "github.com/redis/go-redis/v9"
)

type TokenService struct {
    rdb *redis.Client
}

// Generate 生成一次性 Token，存入 Redis 并设置 30 分钟过期
func (s *TokenService) Generate(ctx context.Context, userID string) (string, error) {
    b := make([]byte, 16)
    if _, err := rand.Read(b); err != nil {
        return "", err
    }
    token := hex.EncodeToString(b)
    key := fmt.Sprintf("submit_token:%s:%s", userID, token)
    return token, s.rdb.Set(ctx, key, "valid", 30*time.Minute).Err()
}

// Consume 使用 Lua 脚本原子地校验并删除 Token
func (s *TokenService) Consume(ctx context.Context, userID, token string) error {
    key := fmt.Sprintf("submit_token:%s:%s", userID, token)
    script := redis.NewScript(`
        local val = redis.call('GET', KEYS[1])
        if val == false then return -1 end
        if val ~= 'valid' then return -2 end
        redis.call('DEL', KEYS[1])
        return 1
    `)
    result, err := script.Run(ctx, s.rdb, []string{key}).Int()
    if err != nil {
        return fmt.Errorf("consume token: %w", err)
    }
    if result < 0 {
        return fmt.Errorf("token invalid or already used (code=%d)", result)
    }
    return nil
}

// Go：并发幂等性测试
func TestIdempotency_ConcurrentRequests(t *testing.T) {
    db := setupTestDB(t)
    rdb := setupTestRedis(t)
    svc := NewPaymentService(db)
    guard := NewIdempotencyGuard(rdb)

    key := "test_order_001_pay"
    concurrency := 10
    var wg sync.WaitGroup

    for i := 0; i < concurrency; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            cached, acquired, err := guard.TryAcquire(context.Background(), key)
            if err != nil || cached != nil {
                return
            }
            if acquired {
                svc.CreatePayment(context.Background(), key, "order_001", "user_123", "CNY", 100.00)
                guard.Complete(context.Background(), key, 201, `{"payment_no":"PAY001"}`)
            }
        }()
    }
    wg.Wait()

    // 验证：只创建了一笔支付记录
    var count int
    db.QueryRow("SELECT COUNT(*) FROM payments WHERE idempotency_key = $1", key).Scan(&count)
    if count != 1 {
        t.Errorf("expected 1 payment record, got %d", count)
    }
}

#!/bin/bash
# 幂等性接口测试脚本
API_BASE="http://localhost:8080"
KEY="test-$(date +%s)-001"
ENDPOINT="${API_BASE}/api/v1/payments"

echo "=== 场景一：首次请求 ==="
R1=$(curl -s -w "\n%{http_code}" -X POST "$ENDPOINT" \
  -H "Content-Type: application/json" -H "Idempotency-Key: $KEY" \
  -d '{"order_id":"ORD001","amount":100.00,"currency":"CNY"}')
C1=$(echo "$R1" | tail -1); echo "状态码: $C1"

echo "=== 场景二：重复请求（相同参数） ==="
R2=$(curl -s -w "\n%{http_code}" -X POST "$ENDPOINT" \
  -H "Content-Type: application/json" -H "Idempotency-Key: $KEY" \
  -d '{"order_id":"ORD001","amount":100.00,"currency":"CNY"}')
C2=$(echo "$R2" | tail -1); echo "状态码: $C2"

echo "=== 场景三：重复请求（不同参数） ==="
R3=$(curl -s -w "\n%{http_code}" -X POST "$ENDPOINT" \
  -H "Content-Type: application/json" -H "Idempotency-Key: $KEY" \
  -d '{"order_id":"ORD002","amount":200.00,"currency":"CNY"}')
C3=$(echo "$R3" | tail -1); echo "状态码: $C3（预期 422）"

echo "=== 验证 ==="
[ "$C1" = "201" ] && [ "$C2" = "200" ] && [ "$C3" = "422" ] \
  && echo "所有幂等性测试通过" || { echo "测试失败"; exit 1; }