幂等性（Idempotency）：从理论到分布式系统实践

CREATE TABLE idempotency_record (
    idempotency_key VARCHAR(128) PRIMARY KEY,
    request_hash    VARCHAR(256) NOT NULL,
    status          ENUM('PROCESSING', 'SUCCESS', 'FAILURE') NOT NULL,
    response_data   JSON,
    created_at      TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at      TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);

-- 业务表中嵌入幂等令牌作为唯一约束
CREATE TABLE payment_order (
    order_id        BIGINT PRIMARY KEY AUTO_INCREMENT,
    idempotency_key VARCHAR(128) NOT NULL UNIQUE,
    amount          DECIMAL(12, 2) NOT NULL,
    status          ENUM('PENDING', 'PAID', 'FAILED') NOT NULL,
    created_at      TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

| 1 bit | 41 bits          | 10 bits     | 12 bits       |
| 符号位 | 时间戳（毫秒级）   | 工作节点 ID  | 序列号         |

| 48 bits          | 80 bits              |
| 时间戳（毫秒级）   | 随机数                |

import etcd3

def acquire_idempotent_lock(key: str, ttl: int = 10) -> bool:
    client = etcd3.client()
    lock = client.lock(f"idempotent:{key}", ttl=ttl)
    acquired = lock.acquire(timeout=5)
    if acquired:
        try:
            execute_business_logic(key)
        finally:
            lock.release()
    return acquired

public class PaymentAggregate {

    private final List<DomainEvent> pendingEvents = new ArrayList<>();
    private final Set<String> processedCommandIds = new HashSet<>();
    private Money balance;

    public void handle(DepositCommand cmd) {
        if (processedCommandIds.contains(cmd.getCommandId())) {
            return;
        }
        if (cmd.getAmount().isNegative()) {
            throw new IllegalArgumentException("Amount must be positive");
        }
        pendingEvents.add(new MoneyDepositedEvent(
            cmd.getAggregateId(),
            cmd.getCommandId(),
            cmd.getAmount(),
            Instant.now()
        ));
        processedCommandIds.add(cmd.getCommandId());
    }

    public void apply(MoneyDepositedEvent event) {
        this.balance = this.balance.add(event.getAmount());
        processedCommandIds.add(event.getCommandId());
    }
}

CREATE TABLE payment_transaction (
    id               BIGINT PRIMARY KEY AUTO_INCREMENT,
    idempotency_key  VARCHAR(128) NOT NULL UNIQUE,
    merchant_id      VARCHAR(64) NOT NULL,
    amount           DECIMAL(12, 2) NOT NULL,
    currency         CHAR(3) NOT NULL DEFAULT 'CNY',
    status           ENUM('PROCESSING', 'SUCCESS', 'FAILURE', 'REFUNDED') NOT NULL,
    channel          VARCHAR(32) NOT NULL,
    channel_tx_id    VARCHAR(128),
    created_at       TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at       TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    INDEX idx_merchant (merchant_id),
    INDEX idx_status (status)
);

public class PaymentService {

    private final PaymentTransactionRepository txRepo;
    private final PaymentChannelClient channelClient;

    public PaymentResult process(PaymentRequest request) {
        String idempotencyKey = request.getIdempotencyKey();

        PaymentTransaction existingTx = txRepo.findByIdempotencyKey(idempotencyKey);
        if (existingTx != null) {
            return handleExistingTransaction(existingTx);
        }

        PaymentTransaction tx = new PaymentTransaction();
        tx.setIdempotencyKey(idempotencyKey);
        tx.setAmount(request.getAmount());
        tx.setStatus(TransactionStatus.PROCESSING);

        try {
            txRepo.save(tx);
        } catch (DuplicateKeyException e) {
            PaymentTransaction concurrentTx = txRepo.findByIdempotencyKey(idempotencyKey);
            return handleExistingTransaction(concurrentTx);
        }

        try {
            ChannelResponse response = channelClient.charge(
                request.getChannel(), request.getAmount(), idempotencyKey
            );
            tx.setStatus(TransactionStatus.SUCCESS);
            tx.setChannelTxId(response.getTransactionId());
            txRepo.update(tx);
            return PaymentResult.success(tx);
        } catch (ChannelException e) {
            tx.setStatus(TransactionStatus.FAILURE);
            txRepo.update(tx);
            return PaymentResult.failure(e.getMessage());
        }
    }

    private PaymentResult handleExistingTransaction(PaymentTransaction tx) {
        return switch (tx.getStatus()) {
            case SUCCESS -> PaymentResult.success(tx);
            case FAILURE -> PaymentResult.failure("Previous attempt failed, retry allowed");
            case PROCESSING -> PaymentResult.conflict("Request is being processed");
            default -> PaymentResult.failure("Unknown status");
        };
    }
}

public class IdempotentMessageConsumer {

    private final JdbcTemplate jdbcTemplate;
    private final BusinessService businessService;

    @KafkaListener(topics = "order-events", groupId = "order-processor")
    public void consume(ConsumerRecord<String, OrderEvent> record, Acknowledgment ack) {
        String messageId = buildMessageId(record);

        int inserted = jdbcTemplate.update(
            "INSERT INTO message_dedup (message_id, topic, partition_offset, created_at) " +
            "VALUES (?, ?, ?, NOW()) ON DUPLICATE KEY UPDATE message_id = message_id",
            messageId, record.topic(), record.offset()
        );

        if (inserted > 0 || isDuplicateAllowed(record)) {
            businessService.process(record.value());
        }

        ack.acknowledge();
    }

    private String buildMessageId(ConsumerRecord<String, ?> record) {
        return record.topic() + ":" + record.partition() + ":" + record.offset();
    }
}

❌ 错误：在重试时重新生成令牌
   请求1: POST /pay { idempotency_key: "abc" }  → 超时
   请求2: POST /pay { idempotency_key: "def" }  → 重复扣款

✅ 正确：令牌在首次请求时生成，重试复用同一令牌
   请求1: POST /pay { idempotency_key: "abc" }  → 超时
   请求2: POST /pay { idempotency_key: "abc" }  → 幂等命中

❌ 错误模式：
   if (!exists(key)) {        // 并发请求均通过此检查
       insert(key, data);     // 多个请求均插入成功
   }

✅ 正确模式：
   try {
       insert(key, data);     // 依赖唯一索引，重复插入抛异常
   } catch (DuplicateKeyException e) {
       return query(key);     // 幂等命中，返回已有结果
   }

❌ 错误：
   1. INSERT idempotency_record (key, PROCESSING)   -- 事务1
   2. 执行业务逻辑                                    -- 事务2
   3. UPDATE idempotency_record SET status=SUCCESS   -- 事务1

   若步骤2失败但步骤1已提交，令牌状态为 PROCESSING，后续重试可能被拒绝

✅ 正确：
   将幂等记录与业务操作放在同一事务中，或使用补偿机制清理 PROCESSING 状态

import time

import boto3
from botocore.exceptions import ConditionalCheckFailedException

dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table("idempotency_store")

def lambda_handler(event, context):
    idempotency_key = event["idempotency_key"]

    try:
        table.put_item(
            Item={
                "idempotency_key": idempotency_key,
                "status": "PROCESSING",
                "ttl": int(time.time()) + 3600,
            },
            ConditionExpression="attribute_not_exists(idempotency_key)",
        )
    except ConditionalCheckFailedException:
        return get_cached_response(idempotency_key)

    try:
        result = execute_business_logic(event)
        table.update_item(
            Key={"idempotency_key": idempotency_key},
            UpdateExpression="SET #s = :s, response = :r",
            ExpressionAttributeNames={"#s": "status"},
            ExpressionAttributeValues={":s": "SUCCESS", ":r": result},
        )
        return result
    except Exception as e:
        table.update_item(
            Key={"idempotency_key": idempotency_key},
            UpdateExpression="SET #s = :s",
            ExpressionAttributeNames={"#s": "status"},
            ExpressionAttributeValues={":s": "FAILURE"},
        )
        raise