接口幂等方案总结：从理论到生产落地

1
-- KEYS[1]: token key
2
-- ARGV[1]: expected token value
3
-- 返回: 1 = 获取成功（首次），0 = token 已失效（重复请求）
4
if redis.call('get', KEYS[1]) == ARGV[1] then
5
    redis.call('del', KEYS[1])
6
    return 1
7
else
8
    return 0
9
end

1
import org.springframework.data.redis.core.StringRedisTemplate
2
import org.springframework.stereotype.Component
3
import java.util.UUID
4
import java.util.concurrent.TimeUnit
5

6
@Component
7
class IdempotentTokenService(
8
    private val redisTemplate: StringRedisTemplate
9
) {
10
    companion object {
11
        private const val TOKEN_PREFIX = "idempotent:token:"
12
        private const val TOKEN_EXPIRE_MINUTES = 10L
13

14
        private val CHECK_AND_DELETE_SCRIPT = """
15
            if redis.call('get', KEYS[1]) == ARGV[1] then
16
                return redis.call('del', KEYS[1])
17
            else
18
                return 0
19
            end
20
        """.trimIndent()
21
    }
22

23
    /** 生成 Token，返回给客户端 */
24
    fun createToken(bizType: String): String {
25
        val token = UUID.randomUUID().toString()
26
        val key = "$TOKEN_PREFIX$bizType:$token"
27
        redisTemplate.opsForValue().set(key, "1", TOKEN_EXPIRE_MINUTES, TimeUnit.MINUTES)
28
        return token
29
    }
30

31
    /** 校验并消费 Token，返回 true 表示首次请求 */
32
    fun checkAndConsume(bizType: String, token: String): Boolean {
33
        val key = "$TOKEN_PREFIX$bizType:$token"
34
        val result = redisTemplate.execute(
35
            org.springframework.scripting.support.ResourceScriptSource(
36
                org.springframework.core.io.ClassPathResource("scripts/idempotent.lua")
37
            ),
38
            listOf(key),
39
            token
40
        )
41
        return result == 1L
42
    }
43
}

1
@Entity
2
@Table(
3
    name = "t_order",
4
    indexes = [Index(name = "uk_request_id", columnList = "requestId", unique = true)]
5
)
6
class Order(
7
    @Id
8
    @GeneratedValue(strategy = GenerationType.IDENTITY)
9
    val id: Long = 0,
10

11
    @Column(nullable = false, unique = true, length = 64)
12
    val requestId: String,
13

14
    @Column(nullable = false)
15
    val userId: Long,
16

17
    @Column(nullable = false)
18
    val amount: BigDecimal,
19

20
    @Column(nullable = false, length = 20)
21
    var status: String = "CREATED"
22
)
23

24
@Service
25
class OrderService(private val orderRepo: OrderRepository) {
26

27
    fun createOrder(userId: Long, amount: BigDecimal, requestId: String): Order {
28
        // 先查询是否已存在（提升体验，避免直接抛异常）
29
        val existing = orderRepo.findByRequestId(requestId)
30
        if (existing != null) {
31
            return existing
32
        }
33

34
        return try {
35
            orderRepo.save(Order(requestId = requestId, userId = userId, amount = amount))
36
        } catch (e: DataIntegrityViolationException) {
37
            // 并发场景下，另一个请求先插入了，这里兜底返回
38
            orderRepo.findByRequestId(requestId)
39
                ?: throw IllegalStateException("订单创建异常，请重试")
40
        }
41
    }
42
}

1
import org.springframework.data.redis.core.StringRedisTemplate
2
import org.springframework.stereotype.Component
3
import java.util.concurrent.TimeUnit
4

5
@Component
6
class RedisIdempotentService(
7
    private val redisTemplate: StringRedisTemplate
8
) {
9
    companion object {
10
        private const val KEY_PREFIX = "idempotent:dedup:"
11
        private const val EXPIRE_SECONDS = 3600L  // 1 小时过期
12
    }
13

14
    /**
15
     * 尝试标记请求，返回 true 表示首次请求
16
     * @param bizType   业务类型，如 "order_create"
17
     * @param bizKey    业务幂等键，如 requestId
18
     */
19
    fun tryAcquire(bizType: String, bizKey: String): Boolean {
20
        val key = "$KEY_PREFIX$bizType:$bizKey"
21
        return redisTemplate.opsForValue()
22
            .setIfAbsent(key, "1", EXPIRE_SECONDS, TimeUnit.SECONDS) == true
23
    }
24
}

1
@Service
2
class PaymentService(
3
    private val redisIdempotent: RedisIdempotentService,
4
    private val orderRepo: OrderRepository
5
) {
6
    fun processPayment(orderId: Long, requestId: String): PaymentResult {
7
        // 前置去重
8
        if (!redisIdempotent.tryAcquire("payment", requestId)) {
9
            // 已处理过，查询并返回之前的结果
10
            return orderRepo.findByRequestId(requestId)
11
                ?.let { PaymentResult.alreadyProcessed(it) }
12
                ?: PaymentResult.duplicate()
13
        }
14

15
        return try {
16
            // 执行业务逻辑
17
            doPayment(orderId)
18
        } catch (e: Exception) {
19
            // 业务失败时，删除去重标记，允许重试
20
            redisIdempotent.release("payment", requestId)
21
            throw e
22
        }
23
    }
24
}

1
UPDATE t_order
2
SET status = 'PAID', paid_at = NOW()
3
WHERE order_id = #{orderId} AND status = 'CREATED'

1
enum class OrderStatus(val code: String) {
2
    CREATED("CREATED"),
3
    PAID("PAID"),
4
    SHIPPED("SHIPPED"),
5
    COMPLETED("COMPLETED"),
6
    CANCELLED("CANCELLED");
7

8
    /** 定义合法的状态流转 */
9
    fun canTransitionTo(target: OrderStatus): Boolean = when (this) {
10
        CREATED  -> target in setOf(PAID, CANCELLED)
11
        PAID     -> target in setOf(SHIPPED, CANCELLED)
12
        SHIPPED  -> target == COMPLETED
13
        else     -> false
14
    }
15
}
16

17
@Repository
18
class OrderRepository(private val jdbcTemplate: JdbcTemplate) {
19

20
    fun updateStatus(orderId: Long, from: OrderStatus, to: OrderStatus): Boolean {
21
        val affected = jdbcTemplate.update("""
22
            UPDATE t_order
23
            SET status = ?, updated_at = NOW()
24
            WHERE order_id = ? AND status = ?
25
        """, to.code, orderId, from.code)
26
        return affected > 0
27
    }
28
}
29

30
@Service
31
class OrderStatusService(private val orderRepo: OrderRepository) {
32

33
    fun payOrder(orderId: Long): PayResult {
34
        val success = orderRepo.updateStatus(orderId, OrderStatus.CREATED, OrderStatus.PAID)
35
        return if (success) {
36
            PayResult.success("支付成功")
37
        } else {
38
            // 已经不在 CREATED 状态，查询当前状态返回
39
            val current = orderRepo.findById(orderId)?.status
40
            PayResult.alreadyHandled("订单当前状态: $current")
41
        }
42
    }
43
}

1
UPDATE t_account
2
SET balance = balance - #{amount}, version = version + 1
3
WHERE user_id = #{userId} AND version = #{currentVersion}

1
@Repository
2
class AccountRepository(private val jdbcTemplate: JdbcTemplate) {
3

4
    fun deductBalance(userId: Long, amount: BigDecimal, expectedVersion: Int): Boolean {
5
        val affected = jdbcTemplate.update("""
6
            UPDATE t_account
7
            SET balance = balance - ?, version = version + 1, updated_at = NOW()
8
            WHERE user_id = ? AND version = ? AND balance >= ?
9
        """, amount, userId, expectedVersion, amount)
10
        return affected > 0
11
    }
12
}

1
@Component
2
class OrderMessageConsumer(
3
    private val redisTemplate: StringRedisTemplate,
4
    private val orderService: OrderService
5
) {
6
    companion object {
7
        private const val DEDUP_KEY_PREFIX = "mq:dedup:order:"
8
        private const val DEDUP_EXPIRE_HOURS = 24L
9
    }
10

11
    fun handleMessage(message: OrderMessage) {
12
        val dedupKey = "$DEDUP_KEY_PREFIX${message.messageId}"
13

14
        // 幂等校验：SET NX
15
        val isNew = redisTemplate.opsForValue()
16
            .setIfAbsent(dedupKey, "1", DEDUP_EXPIRE_HOURS, TimeUnit.HOURS)
17

18
        if (isNew != true) {
19
            // 已处理过，跳过
20
            return
21
        }
22

23
        try {
24
            orderService.processOrderMessage(message)
25
        } catch (e: Exception) {
26
            // 处理失败，删除去重标记，允许 MQ 重试
27
            redisTemplate.delete(dedupKey)
28
            throw e
29
        }
30
    }
31
}

1
// 错误示例：先去重再执行，中间如果宕机，去重标记在但业务没执行
2
redis.setIfAbsent(key, "1")  // 成功
3
// ← 这里服务宕机了
4
doBusiness()                   // 没执行

1
// 支付回调必须幂等
2
@PostMapping("/callback/payment")
3
fun handlePaymentCallback(@RequestBody callback: PaymentCallback): Response {
4
    // 即使是回调，也要走幂等逻辑
5
    if (!idempotentService.tryAcquire("pay_callback", callback.tradeNo)) {
6
        return Response.success("已处理")
7
    }
8
    orderService.confirmPayment(callback)
9
    return Response.success("处理成功")
10
}