模擬一個電商裡面下單減庫存的場景。

1.首先在redis里加入商品庫存數量。

2.新建一個Spring Boot專案，在pom裡面引入相關的依賴。

 <dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-web</artifactId>
 </dependency>

 <dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-data-redis</artifactId>
 </dependency>

3.接下來，在application.yml配置redis屬性和指定應用的埠號：

server:
 port: 8090

spring:
 redis:
 host: 192.168.0.60
 port: 6379

4.新建一個Controller類，扣減庫存第一版程式碼：

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.Objects;

@RestController
public class StockController {

 private static final Logger logger = LoggerFactory.getLogger(StockController.class);

 @Resource
 private StringRedisTemplate stringRedisTemplate;

 @RequestMapping("/reduceStock")
 public String reduceStock() {
 // 從redis中獲取庫存數量
 int stock = Integer.parseInt(Objects.requireNonNull(stringRedisTemplate.opsForValue().get("stockCount")));
 if (stock > 0) {
  // 減庫存
  int restStock = stock - 1;
  // 剩餘庫存再重新設定到redis中
  stringRedisTemplate.opsForValue().set("stockCount",String.valueOf(restStock));
  logger.info("扣減成功，剩餘庫存：{}",restStock);
 } else {
  logger.info("庫存不足，扣減失敗。");
 }

 return "success";
 }
}

上面第一版的程式碼存在什麼問題：超賣。假如多個執行緒同時呼叫獲取庫存數量的程式碼，那麼每個執行緒拿到的都是100，判斷庫存都大於0，都可以執行減庫存的操作。假如兩個執行緒都做減庫存更新快取，那麼快取的庫存變成99，但實際上，應該是減掉2個庫存。

那麼很多人的第一個想法是加synchronized同步程式碼塊，因為獲取數量和減庫存不是原子性操作，有多個執行緒來執行程式碼的時候，只允許一個執行緒執行程式碼塊裡的程式碼。那麼改完的第二版的程式碼如下：

 @RequestMapping("/reduceStock")
 public String reduceStock() {
 synchronized (this) {
  // 從redis中獲取庫存數量
  int stock = Integer.parseInt(Objects.requireNonNull(stringRedisTemplate.opsForValue().get("stockCount")));
  if (stock > 0) {
  // 減庫存
  int restStock = stock - 1;
  // 剩餘庫存再重新設定到redis中
  stringRedisTemplate.opsForValue().set("stockCount",restStock);
  } else {
  logger.info("庫存不足，扣減失敗。");
  }
 }

 return "success";
 }

但使用synchronize存在的問題，就是隻能保證單機環境執行時沒有問題的。但現在的軟體公司裡，基本上都是叢集架構，是多例項，前面使用Nginx做負載均衡，大概架構如下：

Nginx分發請求，把請求傳送到不同的Tomcat容器，而synchronize只能保證一個應用是沒有問題的。

那麼程式碼改進第三版，就是引入redis分散式鎖，具體程式碼如下：

 @RequestMapping("/reduceStock")
 public String reduceStock() {
 String lockKey = "stockKey";
 try {
  boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey,"1");
  if (!result) {
  return "errorCode";
  }
  // 從redis中獲取庫存數量
  int stock = Integer.parseInt(Objects.requireNonNull(stringRedisTemplate.opsForValue().get("stockCount")));
  if (stock > 0) {
  // 減庫存
  int restStock = stock - 1;
  // 剩餘庫存再重新設定到redis中
  stringRedisTemplate.opsForValue().set("stockCount",restStock);
  } else {
  logger.info("庫存不足，扣減失敗。");
  }
 } finally {
  stringRedisTemplate.delete(lockKey)
 }
 return "success";
 }

如果有一個執行緒拿到鎖，那麼其他的執行緒就會等待。一定要記得在finally裡面把使用完的鎖要刪除掉。否則一旦丟擲異常，只有一個執行緒會一直持有鎖，其他執行緒沒有機會獲取。

但如果在執行if (stock > 0) {程式碼塊裡的程式碼，因為宕機或重啟沒有執行完，也會一直持有鎖，所以，這裡需要把鎖加一個超時時間：

 boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey,"1");
 stringRedisTemplate.expire(lockKey,10,TimeUnit.SECONDS);

但如果上面兩行程式碼在中間執行出問題了，設定超時時間的程式碼還沒執行，也會出現鎖不能釋放的問題。好在有對應的方法：就是把上面兩行程式碼設定成一個原子操作：

 // 這裡預設設定超時時間為10秒
 boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey,"1",TimeUnit.SECONDS);

到此為止，如果併發量不是很大的話，基本上是沒有問題的。

但是，如果請求的併發量很大，就會出現新的問題：有種比較特殊的情況，第一個執行緒執行了15秒，但是執行到10秒鐘的時候，鎖已經失效釋放了，那麼在高併發場景下，第二個執行緒發現鎖已經失效，那麼它就可以拿到這把鎖進行加鎖，
假設第二個執行緒執行需要8秒，它執行到5秒鐘後，此時第一個執行緒已經執行完了，執行完那一刻，進行了刪除key的操作，但是此時的鎖是第二個執行緒加的，這樣第一個執行緒把第二個執行緒加的鎖刪掉了。

那意味著第三個執行緒又可以拿到鎖，第三個執行緒執行了3秒鐘，此時第二個執行緒執行完畢，那麼第二個執行緒把第三個執行緒的鎖又刪除了。導致鎖失效。

那麼解決的思路就是，我自己加的鎖，不要被別人刪掉。那麼可以為每個進來的請求生成一個唯一的id，作為分散式鎖的值，然後在釋放時，判斷一下當前執行緒的id，是不是和快取裡的id是否相等。

 @RequestMapping("/reduceStock")
 public String reduceStock() {
 String lockKey = "stockKey";
 String id = UUID.randomUUID().toString();
 try {
  // 這裡預設設定超時時間為30秒
  boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey,id,30,TimeUnit.SECONDS);
  if (!result) {
  return "errorCode";
  }
  // 從redis中獲取庫存數量
  int stock = Integer.parseInt(Objects.requireNonNull(stringRedisTemplate.opsForValue().get("stockCount")));
  if (stock > 0) {
  // 減庫存
  int restStock = stock - 1;
  // 剩餘庫存再重新設定到redis中
  stringRedisTemplate.opsForValue().set("stockCount",restStock);
  } else {
  logger.info("庫存不足，扣減失敗。");
  }
 } finally {
  if (id.contentEquals(Objects.requireNonNull(stringRedisTemplate.opsForValue().get(lockKey)))) {
  stringRedisTemplate.delete(lockKey);
  }
 }
 return "success";
 }

到此為止，一個比較完善的鎖就實現了，可以應付大部分場景。
當然，上面的程式碼還有一個問題，就是一個執行緒執行時間超過了過期時間，後面的程式碼還沒有執行完，鎖就已經刪除了，還是會有些bug存在。解決的方法是給鎖續命的操作。
在當前主執行緒獲取到鎖以後，可以fork出一個執行緒，執行Timer定時器操作，假如預設超時時間為30秒，那麼定時器每隔10秒去看下這把鎖還是否存在，存在就說明這個鎖裡的邏輯還沒有執行完，那麼就可以把當前主執行緒的超時時間重新設定為30秒；如果不存在，就直接結束掉。

但是上面的邏輯，在高併發場景下，實現比較完善還是比較困難的。好在現在已經有比較成熟的框架，那就是Redisson。官方地址https://redisson.org。

下面用Redisson來實現分散式鎖。

首先引入依賴包：

  <dependency>
   <groupId>org.redisson</groupId>
   <artifactId>redisson</artifactId>
   <version>3.6.5</version>
  </dependency>

配置類：

@Configuration
public class RedissonConfig {
 @Bean
 public Redisson redisson() {
  // 單機模式
  Config config = new Config();
  config.useSingleServer().setAddress("redis://192.168.0.60:6379").setDatabase(0);
  return (Redisson) Redisson.create(config);
 }
}

接下來用redisson重寫上面的減庫存操作：

 @Resource
 private Redisson redisson;
 
 @RequestMapping("/reduceStock")
 public String reduceStock() {
  String lockKey = "stockKey";
  RLock redissonLock = redisson.getLock(lockKey);
  try {
   // 加鎖，鎖續命
   redissonLock.lock();
   // 從redis中獲取庫存數量
   int stock = Integer.parseInt(Objects.requireNonNull(stringRedisTemplate.opsForValue().get("stockCount")));
   if (stock > 0) {
    // 減庫存
    int restStock = stock - 1;
    // 剩餘庫存再重新設定到redis中
    stringRedisTemplate.opsForValue().set("stockCount",String.valueOf(restStock));
    logger.info("扣減成功，剩餘庫存：{}",restStock);
   } else {
    logger.info("庫存不足，扣減失敗。");
   }
  } finally {
   redissonLock.unlock();
  }
  return "success";
 }

其實就是三個步驟：獲取鎖，加鎖，釋放鎖。

先簡單看下Redisson的實現原理：

這裡先說一下Redis很多操作使用Lua指令碼來實現原子性操作，關於Lua語法，可以去網上找下相關教程。
使用Lua指令碼的好處有：

1.減少網路開銷，多個命令可以使用一次請求完成；

2.實現了原子性操作，Redis會把Lua指令碼作為一個整體去執行；

3.實現事務，Redis自帶的事務功能有限，而Lua指令碼實現了事務的常規操作，而且還支援回滾。

但是Lua實際上不會使用很多，如果Lua指令碼執行時間過長，因為Redis是單執行緒，因此會導致堵塞。

最後，說下Redisson分散式鎖的程式碼實現，
找到上面的redissonLock.lock()；
lock方法點進去，一直點到RedissonLock類裡面的lockInterruptibly方法：

 @Override
 public void lockInterruptibly(long leaseTime,TimeUnit unit) throws InterruptedException {
  // 獲取執行緒id
  long threadId = Thread.currentThread().getId();
  Long ttl = tryAcquire(leaseTime,unit,threadId);
  // lock acquired
  if (ttl == null) {
   return;
  }

  RFuture<RedissonLockEntry> future = subscribe(threadId);
  commandExecutor.syncSubscription(future);

  try {
   while (true) {
    ttl = tryAcquire(leaseTime,threadId);
    // lock acquired
    if (ttl == null) {
     break;
    }

    // waiting for message
    if (ttl >= 0) {
     getEntry(threadId).getLatch().tryAcquire(ttl,TimeUnit.MILLISECONDS);
    } else {
     getEntry(threadId).getLatch().acquire();
    }
   }
  } finally {
   unsubscribe(future,threadId);
  }
//  get(lockAsync(leaseTime,unit));
 }

重點看下tryAcquire方法，把執行緒id作為一個引數傳遞進來，在這個方法裡面，找到tryLockInnerAsync方法點進去，

 <T> RFuture<T> tryLockInnerAsync(long leaseTime,TimeUnit unit,long threadId,RedisStrictCommand<T> command) {
  internalLockLeaseTime = unit.toMillis(leaseTime);

  return commandExecutor.evalWriteAsync(getName(),LongCodec.INSTANCE,command,"if (redis.call('exists',KEYS[1]) == 0) then " +
      "redis.call('hset',KEYS[1],ARGV[2],1); " +
      "redis.call('pexpire',ARGV[1]); " +
      "return nil; " +
     "end; " +
     "if (redis.call('hexists',ARGV[2]) == 1) then " +
      "redis.call('hincrby',ARGV[1]); " +
      "return nil; " +
     "end; " +
     "return redis.call('pttl',KEYS[1]);",Collections.<Object>singletonList(getName()),internalLockLeaseTime,getLockName(threadId));
 }

這裡就是一堆Lua指令碼，先看第一個if命令，先去判斷 KEYS[1]（就是對應的鎖key的名字），如果不存在，在hashmap裡，設定一個屬性為執行緒id，值為1，再把map的過期時間設定為internalLockLeaseTime，這個值預設是30秒，

上面的操作對應的命令是：

hset keyname id:thread 1
pexpire keyname 30

然後返回nil，相當於null，那程式return了。

另外，Redisson還支援重入鎖，那第二個if就是執行重入鎖的操作，會判斷鎖是否存在，並且傳入的執行緒id是否是當前執行緒的id，若果是，支援重複加鎖進行自增操作；

如果是其他執行緒呼叫lock方法，上面兩個if判斷不會走，會返回鎖剩餘過期時間。

接著返回到tryAcquireAsync方法裡面往下看：

實際上是加了一個監聽器，在監聽器裡面有個很重要的方法scheduleExpirationRenewal，一看這個名字就能大概猜出是什麼功能，

裡面有個定時任務的輪詢，

private void scheduleExpirationRenewal(final long threadId) {
  if (expirationRenewalMap.containsKey(getEntryName())) {
   return;
  }

  Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
   @Override
   public void run(Timeout timeout) throws Exception {
    // 判斷傳遞進來的執行緒id是否是我們之前主執行緒設定的id，如果是，則增加續命，增加30秒。
    RFuture<Boolean> future = commandExecutor.evalWriteAsync(getName(),RedisCommands.EVAL_BOOLEAN,"if (redis.call('hexists',ARGV[2]) == 1) then " +
       "redis.call('pexpire',ARGV[1]); " +
       "return 1; " +
      "end; " +
      "return 0;",getLockName(threadId));
    
    future.addListener(new FutureListener<Boolean>() {
     @Override
     public void operationComplete(Future<Boolean> future) throws Exception {
      expirationRenewalMap.remove(getEntryName());
      if (!future.isSuccess()) {
       log.error("Can't update lock " + getName() + " expiration",future.cause());
       return;
      }
      
      if (future.getNow()) {
       // reschedule itself
       scheduleExpirationRenewal(threadId);
      }
     }
    });
   }
  },internalLockLeaseTime / 3,TimeUnit.MILLISECONDS);

  if (expirationRenewalMap.putIfAbsent(getEntryName(),task) != null) {
   task.cancel();
  }
 }

接著推遲10秒鐘（internalLockLeaseTime / 3），再執行續命操作邏輯。

到最後，再回到lockInterruptibly方法，如果ttl 為null，說明加鎖成功了，就返回null，那如果其他執行緒的話，就會返回剩餘過期時間，那麼就會進入到while死迴圈裡，一直嘗試加鎖，呼叫tryAcquire方法，在瑣失效以後，再會嘗試獲取加鎖。

到此為止，分析完畢。

總結

到此這篇關於Redis分散式鎖的使用和實現原理的文章就介紹到這了,更多相關Redis分散式鎖的使用和原理內容請搜尋我們以前的文章或繼續瀏覽下面的相關文章希望大家以後多多支援我們！