分布式鎖
在分布式環(huán)境中，為了保證業(yè)務(wù)數(shù)據(jù)的正常訪問，防止出現(xiàn)重復(fù)請(qǐng)求的問題，會(huì)使用分布式鎖來阻攔后續(xù)請(qǐng)求。我們先寫一段有問題的業(yè)務(wù)代碼：
public void doSomething(String userId){ User user=getUser(userId); if(user==null){ user.setUserName("xxxxx"); user.setUserId(userId); insert(user); return; } update(user); }上面的代碼很簡單，查詢db中有沒有對(duì)應(yīng)的user數(shù)據(jù)，如果有的話，執(zhí)行更新操作，如果沒有則插入。
我們知道，上面的代碼是線程不安全的，在多線程的環(huán)境中，就會(huì)出現(xiàn)問題。為了能夠保證數(shù)據(jù)的正確性，在單機(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); }在單機(jī)器的環(huán)境下，能夠解決線程安全的問題，那在分布式環(huán)境下呢？ 這個(gè)時(shí)候需要用到分布式鎖.
分布式鎖需要借助其他組件來實(shí)現(xiàn)，常用的有redis和zookeeper。下面我們就用redis的實(shí)現(xiàn)，來說明下問題，分布式鎖具體的實(shí)現(xiàn)方法如下
public void doSomething(String userId){ String lock=RedisUtils.get("xxxx"+userId); if(StringUtils.isNotEmpty(lock)){//說明當(dāng)前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ā)的問題。但同樣需要考慮一個(gè)問題，如果insert操作和update操作異常了，分布式鎖不會(huì)釋放，后續(xù)的請(qǐng)求還會(huì)被攔截。
所以我們?cè)賰?yōu)化，增加對(duì)異常的捕獲。
public void doSomething(String userId){ try { String lock=RedisUtils.get("xxxx"+userId); if(StringUtils.isNotEmpty(lock)){//說明當(dāng)前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)在即使是程序異常了，鎖會(huì)自動(dòng)釋放。但redis的get和set也會(huì)存在并發(fā)問題，我們?cè)倮^續(xù)優(yōu)化，使用redis中的setnx方法
public void doSomething(String userId){ try { boolean lock=RedisUtils.setnx("xxxx"+userId,userId,1000);//鎖定1s if(!lock){//說明當(dāng)前userId已經(jīng)被鎖定 return; } User user=getUser(userId); if(user==null){ insert(user); return; } update(user); } catch(Exception ex){ } finally{ RedisUtils.delete("xxxx"+userId); } }上面的代碼好像沒有什么問題了，但也存在很大的隱患。 我們分析下，假設(shè)第一個(gè)請(qǐng)求過來，執(zhí)行鎖定成功，程序開始運(yùn)行，但是insert和update操作阻塞了1s，第二個(gè)請(qǐng)求過來，鎖的緩存已經(jīng)過期，第二個(gè)執(zhí)行鎖定成功，這個(gè)時(shí)候第一個(gè)請(qǐng)求完成了鎖被釋放，第二個(gè)請(qǐng)求的鎖就被第一次請(qǐng)求釋放了，第三次的請(qǐng)求就會(huì)造成線程不安全問題。
怎么再去優(yōu)化呢？問題主要是出現(xiàn)在第一次請(qǐng)求誤刪鎖的問題，所以我們?cè)谝瞥i的時(shí)候要判斷能否移除。
思路：我們?cè)阪i定的時(shí)候，value使用當(dāng)前的時(shí)間戳，刪除時(shí)判斷是否過期如果不過期就不要?jiǎng)h除，具體代碼如下：
public void doSomething(String userId){ try { boolean lock=RedisUtils.setnx("xxxx"+userId,LocalDateTime.now(),1000);//鎖定10s if(!lock){//說明當(dāng)前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)阻塞，第二次的時(shí)間戳覆蓋了第一次的鎖定，這樣即使第一次完成了，也不會(huì)釋放鎖。
總結(jié)
以上就是這篇文章的全部內(nèi)容了，希望本文的內(nèi)容對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，謝謝大家對(duì)武林網(wǎng)的支持。
新聞熱點(diǎn)
疑難解答
圖片精選
