redis分布式鎖的問題與解決方法

更新時間：2019年07月21日 10:16:14 作者：付威的網(wǎng)絡博客

這篇文章主要給大家介紹了關于redis分布式鎖的問題與解決方法，文中通過示例代碼介紹的非常詳細，對大家學習或者使用redis具有一定的參考學習價值，需要的朋友們下面來一起學習學習吧

分布式鎖

在分布式環(huán)境中，為了保證業(yè)務數(shù)據(jù)的正常訪問，防止出現(xiàn)重復請求的問題，會使用分布式鎖來阻攔后續(xù)請求。我們先寫一段有問題的業(yè)務代碼：

public void doSomething(String userId){
  User user=getUser(userId);
  if(user==null){
   user.setUserName("xxxxx");
   user.setUserId(userId);
   insert(user);
   return;
  }
  update(user);
 }

上面的代碼很簡單，查詢db中有沒有對應的user數(shù)據(jù)，如果有的話，執(zhí)行更新操作，如果沒有則插入。

我們知道，上面的代碼是線程不安全的，在多線程的環(huán)境中，就會出現(xiàn)問題。為了能夠保證數(shù)據(jù)的正確性，在單機環(huán)境下，我們可以使用synchronized的方法，來保證線程安全，具體修改：

public synchronized void doSomething(String userId){
  User user=getUser(userId);
  if(user==null){
   user.setUserName("xxxxx");
   user.setUserId(userId);
   insert(user);
   return;
  }
  update(user);
 }

在單機器的環(huán)境下，能夠解決線程安全的問題，那在分布式環(huán)境下呢？這個時候需要用到分布式鎖.

分布式鎖需要借助其他組件來實現(xiàn)，常用的有redis和zookeeper。下面我們就用redis的實現(xiàn)，來說明下問題，分布式鎖具體的實現(xiàn)方法如下

public void doSomething(String userId){
  String lock=RedisUtils.get("xxxx"+userId);
  if(StringUtils.isNotEmpty(lock)){//說明當前userId已經(jīng)被鎖定
   return;
  }
  RedisUtils.set("xxxx"+userId,userId,1000);//鎖定10s
  User user=getUser(userId);
  if(user==null){
   insert(user);
   RedisUtils.delete("xxxx"+userId);
   return;
  }
  update(user);
  RedisUtils.delete("xxxx"+userId);
  
 }

上面的代碼解決了在分布式環(huán)境中的并發(fā)的問題。但同樣需要考慮一個問題，如果insert操作和update操作異常了，分布式鎖不會釋放，后續(xù)的請求還會被攔截。

所以我們再優(yōu)化，增加對異常的捕獲。

public void doSomething(String userId){
  try {
    String lock=RedisUtils.get("xxxx"+userId);
    if(StringUtils.isNotEmpty(lock)){//說明當前userId已經(jīng)被鎖定
     return;
    }
    RedisUtils.set("xxxx"+userId,userId,1000);//鎖定1s
    User user=getUser(userId);
    if(user==null){
     insert(user);
     return;
    }
    update(user);
  }
  catch(Exception ex){

  }
  finally{
   RedisUtils.delete("xxxx"+userId);
  }
 }

現(xiàn)在即使是程序異常了，鎖會自動釋放。但redis的get和set也會存在并發(fā)問題，我們再繼續(xù)優(yōu)化，使用redis中的setnx方法

public void doSomething(String userId){
  try {
    boolean lock=RedisUtils.setnx("xxxx"+userId,userId,1000);//鎖定1s
    if(!lock){//說明當前userId已經(jīng)被鎖定
     return;
    }
    User user=getUser(userId);
    if(user==null){
     insert(user);
     return;
    }
    update(user);
  }
  catch(Exception ex){

  }
  finally{
   RedisUtils.delete("xxxx"+userId);
  }
 }

上面的代碼好像沒有什么問題了，但也存在很大的隱患。我們分析下，假設第一個請求過來，執(zhí)行鎖定成功，程序開始運行，但是insert和update操作阻塞了1s，第二個請求過來，鎖的緩存已經(jīng)過期，第二個執(zhí)行鎖定成功，這個時候第一個請求完成了鎖被釋放，第二個請求的鎖就被第一次請求釋放了，第三次的請求就會造成線程不安全問題。

怎么再去優(yōu)化呢？問題主要是出現(xiàn)在第一次請求誤刪鎖的問題，所以我們在移除鎖的時候要判斷能否移除。

思路：我們在鎖定的時候，value使用當前的時間戳，刪除時判斷是否過期如果不過期就不要刪除，具體代碼如下：

public void doSomething(String userId){
  try {
    boolean lock=RedisUtils.setnx("xxxx"+userId,LocalDateTime.now(),1000);//鎖定10s
    if(!lock){//說明當前userId已經(jīng)被鎖定
     return;
    }
    User user=getUser(userId);
    if(user==null){
     insert(user);
     return;
    }
    update(user);
  }
  catch(Exception ex){

  }
  finally{
   LocalDateTime lockTIme= RedisUtils.get("xxxx"+userId);
   if(lockTIme.compare(LocalDateTime.now())<0){
    //說明已經(jīng)過期，可以刪除key
    RedisUtils.delete("xxxx"+userId);
   }
  }
 }

這樣即使出現(xiàn)阻塞，第二次的時間戳覆蓋了第一次的鎖定，這樣即使第一次完成了，也不會釋放鎖。

總結(jié)

以上就是這篇文章的全部內(nèi)容了，希望本文的內(nèi)容對大家的學習或者工作具有一定的參考學習價值，謝謝大家對腳本之家的支持。

您可能感興趣的文章: