一 基础准备
1.1 Redis setnx 命令
命令语义：【set if not exsits】
格式：setnx key value
相应： 将 key 的值设为 value ，当且仅当 key 不存在。
若给定的 key 已经存在，则 SETNX 不做任何动作。
1.2 lua基本语法
二 redis分布式锁实现
假设我们有一个秒杀业务，固定的库存，要求不能超卖，即有库存就卖出，直到无库存为止。我们试着用redis存储库存，实现这个需求。
2.1 简易实现
代码
@RestController
@Slf4j
public class GoodStockController {
@Autowired
private StringRedisTemplate stringRedisTemplate;
@RequestMapping("/deduct_stock")
public String deductStock() {
//每次查询库存，如果库存大于0，则扣件一个库存，返回成功。
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); // jedis.get("stock")
if (stock > 0) {
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock", realStock + ""); // jedis.set(key,value)
log.info("扣减成功，剩余库存:" + realStock);
return "Success";
}
log.info("扣减失败，库存不足");
return "Failure";
}
}
问题
问题时显而易见的。 当多个线程并发访问时，前一线程刚读到的库存可能已经被其他线程修改，导致超买。
2.2 锁实现
用synchronized将该方法锁住，防止并发。
代码
public class GoodStockController {
@Autowired
private StringRedisTemplate stringRedisTemplate;
@RequestMapping("/deduct_stock")
public String deductStock() {
synchronized (this) {
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); // jedis.get("stock")
if (stock > 0) {
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock", realStock + ""); // jedis.set(key,value)
log.info("扣减成功，剩余库存:" + realStock);
return "Success";
}
log.info("扣减失败，库存不足");
return "Failure";
}
}
}
问题
用锁，只能保证当前实例线程安全，对于分布式系统，多个实例提供服务，这时，还是存在并发问题。 对于分布式系统，要解决并发问题，就需要——分布式锁。
2.3 Redis setnx实现1
利用 Redis setnx实现简单的分布式锁。
代码
@RestController
@Slf4j
public class GoodStockController {
@Autowired
private StringRedisTemplate stringRedisTemplate;
@RequestMapping("/deduct_stock")
public String deductStock() {
String lockKey = "stockLock";
String lockValue = "1";
Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, lockValue);
if (!lock) {
return "请稍后再试";
}
String result = "Success";
try{
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); // jedis.get("stock")
if (stock > 0) {
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock", realStock + ""); // jedis.set(key,value)
log.info("扣减成功，剩余库存:" + realStock);
} else {
log.info("扣减失败，库存不足");
result = "Failure";
}
}finally {
stringRedisTemplate.delete(lockKey);
}
return result;
}
}
问题
这样实现，看起来确实解决了系统超卖问题。但是如果加锁完成后，系统宕机了，会造成一致锁一致无法释放，商品卖不出了。所以需要给锁加一个过期时间。
2.4 Redis setnx实现2
利用原子化操作，给锁设置过期时间。
代码
@RestController
@Slf4j
public class GoodStockController {
@Autowired
private StringRedisTemplate stringRedisTemplate;
@RequestMapping("/deduct_stock")
public String deductStock() {
String lockKey = "stockLock";
String lockValue = "1";
//Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, lockValue,10, TimeUnit.SECONDS);
//stringRedisTemplate.expire(lockKey,10, TimeUnit.SECONDS);
Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, lockValue,10, TimeUnit.SECONDS);
if (!lock) {
return "请稍后再试";
}
String result = "Success";
try{
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); // jedis.get("stock")
if (stock > 0) {
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock", realStock + ""); // jedis.set(key,value)
log.info("扣减成功，剩余库存:" + realStock);
} else {
log.info("扣减失败，库存不足");
result = "Failure";
}
}finally {
stringRedisTemplate.delete(lockKey);
}
return result;
}
}
代码中，设置锁过期时间是用的setnx的重载方法，保证了设置值和过期时间为一个原子操作。我们设置了10s的到期时间，看起来没有问题了。 实际上，在极端情况下还是存在问题可能导致超卖。
三个线程A和B，C。
- 由于网络波动，A在执行完扣件库存操作，删除锁之前，锁到期了
- 此时,线程B正好加锁成功。
- 然后线程A开始删除锁，把B家的锁删除了。
- 此时线程c开始进入加锁流程，因为B的线程被A删除了，导致C加锁成功，B执行完又把C的锁删除了。
这种场景下，线程删除了不属于自己的锁。那么，我们需要保证每个线程删除的是自己的锁。
2.4 Redis setnx实现3
给每个锁设置唯一的值
代码
@RestController
@Slf4j
public class GoodStockController {
@Autowired
private StringRedisTemplate stringRedisTemplate;
@RequestMapping("/deduct_stock")
public String deductStock() {
String lockKey = "stockLock";
String lockValue = UUID.randomUUID().toString();
// Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, lockValue,10, TimeUnit.SECONDS);
// stringRedisTemplate.expire(lockKey,10, TimeUnit.SECONDS);
Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, lockValue, 10, TimeUnit.SECONDS);
if (!lock) {
return "请稍后再试";
}
String result = "Success";
try {
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); // jedis.get("stock")
if (stock > 0) {
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock", realStock + ""); // jedis.set(key,value)
log.info("扣减成功，剩余库存:" + realStock);
} else {
log.info("扣减失败，库存不足");
result = "Failure";
}
} finally {
if (Objects.equals(stringRedisTemplate.opsForValue().get(lockKey), lockValue))
stringRedisTemplate.delete(lockKey);
}
return result;
}
}
问题
我们在删除锁之前判断了当前锁是否是本线程加的锁，似乎解决了删除其他锁的问题，实际上，很多公司也确实是这样实现的。但是极端情况下还是会有问题， 因为判断代码和删除代码非原子操作。 这个问题的经典解决方案——锁续命。
2.4 关于锁续命
锁续命，实际上是通过另一个线程监听锁状态，在锁到期之前，给锁续命。这个方案看起来简单，但真正实现起来，并不简单。 另外，例子中，如果加锁失败，直接返回操作失败。如果要实现阻塞，通知功能，还是很复杂的。
市面上其实已经有经过实践证明的redis分布式锁实现方案。如 Redisson。
2.5 Redisson实现分布式锁
Redisson 不仅实现了锁续命，还优雅的实现了高并发场景下的阻塞，通知，唤醒。
-- 引入一个RedissonBean
@Bean
public Redisson redisson() {
// 此为单机模式
Config config = new Config();
config.useSingleServer().setAddress("redis://localhost:6379").setDatabase(0);
return (Redisson) Redisson.create(config);
}
RestController
@Slf4j
public class GoodStockController {
@Autowired
private StringRedisTemplate stringRedisTemplate;
@Autowired
private Redisson redisson;
@RequestMapping("/deduct_stock")
public String deductStock() {
String lockKey = "stockLock";
RLock redissonLock = redisson.getLock(lockKey);
//加分布式锁
redissonLock.lock();
String result = "Success";
try {
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); // jedis.get("stock")
if (stock > 0) {
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock", realStock + ""); // jedis.set(key,value)
log.info("扣减成功，剩余库存:" + realStock);
} else {
log.info("扣减失败，库存不足");
result = "Failure";
}
} finally {
redissonLock.unlock();
}
return result;
}
}
