面试题解析：如何解决分布式秒杀系统中的库存超卖问题？
问题背景
在构建分布式秒杀系统时，一个常见的挑战是如何防止库存超卖问题。当多个用户同时抢购同一商品时，如果不加以控制，可能导致库存出现负数，影响系统的稳定性和用户体验。本文将讨论这个问题，并提供一种综合的解决方案。
解决思路
1. 乐观锁机制
在数据库层面使用乐观锁，通过版本号或时间戳来确保并发更新的一致性。在减库存的操作中，先查询当前库存版本，然后在更新库存的同时更新版本号，确保在更新时库存版本没有被其他线程修改。
2. Redis预减库存
通过将商品库存提前加载到Redis缓存中，用户抢购时，先从Redis中扣减库存，再异步将扣减后的库存同步到数据库。这减轻了数据库的压力，提高了系统的并发处理能力。
3. 分布式锁
在关键操作上使用分布式锁，确保同一时刻只有一个请求能够执行关键操作，防止多个用户并发执行导致的问题。使用Redis的分布式锁实现，保证锁的互斥性和超时处理。
4. 消息队列确保顺序
将用户抢购请求放入消息队列，保证抢购的顺序。在消息队列中使用分布式锁来确保同一时刻只有一个消息能够被消费，以保证订单生成的有序性。
5. 限制抢购频率
使用Redis的计数器来记录用户的请求次数，并设置一个合理的抢购频率限制。这样可以避免某个用户通过高频请求导致超卖。
详细实现方案
1. 乐观锁机制
@Entity public class Product { @Id private Long id; private Integer stock; @Version private Long version; } @Service public class ProductService { @Transactional public void purchaseProduct(Long productId, int quantity) { Product product = productRepository.findById(productId).orElseThrow(ProductNotFoundException::new); if (product.getStock() >= quantity) { product.setStock(product.getStock() - quantity); productRepository.save(product); // 生成订单等后续操作... } else { // 库存不足，处理失败逻辑... } } }
2. Redis预减库存
@Service public class RedisStockService { private final String STOCK_KEY_PREFIX = "stock:product:"; @Autowired private RedisTemplate<String, Integer> redisTemplate; public int getStock(Long productId) { String key = STOCK_KEY_PREFIX + productId; return redisTemplate.opsForValue().get(key); } public void reduceStock(Long productId, int quantity) { String key = STOCK_KEY_PREFIX + productId; redisTemplate.opsForValue().decrement(key, quantity); // 异步更新数据库库存... } }
3. 分布式锁
@Component public class DistributedLockService { @Autowired private StringRedisTemplate redisTemplate; public boolean tryLock(String lockKey, String requestId, long expireTime) { Boolean locked = redisTemplate.opsForValue().setIfAbsent(lockKey, requestId, expireTime, TimeUnit.MILLISECONDS); return Boolean.TRUE.equals(locked); } public void unlock(String lockKey, String requestId) { String storedRequestId = redisTemplate.opsForValue().get(lockKey); if (requestId.equals(storedRequestId)) { redisTemplate.delete(lockKey); } } }
4. 消息队列确保顺序
@Service public class RabbitMQService { @Autowired private RabbitTemplate rabbitTemplate; public void sendSeckillOrderRequest(Long userId, Long productId) { // 构建消息体... rabbitTemplate.convertAndSend("seckill.exchange", "seckill.order", message); } }
5. 限制抢购频率
@Service public class RequestLimitService { private final String REQUEST_LIMIT_KEY_PREFIX = "request:limit:user:"; @Autowired private RedisTemplate<String, Integer> redisTemplate; public boolean checkRequestLimit(Long userId, int limit) { String key = REQUEST_LIMIT_KEY_PREFIX + userId; int count = redisTemplate.opsForValue().increment(key, 1); if (count > limit) { // 超过频率限制，处理失败逻辑... return false; } return true; } }
示例回答：
“首先，我们采用乐观锁的机制，通过数据库版本号或时间戳来确保并发更新的一致性。这可以在减库存的操作中先查询当前库存版本，然后在更新库存的同时更新版本号。”
“其次，为了减轻数据库压力，我们通过Redis预减库存的方式。将商品库存提前加载到Redis缓存中，用户抢购时先从Redis中扣减库存，再异步将扣减后的库存同步到数据库。”
“为了确保关键操作的原子性，我们使用分布式锁，主要采用Redis的分布式锁实现。这可以确保在同一时刻只有一个请求能够执行关键操作，防止多个用户并发执行导致的问题。”
“此外，通过将用户抢购请求放入消息队列，保证抢购的顺序。在消息队列中使用分布式锁来确保同一时刻只有一个消息能够被消费，从而保证订单生成的有序性。”
“最后，为了避免某个用户通过高频请求导致超卖，我们使用Redis的计数器来记录用户的请求次数，并设置一个合理的抢购频率限制。”
- 重点突出分布式锁的设计，包括锁的获取和释放机制、超时处理等。
示例回答：
锁的获取机制
在分布式环境中，为了确保同一时刻只有一个实例能够成功获取锁，我们使用了Redis的原子操作 setIfAbsent。这个操作是原子的，即在单个 Redis 命令中完成，可以确保在高并发情况下的互斥性。setIfAbsent会在键不存在的情况下设置键的值，如果键已经存在，那么该操作将不执行任何操作。
示例代码：
public boolean tryLock(String lockKey, String requestId, long expireTime) { Boolean locked = redisTemplate.opsForValue().setIfAbsent(lockKey, requestId, expireTime, TimeUnit.MILLISECONDS); return Boolean.TRUE.equals(locked); }
在这个方法中，lockKey 是锁的唯一标识，requestId 是请求的唯一标识（通常可以使用UUID）。expireTime 是锁的过期时间，确保在极端情况下锁会自动释放，避免死锁。
超时处理
合理设置锁的过期时间是非常重要的，过长可能导致资源长时间被占用，而过短可能在执行关键操作时锁已经被释放。这里，我们使用 expireTime 参数来设置锁的过期时间，通常使用毫秒为单位。
示例代码：
public boolean tryLock(String lockKey, String requestId, long expireTime) { Boolean locked = redisTemplate.opsForValue().setIfAbsent(lockKey, requestId, expireTime, TimeUnit.MILLISECONDS); return Boolean.TRUE.equals(locked); }
在上述代码中，expireTime 即为锁的过期时间，以毫秒为单位。合理设置这个值，可以避免因异常情况而导致的死锁，确保系统在正常情况下能够及时释放锁。
锁的释放机制
释放锁的时候，我们首先获取存储在 Redis 中的请求 ID，确保只有持有锁的实例才能释放锁。这一步是为了确保锁的互斥性，即同一时刻只有一个实例能够执行关键操作。
示例代码：
public void unlock(String lockKey, String requestId) { String storedRequestId = redisTemplate.opsForValue().get(lockKey); if (requestId.equals(storedRequestId)) { redisTemplate.delete(lockKey); } }
在这个方法中，我们首先通过 get 操作获取存储在 Redis 中的请求 ID。如果当前请求的 ID 与存储的 ID 相同，说明当前实例持有该锁，然后通过 delete 操作删除该锁。
