分布式锁与幂等设计

× 错误理解：有了锁就不需要幂等
   → 锁超时释放后，请求可能重复执行

× 错误理解：有了幂等就不需要锁
   → 并发写入可能导致数据不一致

√ 正确理解：锁和幂等各司其职，互相补充

f(x) = f(f(x))

即：执行一次和执行多次的结果相同

SET key value NX EX seconds

@Service
public class RedisDistributedLock {
    
    @Autowired
    private StringRedisTemplate redisTemplate;
    
    private static final String LOCK_PREFIX = "lock:";
    private static final long DEFAULT_EXPIRE = 30; // 默认 30 秒
    
    /**
     * 加锁
     * @param lockKey 锁的 key
     * @param requestId 请求唯一标识（用于防误删）
     * @param expireSeconds 过期时间（秒）
     * @return 是否加锁成功
     */
    public boolean tryLock(String lockKey, String requestId, long expireSeconds) {
        String key = LOCK_PREFIX + lockKey;
        Boolean result = redisTemplate.opsForValue()
            .setIfAbsent(key, requestId, expireSeconds, TimeUnit.SECONDS);
        return Boolean.TRUE.equals(result);
    }
    
    /**
     * 释放锁（Lua 脚本保证原子性）
     * @param lockKey 锁的 key
     * @param requestId 请求唯一标识
     * @return 是否释放成功
     */
    public boolean releaseLock(String lockKey, String requestId) {
        String key = LOCK_PREFIX + lockKey;
        String script = 
            "if redis.call('get', KEYS[1]) == ARGV[1] then " +
            "    return redis.call('del', KEYS[1]) " +
            "else " +
            "    return 0 " +
            "end";
        
        RedisScript<Long> redisScript = RedisScript.of(script, Long.class);
        Long result = redisTemplate.execute(
            redisScript, 
            Collections.singletonList(key), 
            requestId
        );
        return Long.valueOf(1).equals(result);
    }
}

@Service
public class DailySettlementService {
    
    @Autowired
    private RedisDistributedLock lockService;
    
    @Autowired
    private SettlementService settlementService;
    
    public void executeDailySettlement() {
        String lockKey = "job:daily-settlement";
        String requestId = UUID.randomUUID().toString();
        
        // 尝试加锁
        if (lockService.tryLock(lockKey, requestId, 300)) { // 5 分钟过期
            try {
                // 执行任务
                settlementService.execute();
            } finally {
                // 释放锁（只释放自己加的锁）
                lockService.releaseLock(lockKey, requestId);
            }
        } else {
            log.info("其他节点正在执行，本次跳过");
        }
    }
}

@Service
public class RedisLockWithWatchdog {
    
    @Autowired
    private StringRedisTemplate redisTemplate;
    
    private ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(10);
    private ConcurrentHashMap<String, ScheduledFuture<?>> watchdogs = new ConcurrentHashMap<>();
    
    /**
     * 加锁并启动看门狗
     */
    public boolean tryLockWithWatchdog(String lockKey, String requestId, long expireSeconds) {
        boolean locked = tryLock(lockKey, requestId, expireSeconds);
        if (locked) {
            // 启动看门狗，每隔过期时间的 1/3 续期一次
            long watchInterval = expireSeconds * 1000 / 3;
            ScheduledFuture<?> future = scheduler.scheduleAtFixedRate(() -> {
                try {
                    renewLock(lockKey, requestId, expireSeconds);
                } catch (Exception e) {
                    log.error("锁续期失败", e);
                }
            }, watchInterval, watchInterval, TimeUnit.MILLISECONDS);
            
            watchdogs.put(lockKey, future);
        }
        return locked;
    }
    
    /**
     * 续期
     */
    private boolean renewLock(String lockKey, String requestId, long expireSeconds) {
        String key = "lock:" + lockKey;
        String script = 
            "if redis.call('get', KEYS[1]) == ARGV[1] then " +
            "    return redis.call('expire', KEYS[1], ARGV[2]) " +
            "else " +
            "    return 0 " +
            "end";
        
        RedisScript<Long> redisScript = RedisScript.of(script, Long.class);
        Long result = redisTemplate.execute(
            redisScript, 
            Collections.singletonList(key), 
            requestId, 
            String.valueOf(expireSeconds)
        );
        return Long.valueOf(1).equals(result);
    }
    
    /**
     * 释放锁并停止看门狗
     */
    public boolean releaseLockWithWatchdog(String lockKey, String requestId) {
        // 停止看门狗
        ScheduledFuture<?> future = watchdogs.remove(lockKey);
        if (future != null) {
            future.cancel(false);
        }
        
        // 释放锁
        return releaseLock(lockKey, requestId);
    }
}

-- 释放锁的 Lua 脚本
if redis.call('get', KEYS[1]) == ARGV[1] then
    return redis.call('del', KEYS[1])
else
    return 0
end

// 使用 Redisson 实现 Redlock
Config config1 = new Config();
config1.useSingleServer().setAddress("redis://node1:6379");

Config config2 = new Config();
config2.useSingleServer().setAddress("redis://node2:6379");

Config config3 = new Config();
config3.useSingleServer().setAddress("redis://node3:6379");

RedissonClient client1 = Redisson.create(config1);
RedissonClient client2 = Redisson.create(config2);
RedissonClient client3 = Redisson.create(config3);

RLock lock1 = client1.getLock("myLock");
RLock lock2 = client2.getLock("myLock");
RLock lock3 = client3.getLock("myLock");

RedissonRedLock redLock = new RedissonRedLock(lock1, lock2, lock3);

try {
    // 尝试加锁
    boolean locked = redLock.tryLock(10, 30, TimeUnit.SECONDS);
    if (locked) {
        // 执行业务
    }
} finally {
    redLock.unlock();
}

@Configuration
public class RedissonConfig {
    
    @Bean
    public RedissonClient redissonClient() {
        Config config = new Config();
        
        // 单节点配置
        config.useSingleServer()
            .setAddress("redis://127.0.0.1:6379")
            .setConnectionPoolSize(64)
            .setConnectionMinimumIdleSize(24)
            .setIdleConnectionTimeout(10000)
            .setConnectTimeout(10000)
            .setTimeout(3000)
            .setRetryAttempts(3)
            .setRetryInterval(1500);
        
        // 集群配置
        // config.useClusterServers()
        //     .addNodeAddress("redis://node1:6379", "redis://node2:6379", "redis://node3:6379");
        
        return Redisson.create(config);
    }
}

@Service
public class DistributedLockService {
    
    @Autowired
    private RedissonClient redissonClient;
    
    public void executeWithLock(String lockKey, Runnable task) {
        RLock lock = redissonClient.getLock(lockKey);
        
        try {
            // 尝试加锁，最多等待 10 秒，锁 30 秒后自动释放
            boolean locked = lock.tryLock(10, 30, TimeUnit.SECONDS);
            if (locked) {
                task.run();
            } else {
                throw new RuntimeException("获取锁失败");
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("线程被中断", e);
        } finally {
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }
}

@Service
public class PaymentService {
    
    @Autowired
    private PaymentRecordRepository paymentRepository;
    
    @Transactional(rollbackFor = Exception.class)
    public void handlePayment(PaymentRequest request) {
        // 幂等检查：是否已处理
        if (paymentRepository.existsByPayNo(request.getPayNo())) {
            log.info("支付已处理，幂等返回: {}", request.getPayNo());
            return;
        }
        
        // 执行业务逻辑
        // ...
        
        // 记录支付流水
        PaymentRecord record = PaymentRecord.builder()
            .payNo(request.getPayNo())
            .amount(request.getAmount())
            .build();
        paymentRepository.save(record);
    }
}

-- 创建唯一索引
CREATE UNIQUE INDEX uk_order_no ON orders(order_no);

@Service
public class OrderService {
    
    @Autowired
    private OrderRepository orderRepository;
    
    @Transactional(rollbackFor = Exception.class)
    public Order createOrder(CreateOrderCommand command) {
        try {
            Order order = Order.create(command);
            return orderRepository.save(order);
        } catch (DataIntegrityViolationException e) {
            // 唯一索引冲突，说明订单已存在
            log.info("订单已存在，幂等返回: {}", command.getOrderNo());
            return orderRepository.findByOrderNo(command.getOrderNo());
        }
    }
}

@Service
public class IdempotentTokenService {
    
    @Autowired
    private StringRedisTemplate redisTemplate;
    
    private static final String TOKEN_PREFIX = "idempotent:token:";
    
    /**
     * 生成 Token
     */
    public String createToken(String bizType) {
        String token = UUID.randomUUID().toString();
        String key = TOKEN_PREFIX + bizType + ":" + token;
        // Token 有效期 5 分钟
        redisTemplate.opsForValue().set(key, "1", 5, TimeUnit.MINUTES);
        return token;
    }
    
    /**
     * 验证并删除 Token（原子操作）
     */
    public boolean validateAndDeleteToken(String bizType, String token) {
        String key = TOKEN_PREFIX + bizType + ":" + token;
        
        // Lua 脚本：存在则删除，返回 1；不存在返回 0
        String script = 
            "if redis.call('exists', KEYS[1]) == 1 then " +
            "    return redis.call('del', KEYS[1]) " +
            "else " +
            "    return 0 " +
            "end";
        
        RedisScript<Long> redisScript = RedisScript.of(script, Long.class);
        Long result = redisTemplate.execute(redisScript, Collections.singletonList(key));
        return Long.valueOf(1).equals(result);
    }
}

@RestController
public class OrderController {
    
    @Autowired
    private IdempotentTokenService tokenService;
    
    @Autowired
    private OrderService orderService;
    
    /**
     * 获取 Token
     */
    @GetMapping("/api/order/token")
    public String getToken() {
        return tokenService.createToken("order");
    }
    
    /**
     * 创建订单（幂等）
     */
    @PostMapping("/api/order/create")
    public Order createOrder(
        @RequestHeader("Idempotent-Token") String token,
        @RequestBody CreateOrderCommand command
    ) {
        // 验证 Token
        if (!tokenService.validateAndDeleteToken("order", token)) {
            throw new RuntimeException("重复请求或 Token 已失效");
        }
        
        // 执行业务
        return orderService.createOrder(command);
    }
}

@Service
public class OrderStateMachine {
    
    @Autowired
    private OrderRepository orderRepository;
    
    // 定义允许的状态流转
    private static final Map<OrderStatus, Set<OrderStatus>> ALLOWED_TRANSITIONS = Map.of(
        OrderStatus.INIT, Set.of(OrderStatus.PAID, OrderStatus.CANCELLED),
        OrderStatus.PAID, Set.of(OrderStatus.SHIPPED, OrderStatus.REFUNDED),
        OrderStatus.SHIPPED, Set.of(OrderStatus.COMPLETED),
        OrderStatus.COMPLETED, Set.of(),
        OrderStatus.CANCELLED, Set.of(),
        OrderStatus.REFUNDED, Set.of()
    );
    
    @Transactional(rollbackFor = Exception.class)
    public void payOrder(String orderNo) {
        Order order = orderRepository.findByOrderNo(orderNo);
        
        // 状态机检查：只允许 INIT -> PAID
        if (!canTransition(order.getStatus(), OrderStatus.PAID)) {
            throw new IllegalStateException(
                "订单状态不允许支付: " + order.getStatus()
            );
        }
        
        // 更新状态
        order.setStatus(OrderStatus.PAID);
        order.setPayTime(LocalDateTime.now());
        orderRepository.save(order);
    }
    
    private boolean canTransition(OrderStatus from, OrderStatus to) {
        Set<OrderStatus> allowedTargets = ALLOWED_TRANSITIONS.get(from);
        return allowedTargets != null && allowedTargets.contains(to);
    }
}

CREATE TABLE idempotent_record (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    biz_type VARCHAR(50) NOT NULL COMMENT '业务类型',
    biz_key VARCHAR(200) NOT NULL COMMENT '业务唯一键',
    request_hash VARCHAR(64) COMMENT '请求参数哈希（可选）',
    create_time DATETIME DEFAULT CURRENT_TIMESTAMP,
    UNIQUE KEY uk_biz_type_key (biz_type, biz_key)
) COMMENT '幂等记录表';

@Service
public class IdempotentRecordService {
    
    @Autowired
    private IdempotentRecordRepository idempotentRepository;
    
    /**
     * 检查并记录（原子操作）
     */
    @Transactional(rollbackFor = Exception.class)
    public boolean checkAndRecord(String bizType, String bizKey) {
        try {
            IdempotentRecord record = IdempotentRecord.builder()
                .bizType(bizType)
                .bizKey(bizKey)
                .build();
            idempotentRepository.save(record);
            return true; // 首次请求
        } catch (DataIntegrityViolationException e) {
            return false; // 重复请求
        }
    }
}

@Service
public class RedisIdempotentService {
    
    @Autowired
    private StringRedisTemplate redisTemplate;
    
    @Autowired
    private IdempotentRecordRepository dbRepository;
    
    private static final String IDEMPOTENT_PREFIX = "idempotent:";
    
    /**
     * 检查幂等（先查 Redis，再查数据库）
     */
    public boolean isProcessed(String bizType, String bizKey) {
        String key = IDEMPOTENT_PREFIX + bizType + ":" + bizKey;
        
        // 1. 先查 Redis
        if (Boolean.TRUE.equals(redisTemplate.hasKey(key))) {
            return true;
        }
        
        // 2. 再查数据库（防止 Redis 数据丢失）
        return dbRepository.existsByBizTypeAndBizKey(bizType, bizKey);
    }
    
    /**
     * 标记已处理
     */
    public void markProcessed(String bizType, String bizKey) {
        String key = IDEMPOTENT_PREFIX + bizType + ":" + bizKey;
        
        // 1. 写入 Redis（过期时间 1 天）
        redisTemplate.opsForValue().set(key, "1", 1, TimeUnit.DAYS);
        
        // 2. 写入数据库（持久化）
        IdempotentRecord record = IdempotentRecord.builder()
            .bizType(bizType)
            .bizKey(bizKey)
            .build();
        dbRepository.save(record);
    }
}

@Service
public class PaymentCallbackService {
    
    @Autowired
    private OrderRepository orderRepository;
    
    @Autowired
    private PaymentRecordRepository paymentRepository;
    
    @Autowired
    private EventPublisher eventPublisher;
    
    @Transactional(rollbackFor = Exception.class)
    public void handlePaymentCallback(PaymentCallback callback) {
        // 1. 幂等检查：是否已处理
        if (paymentRepository.existsByPayNo(callback.getPayNo())) {
            log.info("支付回调已处理，幂等返回: {}", callback.getPayNo());
            return;
        }
        
        // 2. 查询订单
        Order order = orderRepository.findByOrderNo(callback.getOrderNo());
        if (order == null) {
            throw new RuntimeException("订单不存在: " + callback.getOrderNo());
        }
        
        // 3. 状态机检查：只允许 INIT -> PAID
        if (order.getStatus() != OrderStatus.INIT) {
            log.warn("订单状态不允许支付: orderNo={}, status={}", 
                order.getOrderNo(), order.getStatus());
            return; // 直接返回，不抛异常
        }
        
        // 4. 更新订单状态
        order.setStatus(OrderStatus.PAID);
        order.setPayTime(callback.getPayTime());
        orderRepository.save(order);
        
        // 5. 记录支付流水
        PaymentRecord record = PaymentRecord.builder()
            .payNo(callback.getPayNo())
            .orderNo(order.getOrderNo())
            .amount(callback.getAmount())
            .payTime(callback.getPayTime())
            .build();
        paymentRepository.save(record);
        
        // 6. 发送下游事件（异步）
        eventPublisher.publish("order-paid", order);
    }
}

@Service
public class DailySettlementJob {
    
    @Autowired
    private RedissonClient redissonClient;
    
    @Autowired
    private SettlementService settlementService;
    
    @Scheduled(cron = "0 0 2 * * ?") // 每天凌晨 2 点执行
    public void execute() {
        RLock lock = redissonClient.getLock("job:daily-settlement");
        
        try {
            // 尝试加锁，最多等待 0 秒，锁 1 小时后自动释放
            boolean locked = lock.tryLock(0, 3600, TimeUnit.SECONDS);
            if (locked) {
                log.info("获取锁成功，开始执行对账任务");
                settlementService.execute();
            } else {
                log.info("其他节点正在执行，本次跳过");
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("获取锁被中断", e);
        } finally {
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }
}

@Service
public class SeckillService {
    
    @Autowired
    private StringRedisTemplate redisTemplate;
    
    @Autowired
    private OrderService orderService;
    
    /**
     * 秒杀下单
     */
    public OrderResult seckill(SeckillRequest request) {
        String productId = request.getProductId();
        String userId = request.getUserId();
        
        // 1. 幂等检查：是否已下单
        String orderKey = "seckill:order:" + productId + ":" + userId;
        if (Boolean.TRUE.equals(redisTemplate.hasKey(orderKey))) {
            throw new RuntimeException("您已参与过秒杀");
        }
        
        // 2. 库存扣减（Lua 脚本保证原子性）
        String stockKey = "seckill:stock:" + productId;
        String stockScript = 
            "if tonumber(redis.call('get', KEYS[1])) > 0 then " +
            "    return redis.call('decr', KEYS[1]) " +
            "else " +
            "    return -1 " +
            "end";
        
        RedisScript<Long> redisScript = RedisScript.of(stockScript, Long.class);
        Long stock = redisTemplate.execute(redisScript, Collections.singletonList(stockKey));
        
        if (stock == null || stock < 0) {
            throw new RuntimeException("商品已售罄");
        }
        
        try {
            // 3. 创建订单（数据库事务）
            Order order = orderService.createOrder(request);
            
            // 4. 标记已下单
            redisTemplate.opsForValue().set(orderKey, order.getOrderNo(), 1, TimeUnit.DAYS);
            
            return OrderResult.success(order);
        } catch (Exception e) {
            // 5. 创建订单失败，回滚库存
            redisTemplate.opsForValue().increment(stockKey);
            throw new RuntimeException("下单失败", e);
        }
    }
}

// × 错误：锁粒度太粗，吞吐低
public void updateInventory(Long productId, Integer quantity) {
    RLock lock = redissonClient.getLock("inventory");
    lock.lock();
    try {
        // 更新库存
    } finally {
        lock.unlock();
    }
}

// √ 正确：锁粒度细，只锁定特定商品
public void updateInventory(Long productId, Integer quantity) {
    RLock lock = redissonClient.getLock("inventory:" + productId);
    lock.lock();
    try {
        // 更新库存
    } finally {
        lock.unlock();
    }
}

// × 错误：无限等待，可能阻塞线程池
RLock lock = redissonClient.getLock("myLock");
lock.lock();

// √ 正确：设置合理的等待时间
RLock lock = redissonClient.getLock("myLock");
boolean locked = lock.tryLock(3, 30, TimeUnit.SECONDS);
if (!locked) {
    // 快速失败或降级处理
    return;
}

// 需要同时锁定多个资源
public void transfer(Account from, Account to, BigDecimal amount) {
    RLock lock1 = redissonClient.getLock("account:" + from.getId());
    RLock lock2 = redissonClient.getLock("account:" + to.getId());
    
    // 按固定顺序加锁，避免死锁
    RLock multiLock = redissonClient.getMultiLock(lock1, lock2);
    
    try {
        multiLock.lock();
        // 执行转账
    } finally {
        multiLock.unlock();
    }
}

问题类型是什么？
├─ 并发访问共享资源 → 分布式锁
│   └─ 是否可以用数据库锁/CAS 解决？
│       ├─ 是 → 使用数据库锁/CAS
│       └─ 否 → 使用分布式锁（Redis/ZooKeeper）
│
└─ 重复请求/消息处理 → 幂等设计
    └─ 是否有业务单号？
        ├─ 是 → 业务单号 + 唯一约束
        └─ 否 → 幂等 Token / 去重表