日韩无码专区无码一级三级片|91人人爱网站中日韩无码电影|厨房大战丰满熟妇|AV高清无码在线免费观看|另类AV日韩少妇熟女|中文日本大黄一级黄色片|色情在线视频免费|亚洲成人特黄a片|黄片wwwav色图欧美|欧亚乱色一区二区三区

RELATEED CONSULTING
相關(guān)咨詢
選擇下列產(chǎn)品馬上在線溝通
服務(wù)時(shí)間：8:30-17:00
你可能遇到了下面的問題
關(guān)閉右側(cè)工具欄

新聞中心

這里有您想知道的互聯(lián)網(wǎng)營銷解決方案
分布式鎖看了又看，優(yōu)秀方案我來告訴你

分布式鎖看了又看，優(yōu)秀方案我來告訴你

作者：老鄭 2021-04-12 08:02:12

開發(fā)

前端

分布式 對(duì)于商品秒殺的場(chǎng)景，我們需要防止庫存超賣或者重復(fù)扣款等并發(fā)問題，我們通常需要使用分布式鎖，來解決共享資源競爭導(dǎo)致數(shù)據(jù)不一致的問題。本篇就講解如何用分布式鎖的來解決此類問題。

成都創(chuàng)新互聯(lián)是專業(yè)的豐滿網(wǎng)站建設(shè)公司，豐滿接單;提供網(wǎng)站設(shè)計(jì)、成都網(wǎng)站制作,網(wǎng)頁設(shè)計(jì),網(wǎng)站設(shè)計(jì),建網(wǎng)站,PHP網(wǎng)站建設(shè)等專業(yè)做網(wǎng)站服務(wù);采用PHP框架,可快速的進(jìn)行豐滿網(wǎng)站開發(fā)網(wǎng)頁制作和功能擴(kuò)展;專業(yè)做搜索引擎喜愛的網(wǎng)站,專業(yè)的做網(wǎng)站團(tuán)隊(duì),希望更多企業(yè)前來合作!

[[392389]]

分布式鎖的場(chǎng)景

秒殺場(chǎng)景案例

對(duì)于商品秒殺的場(chǎng)景，我們需要防止庫存超賣或者重復(fù)扣款等并發(fā)問題，我們通常需要使用分布式鎖，來解決共享資源競爭導(dǎo)致數(shù)據(jù)不一致的問題。

以手機(jī)秒殺的場(chǎng)景為例子，在搶購的過程中通常我們有三個(gè)步驟：

扣掉對(duì)應(yīng)商品的庫存;2. 創(chuàng)建商品的訂單;3. 用戶支付。

對(duì)于這樣的場(chǎng)景我們就可以采用分布式鎖的來解決，比如我們?cè)谟脩暨M(jìn)入秒殺 “下單“ 鏈接的過程中，我們可以對(duì)商品庫存進(jìn)行加鎖，然后完成扣庫存和其他操作，操作完成后。釋放鎖，讓下一個(gè)用戶繼續(xù)進(jìn)入保證庫存的安全性;也可以減少因?yàn)槊霘⑹?，?dǎo)致 DB 回滾的次數(shù)。整個(gè)流程如下圖所示：

注：對(duì)于鎖的粒度要根據(jù)具體的場(chǎng)景和需求來權(quán)衡。

三種分布式鎖

對(duì)于 Zookeeper 的分布式鎖實(shí)現(xiàn)，主要是利用 Zookeeper 的兩個(gè)特征來實(shí)現(xiàn)：

  1. Zookeeper 的一個(gè)節(jié)點(diǎn)不能被重復(fù)創(chuàng)建
  2. Zookeeper 的 Watcher 監(jiān)聽機(jī)制

非公平鎖

對(duì)于非公平鎖，我們?cè)诩渔i的過程如下圖所示。

優(yōu)點(diǎn)和缺點(diǎn)

其實(shí)上面的實(shí)現(xiàn)有優(yōu)點(diǎn)也有缺點(diǎn)：

優(yōu)點(diǎn)：

實(shí)現(xiàn)比較簡單，有通知機(jī)制，能提供較快的響應(yīng)，有點(diǎn)類似 ReentrantLock 的思想，對(duì)于節(jié)點(diǎn)刪除失敗的場(chǎng)景由 Session 超時(shí)保證節(jié)點(diǎn)能夠刪除掉。

缺點(diǎn)：

重量級(jí)，同時(shí)在大量鎖的情況下會(huì)有 “驚群” 的問題。

“驚群” 就是在一個(gè)節(jié)點(diǎn)刪除的時(shí)候，大量對(duì)這個(gè)節(jié)點(diǎn)的刪除動(dòng)作有訂閱 Watcher 的線程會(huì)進(jìn)行回調(diào)，這對(duì)Zk集群是十分不利的。所以需要避免這種現(xiàn)象的發(fā)生。

解決“驚群”：

為了解決“驚群“問題，我們需要放棄訂閱一個(gè)節(jié)點(diǎn)的策略，那么怎么做呢?

  1. 我們將鎖抽象成目錄，多個(gè)線程在此目錄下創(chuàng)建瞬時(shí)的順序節(jié)點(diǎn)，因?yàn)?Zookeeper 會(huì)為我們保證節(jié)點(diǎn)的順序性，所以可以利用節(jié)點(diǎn)的順序進(jìn)行鎖的判斷。
  2. 首先創(chuàng)建順序節(jié)點(diǎn)，然后獲取當(dāng)前目錄下最小的節(jié)點(diǎn)，判斷最小節(jié)點(diǎn)是不是當(dāng)前節(jié)點(diǎn)，如果是那么獲取鎖成功，如果不是那么獲取鎖失敗。
  3. 獲取鎖失敗的節(jié)點(diǎn)獲取當(dāng)前節(jié)點(diǎn)上一個(gè)順序節(jié)點(diǎn)，對(duì)此節(jié)點(diǎn)注冊(cè)監(jiān)聽，當(dāng)節(jié)點(diǎn)刪除的時(shí)候通知當(dāng)前節(jié)點(diǎn)。
  4. 當(dāng)unlock的時(shí)候刪除節(jié)點(diǎn)之后會(huì)通知下一個(gè)節(jié)點(diǎn)。

公平鎖

基于非公平鎖的缺點(diǎn)，我們可以通過一下的方案來規(guī)避。

優(yōu)點(diǎn)和缺點(diǎn)

優(yōu)點(diǎn)： 如上借助于臨時(shí)順序節(jié)點(diǎn)，可以避免同時(shí)多個(gè)節(jié)點(diǎn)的并發(fā)競爭鎖，緩解了服務(wù)端壓力。

缺點(diǎn)： 對(duì)于讀寫場(chǎng)景來說，無法解決一致性的問題，如果讀的時(shí)候也去獲取鎖的話，這樣會(huì)導(dǎo)致性能下降，對(duì)于這樣的問題，我們可以通過讀寫鎖來實(shí)現(xiàn)如類似 jdk 中的 ReadWriteLock

讀寫鎖實(shí)現(xiàn)

對(duì)于讀寫鎖的特點(diǎn)：讀寫鎖在如果多個(gè)線程都是在讀的時(shí)候，是可以并發(fā)讀的，就是一個(gè)無鎖的狀態(tài)，如果有寫鎖正在操作的時(shí)候，那么讀鎖需要等待寫鎖。在加寫鎖的時(shí)候，由于前面的讀鎖都是并發(fā)，所以需要監(jiān)聽最后一個(gè)讀鎖完成后執(zhí)行寫鎖。步驟如下：

  1. read 請(qǐng)求， 如果前面是讀鎖，可以直接讀取，不需要監(jiān)聽。如果前面是一個(gè)或者多個(gè)寫鎖那么只需要監(jiān)聽最后一個(gè)寫鎖。
  2. write 請(qǐng)求，只需要對(duì)前面的節(jié)點(diǎn)監(jiān)聽。Watcher 機(jī)制和互斥鎖一樣。

分布式鎖實(shí)戰(zhàn)

本文源碼中使用環(huán)境：JDK 1.8 、Zookeeper 3.6.x

Curator 組件實(shí)現(xiàn)

POM 依賴

  
 
 
 
    
   
  
  
  
  1.  
    2. 
   
  
  
  
  2.   org.apache.curator 
    3. 
   
  
  
  
  3.   curator-framework 
    4. 
   
  
  
  
  4.   2.13.0 
    5. 
   
  
  
  
  5.  
    6. 
   
  
  
  
  6.  
    7. 
   
  
  
  
  7.   org.apache.curator 
    8. 
   
  
  
  
  8.   curator-recipes 
    9. 
   
  
  
  
  9.   2.13.0 
    10. 
   
  
  
  
  10.  
    11.

互斥鎖運(yùn)用

由于 Zookeeper 非公平鎖的 “驚群” 效應(yīng)，非公平鎖在 Zookeeper 中其實(shí)并不是最好的選擇。下面是一個(gè)模擬秒殺的例子來使用 Zookeeper 分布式鎖。

  
 
 
 
    
   
  
  
  
  1. public class MutexTest { 
    2. 
   
  
  
  
  2.     static ExecutorService executor = Executors.newFixedThreadPool(8); 
    3. 
   
  
  
  
  3.     static AtomicInteger stock = new AtomicInteger(3); 
    4. 
   
  
  
  
  4.     public static void main(String[] args) throws InterruptedException { 
    5. 
   
  
  
  
  5.         CuratorFramework client = getZkClient(); 
    6. 
   
  
  
  
  6.         String key = "/lock/lockId_111/111"; 
    7. 
   
  
  
  
  7.         final InterProcessMutex mutex = new InterProcessMutex(client, key); 
    8. 
   
  
  
  
  8.         for (int i = 0; i < 99; i++) { 
    9. 
   
  
  
  
  9.             executor.submit(() -> { 
    10. 
   
  
  
  
  10.                 if (stock.get() < 0) { 
    11. 
   
  
  
  
  11.                     System.err.println("庫存不足, 直接返回"); 
    12. 
   
  
  
  
  12.                     return; 
    13. 
   
  
  
  
  13.                 } 
    14. 
   
  
  
  
  14.                 try { 
    15. 
   
  
  
  
  15.                     boolean acquire = mutex.acquire(200, TimeUnit.MILLISECONDS); 
    16. 
   
  
  
  
  16.                     if (acquire) { 
    17. 
   
  
  
  
  17.                         int s = stock.decrementAndGet(); 
    18. 
   
  
  
  
  18.                         if (s < 0) { 
    19. 
   
  
  
  
  19.                             System.err.println("進(jìn)入秒殺，庫存不足"); 
    20. 
   
  
  
  
  20.                         } else { 
    21. 
   
  
  
  
  21.                             System.out.println("購買成功, 剩余庫存: " + s); 
    22. 
   
  
  
  
  22.                         } 
    23. 
   
  
  
  
  23.                     } 
    24. 
   
  
  
  
  24.                 } catch (Exception e) { 
    25. 
   
  
  
  
  25.                     e.printStackTrace(); 
    26. 
   
  
  
  
  26.                 } finally { 
    27. 
   
  
  
  
  27.                     try { 
    28. 
   
  
  
  
  28.                         if (mutex.isAcquiredInThisProcess()) 
    29. 
   
  
  
  
  29.                             mutex.release(); 
    30. 
   
  
  
  
  30.                     } catch (Exception e) { 
    31. 
   
  
  
  
  31.                         e.printStackTrace(); 
    32. 
   
  
  
  
  32.                     } 
    33. 
   
  
  
  
  33.                 } 
    34. 
   
  
  
  
  34.             }); 
    35. 
   
  
  
  
  35.         } 
    36. 
   
  
  
  
  36.         while (true) { 
    37. 
   
  
  
  
  37.             if (executor.isTerminated()) { 
    38. 
   
  
  
  
  38.                 executor.shutdown(); 
    39. 
   
  
  
  
  39.                 System.out.println("秒殺完畢剩余庫存為:" + stock.get()); 
    40. 
   
  
  
  
  40.             } 
    41. 
   
  
  
  
  41.             TimeUnit.MILLISECONDS.sleep(100); 
    42. 
   
  
  
  
  42.         } 
    43. 
   
  
  
  
  43.     } 
    44. 
   
  
  
  
  44.     private static CuratorFramework getZkClient() { 
    45. 
   
  
  
  
  45.         String zkServerAddress = "127.0.0.1:2181"; 
    46. 
   
  
  
  
  46.         ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3, 5000); 
    47. 
   
  
  
  
  47.         CuratorFramework zkClient = CuratorFrameworkFactory.builder() 
    48. 
   
  
  
  
  48.                 .connectString(zkServerAddress) 
    49. 
   
  
  
  
  49.                 .sessionTimeoutMs(5000) 
    50. 
   
  
  
  
  50.                 .connectionTimeoutMs(5000) 
    51. 
   
  
  
  
  51.                 .retryPolicy(retryPolicy) 
    52. 
   
  
  
  
  52.                 .build(); 
    53. 
   
  
  
  
  53.         zkClient.start(); 
    54. 
   
  
  
  
  54.         return zkClient; 
    55. 
   
  
  
  
  55.     } 
    56. 
   
  
  
  
  56. } 
    57.

讀寫鎖運(yùn)用

讀寫鎖可以用來保證緩存雙寫的強(qiáng)一致性的，因?yàn)樽x寫鎖在多線程讀的時(shí)候是無鎖的， 只有在前面有寫鎖的時(shí)候才會(huì)等待寫鎖完成后訪問數(shù)據(jù)。

  
 
 
 
    
   
  
  
  
  1. public class ReadWriteLockTest { 
    2. 
   
  
  
  
  2.     static ExecutorService executor = Executors.newFixedThreadPool(8); 
    3. 
   
  
  
  
  3.     static AtomicInteger stock = new AtomicInteger(3); 
    4. 
   
  
  
  
  4.     static InterProcessMutex readLock; 
    5. 
   
  
  
  
  5.     static InterProcessMutex writeLock; 
    6. 
   
  
  
  
  6.     public static void main(String[] args) throws InterruptedException { 
    7. 
   
  
  
  
  7.         CuratorFramework client = getZkClient(); 
    8. 
   
  
  
  
  8.         String key = "/lock/lockId_111/1111"; 
    9. 
   
  
  
  
  9.         InterProcessReadWriteLock readWriteLock = new InterProcessReadWriteLock(client, key); 
    10. 
   
  
  
  
  10.         readLock = readWriteLock.readLock(); 
    11. 
   
  
  
  
  11.         writeLock = readWriteLock.writeLock(); 
    12. 
   
  
  
  
  12.         for (int i = 0; i < 16; i++) { 
    13. 
   
  
  
  
  13.             executor.submit(() -> { 
    14. 
   
  
  
  
  14.                 try { 
    15. 
   
  
  
  
  15.                     boolean read = readLock.acquire(2000, TimeUnit.MILLISECONDS); 
    16. 
   
  
  
  
  16.                     if (read) { 
    17. 
   
  
  
  
  17.                         int num = stock.get(); 
    18. 
   
  
  
  
  18.                         System.out.println("讀取庫存，當(dāng)前庫存為: " + num); 
    19. 
   
  
  
  
  19.                         if (num < 0) { 
    20. 
   
  
  
  
  20.                             System.err.println("庫存不足, 直接返回"); 
    21. 
   
  
  
  
  21.                             return; 
    22. 
   
  
  
  
  22.                         } 
    23. 
   
  
  
  
  23.                     } 
    24. 
   
  
  
  
  24.                 } catch (Exception e) { 
    25. 
   
  
  
  
  25.                     e.printStackTrace(); 
    26. 
   
  
  
  
  26.                 }finally { 
    27. 
   
  
  
  
  27.                     if (readLock.isAcquiredInThisProcess()) { 
    28. 
   
  
  
  
  28.                         try { 
    29. 
   
  
  
  
  29.                             readLock.release(); 
    30. 
   
  
  
  
  30.                         } catch (Exception e) { 
    31. 
   
  
  
  
  31.                             e.printStackTrace(); 
    32. 
   
  
  
  
  32.                         } 
    33. 
   
  
  
  
  33.                     } 
    34. 
   
  
  
  
  34.                 } 
    35. 
   
  
  
  
  35.                 try { 
    36. 
   
  
  
  
  36.                     boolean acquire = writeLock.acquire(2000, TimeUnit.MILLISECONDS); 
    37. 
   
  
  
  
  37.                     if (acquire) { 
    38. 
   
  
  
  
  38.                         int s = stock.get(); 
    39. 
   
  
  
  
  39.                         if (s <= 0) { 
    40. 
   
  
  
  
  40.                             System.err.println("進(jìn)入秒殺，庫存不足"); 
    41. 
   
  
  
  
  41.                         } else { 
    42. 
   
  
  
  
  42.                             s = stock.decrementAndGet(); 
    43. 
   
  
  
  
  43.                             System.out.println("購買成功, 剩余庫存: " + s); 
    44. 
   
  
  
  
  44.                         } 
    45. 
   
  
  
  
  45.                     } 
    46. 
   
  
  
  
  46.                 } catch (Exception e) { 
    47. 
   
  
  
  
  47.                     e.printStackTrace(); 
    48. 
   
  
  
  
  48.                 } finally { 
    49. 
   
  
  
  
  49.                     try { 
    50. 
   
  
  
  
  50.                         if (writeLock.isAcquiredInThisProcess()) 
    51. 
   
  
  
  
  51.                             writeLock.release(); 
    52. 
   
  
  
  
  52.                     } catch (Exception e) { 
    53. 
   
  
  
  
  53.                         e.printStackTrace(); 
    54. 
   
  
  
  
  54.                     } 
    55. 
   
  
  
  
  55.                 } 
    56. 
   
  
  
  
  56.             }); 
    57. 
   
  
  
  
  57.         } 
    58. 
   
  
  
  
  58.         while (true) { 
    59. 
   
  
  
  
  59.             if (executor.isTerminated()) { 
    60. 
   
  
  
  
  60.                 executor.shutdown(); 
    61. 
   
  
  
  
  61.                 System.out.println("秒殺完畢剩余庫存為:" + stock.get()); 
    62. 
   
  
  
  
  62.             } 
    63. 
   
  
  
  
  63.             TimeUnit.MILLISECONDS.sleep(100); 
    64. 
   
  
  
  
  64.         } 
    65. 
   
  
  
  
  65.     } 
    66. 
   
  
  
  
  66.     private static CuratorFramework getZkClient() { 
    67. 
   
  
  
  
  67.         String zkServerAddress = "127.0.0.1:2181"; 
    68. 
   
  
  
  
  68.         ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3, 5000); 
    69. 
   
  
  
  
  69.         CuratorFramework zkClient = CuratorFrameworkFactory.builder() 
    70. 
   
  
  
  
  70.                 .connectString(zkServerAddress) 
    71. 
   
  
  
  
  71.                 .sessionTimeoutMs(5000) 
    72. 
   
  
  
  
  72.                 .connectionTimeoutMs(5000) 
    73. 
   
  
  
  
  73.                 .retryPolicy(retryPolicy) 
    74. 
   
  
  
  
  74.                 .build(); 
    75. 
   
  
  
  
  75.         zkClient.start(); 
    76. 
   
  
  
  
  76.         return zkClient; 
    77. 
   
  
  
  
  77.     } 
    78. 
   
  
  
  
  78. } 
    79.

打印結(jié)果如下，一開始會(huì)有 8 個(gè)輸出結(jié)果為 讀取庫存，當(dāng)前庫存為: 3 然后在寫鎖中回去順序的扣減少庫存。

  
 
 
 
    
   
  
  
  
  1. 讀取庫存，當(dāng)前庫存為: 3 
    2. 
   
  
  
  
  2. 讀取庫存，當(dāng)前庫存為: 3 
    3. 
   
  
  
  
  3. 讀取庫存，當(dāng)前庫存為: 3 
    4. 
   
  
  
  
  4. 讀取庫存，當(dāng)前庫存為: 3 
    5. 
   
  
  
  
  5. 讀取庫存，當(dāng)前庫存為: 3 
    6. 
   
  
  
  
  6. 讀取庫存，當(dāng)前庫存為: 3 
    7. 
   
  
  
  
  7. 讀取庫存，當(dāng)前庫存為: 3 
    8. 
   
  
  
  
  8. 讀取庫存，當(dāng)前庫存為: 3 
    9. 
   
  
  
  
  9. 購買成功, 剩余庫存: 2 
    10. 
   
  
  
  
  10. 購買成功, 剩余庫存: 1 
    11. 
   
  
  
  
  11. 購買成功, 剩余庫存: 0 
    12. 
   
  
  
  
  12. 進(jìn)入秒殺，庫存不足 
    13. 
   
  
  
  
  13. 進(jìn)入秒殺，庫存不足 
    14. 
   
  
  
  
  14. 進(jìn)入秒殺，庫存不足 
    15. 
   
  
  
  
  15. 進(jìn)入秒殺，庫存不足 
    16. 
   
  
  
  
  16. 進(jìn)入秒殺，庫存不足 
    17. 
   
  
  
  
  17. 讀取庫存，當(dāng)前庫存為: 0 
    18. 
   
  
  
  
  18. 讀取庫存，當(dāng)前庫存為: 0 
    19. 
   
  
  
  
  19. 讀取庫存，當(dāng)前庫存為: 0 
    20. 
   
  
  
  
  20. 讀取庫存，當(dāng)前庫存為: 0 
    21. 
   
  
  
  
  21. 讀取庫存，當(dāng)前庫存為: 0 
    22. 
   
  
  
  
  22. 讀取庫存，當(dāng)前庫存為: 0 
    23. 
   
  
  
  
  23. 讀取庫存，當(dāng)前庫存為: 0 
    24. 
   
  
  
  
  24. 讀取庫存，當(dāng)前庫存為: 0 
    25. 
   
  
  
  
  25. 進(jìn)入秒殺，庫存不足 
    26. 
   
  
  
  
  26. 進(jìn)入秒殺，庫存不足 
    27. 
   
  
  
  
  27. 進(jìn)入秒殺，庫存不足 
    28. 
   
  
  
  
  28. 進(jìn)入秒殺，庫存不足 
    29. 
   
  
  
  
  29. 進(jìn)入秒殺，庫存不足 
    30. 
   
  
  
  
  30. 進(jìn)入秒殺，庫存不足 
    31. 
   
  
  
  
  31. 進(jìn)入秒殺，庫存不足 
    32. 
   
  
  
  
  32. 進(jìn)入秒殺，庫存不足 
    33.

分布式鎖的選擇

咱們最常用的就是 Redis 的分布式鎖和 Zookeeper 的分布式鎖，在性能方面 Redis 的每秒鐘 TPS 可以上輕松上萬。在大規(guī)模的高并發(fā)場(chǎng)景我推薦使用 Redis 分布式鎖來作為推薦的技術(shù)方案。如果對(duì)并發(fā)要求不是特別高的場(chǎng)景可以使用 Zookeeper 分布式來處理。

參考資料

https://www.cnblogs.com/leeego-123/p/12162220.html

http://curator.apache.org/

https://blog.csdn.net/hosaos/article/details/89521537


本文標(biāo)題：分布式鎖看了又看，優(yōu)秀方案我來告訴你
鏈接分享：http://m.5511xx.com/article/cdjpepo.html