什么是分布式鎖
當多個行程在同一個系統中，用分布式鎖控制多個行程對資源的訪問
分布式鎖應用場景
- 傳統的單體應用單機部署情況下，可以使用java并發處理相關的API進行互斥控制，
- 分布式系統后由于多執行緒，多行程分布在不同機器上，使單機部署情況下的并發控制鎖策略失效，為了解決跨JVM互斥機制來控制共享資源的訪問，這就是分布式鎖的來源；分布式鎖應用場景大都是高并發、大流量場景，
分布式鎖實作
1、基于redis的分布式鎖
redis分布式鎖的實作
- 加鎖機制：根據hash節點選擇一個客戶端執行lua腳本
- 鎖互斥機制：再來一個客戶端執行同樣的lua腳本會提示已經存在鎖，然后進入回圈一直嘗試加鎖
- 可重入機制
- watch dog自動延期機制
- 釋放鎖機制
測驗用例
單機
1 private RedissonClient getClient(){ 2 Config config = new Config(); 3 config.useSingleServer().setAddress("redis://127.0.0.1:6379");//.setPassword("");//.setConnectionMinimumIdleSize(10).setConnectionPoolSize(10);//.setConnectionPoolSize();//172.16.10.164 4 RedissonClient redissonClient = Redisson.create(config); 5 return redissonClient; 6 } 7 private ExecutorService executorService = Executors.newCachedThreadPool(); 8 ---------------------------------------------------------------- 9 int[] count = {0}; 10 for (int i = 0; i < 10; i++) { 11 RedissonClient client = getClient(); 12 final RedisLock redisLock = new RedisLock(client,"lock_key"); 13 executorService.submit(() -> { 14 try { 15 redisLock.lock(); 16 count[0]++; 17 } catch (Exception e) { 18 e.printStackTrace(); 19 } finally { 20 try { 21 redisLock.unlock(); 22 } catch (Exception e) { 23 e.printStackTrace(); 24 } 25 } 26 });
RedLock
1 public static RLock create (String url, String key){ 2 Config config = new Config(); 3 config.useSingleServer().setAddress(url); 4 RedissonClient redissonClient = Redisson.create(config); 5 return redissonClient.getLock(key); 6 } 7 8 RedissonRedLock redissonRedLock = new RedissonRedLock( 9 create("redis://redis://127.0.0.1:6379","lock_key1"), 10 create("redis://redis://127.0.0.1:6380","lock_key2"), 11 create("redis://redis://127.0.0.1:6381","lock_key3")); 12 RedisRedLock redLock = new RedisRedLock(redissonRedLock); 13 14 private ExecutorService executorService = Executors.newCachedThreadPool(); 15 16 ------------------------------------------------------------------ 17 int[] count = {0}; 18 for (int i = 0; i < 2; i++) { 19 executorService.submit(() -> { 20 try { 21 redLock.lock(); 22 count[0]++; 23 } catch (Exception e) { 24 e.printStackTrace(); 25 } finally { 26 try { 27 redLock.unlock(); 28 } catch (Exception e) { 29 e.printStackTrace(); 30 } 31 } 32 }); 33 }
redis原始碼分析
redisson
1 public void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException { 2 //獲取當前執行緒id 3 long threadId = Thread.currentThread().getId(); 4 //嘗試獲得鎖，回傳還剩余的鎖過期時間 5 Long ttl = tryAcquire(leaseTime, unit, threadId); 6 // lock acquired 7 //如果ttl為空，代表當前沒有鎖，獲取成功 8 if (ttl == null) { 9 return; 10 } 11 12 //如果獲取鎖失敗，則訂閱到對應這個鎖的channel，一旦其他執行緒釋放鎖時，通知執行緒去獲取鎖 13 RFuture<RedissonLockEntry> future = subscribe(threadId); 14 commandExecutor.syncSubscription(future); 15 16 try { 17 //回圈等待 18 while (true) { 19 //嘗試獲得鎖 20 ttl = tryAcquire(leaseTime, unit, threadId); 21 // lock acquired 22 if (ttl == null) { 23 break; 24 } 25 26 // waiting for message //ttl大于0，則等待ttl時間后繼續嘗試獲取鎖 27 if (ttl >= 0) { 28 getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS); 29 } else { 30 getEntry(threadId).getLatch().acquire(); 31 } 32 } 33 } finally { 34 unsubscribe(future, threadId); 35 } 36 // get(lockAsync(leaseTime, unit)); 37 }
2、基于ETCD實作分布式鎖分析
ETCD分布式鎖的實作
-
Lease機制：租約機制（TTL，Time To Live），Etcd 可以為存盤的 key-value 對設定租約，
當租約到期，key-value 將失效洗掉；同時也支持續約，通過客戶端可以在租約到期之前續約，
以避免 key-value 對過期失效，Lease 機制可以保證分布式鎖的安全性，為鎖對應的 key 配置租約，
即使鎖的持有者因故障而不能主動釋放鎖，鎖也會因租約到期而自動釋放 -
Revision機制：每個 key 帶有一個 Revision 號，每進行一次事務加一，它是全域唯一的，
通過 Revision 的大小就可以知道進行寫操作的順序，在實作分布式鎖時，多個客戶端同時搶鎖，
根據 Revision 號大小依次獲得鎖，可以避免 “羊群效應” ，實作公平鎖 -
Prefix機制：即前綴機制，例如，一個名為 /etcdlock 的鎖，兩個爭搶它的客戶端進行寫操作，
實際寫入的 key 分別為：key1="/etcdlock/UUID1"，key2="/etcdlock/UUID2"，
其中，UUID 表示全域唯一的 ID，確保兩個 key 的唯一性，寫操作都會成功，但回傳的 Revision 不一樣，
那么，如何判斷誰獲得了鎖呢？通過前綴 /etcdlock 查詢，回傳包含兩個 key-value 對的的 KeyValue 串列，
同時也包含它們的 Revision，通過 Revision 大小，客戶端可以判斷自己是否獲得鎖 -
Watch機制：即監聽機制，Watch 機制支持 Watch 某個固定的 key，也支持 Watch 一個范圍（前綴機制），
當被 Watch 的 key 或范圍發生變化，客戶端將收到通知；在實作分布式鎖時，如果搶鎖失敗，
可通過 Prefix 機制回傳的 KeyValue 串列獲得 Revision 比自己小且相差最小的 key（稱為 pre-key），
對 pre-key 進行監聽，因為只有它釋放鎖，自己才能獲得鎖，如果 Watch 到 pre-key 的 DELETE 事件，
則說明 pre-key 已經釋放，自己已經持有鎖
基于ETCD分布式鎖
步驟1、建立連接
客戶端連接 Etcd，以 /etcd/lock 為前綴創建全域唯一的 key，
假設第一個客戶端對應的 key="/etcd/lock/UUID1"，第二個為 key="/etcd/lock/UUID2"；
客戶端分別為自己的 key 創建租約 - Lease，租約的長度根據業務耗時確定；
步驟2、創建定時任務作為租約的“心跳”
當一個客戶端持有鎖期間，其它客戶端只能等待，為了避免等待期間租約失效，
客戶端需創建一個定時任務作為“心跳”進行續約，此外，如果持有鎖期間客戶端崩潰，
心跳停止，key 將因租約到期而被洗掉，從而鎖釋放，避免死鎖
步驟3、客戶端將自己全域唯一的 key 寫入 Etcd
執行 put 操作，將步驟 1 中創建的 key 系結租約寫入 Etcd，根據 Etcd 的 Revision 機制，
假設兩個客戶端 put 操作回傳的 Revision 分別為 1、2，客戶端需記錄 Revision 用以
接下來判斷自己是否獲得鎖
步驟 4、客戶端判斷是否獲得鎖
客戶端以前綴 /etcd/lock/ 讀取 keyValue 串列，判斷自己 key 的 Revision 是否為當前串列中
最小的，如果是則認為獲得鎖；否則監聽串列中前一個 Revision 比自己小的 key 的洗掉事件，一旦監聽到洗掉事件或者因租約失效而洗掉的事件，則自己獲得鎖，
步驟 5、執行業務
獲得鎖后，操作共享資源，執行業務代碼
步驟 6、釋放鎖
完成業務流程后，洗掉對應的key釋放鎖
測驗用例
1 public class EtcdDistributeLock extends AbstractLock{ 2 3 private Client client; 4 private Lock lockClient; 5 private Lease leaseClient; 6 private String lockKey; 7 private String lockPath; 8 /** 鎖的次數 */ 9 private AtomicInteger lockCount; 10 /** 租約有效期,防止客戶端崩潰，可在租約到期后自動釋放鎖；另一方面，正常執行程序中，會自動進行續租,單位 ns */ 11 private Long leaseTTL; 12 /** 續約鎖租期的定時任務，初次啟動延遲，單位默認為 s,默認為1s，可根據業務定制設定*/ 13 private Long initialDelay = 0L; 14 /** 定時任務執行緒池類 */ 15 ScheduledExecutorService service = null; 16 /** 保存執行緒與鎖物件的映射，鎖物件包含重入次數，重入次數的最大限制為Int的最大值 */ 17 private final ConcurrentMap<Thread, LockData> threadData =https://www.cnblogs.com/selfsalvation/archive/2022/04/24/ Maps.newConcurrentMap(); 18 19 public EtcdDistributeLock(){} 20 21 public EtcdDistributeLock(Client client, String lockKey, long leaseTTL,TimeUnit unit){ 22 this.client = client; 23 lockClient = client.getLockClient(); 24 leaseClient = client.getLeaseClient(); 25 this.lockKey = lockKey; 26 // 轉納秒 27 this.leaseTTL = unit.toNanos(leaseTTL); 28 service = Executors.newSingleThreadScheduledExecutor(); 29 } 30 31 32 @Override 33 public void lock() { 34 // 檢查重入性 35 Thread currentThread = Thread.currentThread(); 36 LockData oldLockData =https://www.cnblogs.com/selfsalvation/archive/2022/04/24/ threadData.get(currentThread); 37 if (oldLockData != null && oldLockData.isLockSuccess()) { 38 // re-entering 39 int lockCount = oldLockData.lockCount.incrementAndGet(); 40 if(lockCount < 0 ){ 41 throw new Error("超出可重入次數限制"); 42 } 43 return; 44 } 45 46 // 記錄租約 ID 47 Long leaseId = 0L; 48 try{ 49 leaseId = leaseClient.grant(TimeUnit.NANOSECONDS.toSeconds(leaseTTL)).get().getID(); 50 // 續租心跳周期 51 long period = leaseTTL - leaseTTL / 5; 52 // 啟動定時任務續約 53 service.scheduleAtFixedRate(new EtcdDistributeLock.KeepAliveRunnable(leaseClient, leaseId), 54 initialDelay,period,TimeUnit.NANOSECONDS); 55 LockResponse lockResponse = lockClient.lock(ByteSequence.from(lockKey.getBytes()), leaseId).get(); 56 if(lockResponse != null){ 57 lockPath = lockResponse.getKey().toString(Charset.forName("utf-8")); 58 log.info("獲取鎖成功,鎖路徑:{},執行緒:{}",lockPath,currentThread.getName()); 59 } 60 }catch (InterruptedException | ExecutionException e){ 61 log.error("獲取鎖失敗",e); 62 return; 63 } 64 // 獲取鎖成功，鎖物件設定 65 LockData newLockData = https://www.cnblogs.com/selfsalvation/archive/2022/04/24/new LockData(currentThread, lockKey); 66 newLockData.setLeaseId(leaseId); 67 newLockData.setService(service); 68 threadData.put(currentThread, newLockData); 69 newLockData.setLockSuccess(true); 70 } 71 72 @Override 73 public void lockInterruptibly() throws InterruptedException { 74 super.lockInterruptibly(); 75 } 76 77 @Override 78 public boolean tryLock() { 79 return super.tryLock(); 80 } 81 82 @Override 83 public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { 84 return super.tryLock(time,unit); 85 } 86 87 88 @Override 89 public void unlock() { 90 Thread currentThread = Thread.currentThread(); 91 LockData lockData =https://www.cnblogs.com/selfsalvation/archive/2022/04/24/ threadData.get(currentThread); 92 if (lockData =https://www.cnblogs.com/selfsalvation/archive/2022/04/24/= null){ 93 throw new IllegalMonitorStateException("You do not own the lock: " + lockKey); 94 } 95 int newLockCount = lockData.lockCount.decrementAndGet(); 96 if ( newLockCount > 0 ) { 97 return; 98 } 99 if ( newLockCount < 0 ) { 100 throw new IllegalMonitorStateException("Lock count has gone negative for lock: " + lockKey); 101 } 102 try { 103 // 釋放鎖 104 if(lockPath != null){ 105 lockClient.unlock(ByteSequence.from(lockPath.getBytes())).get(); 106 } 107 if(lockData != null){ 108 // 關閉定時任務 109 lockData.getService().shutdown(); 110 // 洗掉租約 111 if (lockData.getLeaseId() != 0L) { 112 leaseClient.revoke(lockData.getLeaseId()); 113 } 114 } 115 } catch (InterruptedException | ExecutionException e) { 116 log.error("解鎖失敗",e); 117 }finally { 118 // 移除當前執行緒資源 119 threadData.remove(currentThread); 120 } 121 } 122 123 124 @Override 125 public Condition newCondition() { 126 return super.newCondition(); 127 } 128 129 /** 130 * 心跳續約執行緒類 131 */ 132 public static class KeepAliveRunnable implements Runnable { 133 private Lease leaseClient; 134 private long leaseId; 135 136 public KeepAliveRunnable(Lease leaseClient, long leaseId) { 137 this.leaseClient = leaseClient; 138 this.leaseId = leaseId; 139 } 140 141 @Override 142 public void run() { 143 // 對該leaseid進行一次續約 144 leaseClient.keepAliveOnce(leaseId); 145 } 146 } 147 148 public class EtcdLockTest { 149 private Client client; 150 private String key = "/etcd/lock"; 151 private static final String server = "http://xxxx:xxxx"; 152 private ExecutorService executorService = Executors.newFixedThreadPool(10000); 153 154 @Before 155 public void before() throws Exception { 156 initEtcdClient(); 157 } 158 159 private void initEtcdClient(){ 160 client = Client.builder().endpoints(server).build(); 161 } 162 163 @Test 164 public void testEtcdDistributeLock() throws InterruptedException { 165 int[] count = {0}; 166 for (int i = 0; i < 100; i++) { 167 executorService.submit(() -> { 168 final EtcdDistributeLock lock = new EtcdDistributeLock(client, key,20,TimeUnit.SECONDS); 169 try { 170 lock.lock(); 171 count[0]++; 172 } catch (Exception e) { 173 e.printStackTrace(); 174 } finally { 175 try { 176 lock.unlock(); 177 } catch (Exception e) { 178 e.printStackTrace(); 179 } 180 } 181 }); 182 } 183 executorService.shutdown(); 184 executorService.awaitTermination(1, TimeUnit.HOURS); 185 System.err.println("執行結果: " + count[0]); 186 } 187 }
3、基于Zookeeper分布式鎖
實作原理
- 啟動客戶端，確認鏈接到了服務器
- 多個客戶端并發的在特定路徑下創建臨時性順序節點
- 客戶端判斷自己的創建的順序節點是否是最小的，如果是最小的，則獲取鎖成功
- 第三步若判定失敗，則采用zk的watch機制監聽自己的前一個順序節點，等待前一個節點的洗掉（放鎖）事件，再開始第三步判定，
zookeeper作為高性能分布式協調框架，可以把其看做一個檔案系統，其中有節點的概念，并且分為4種：1.持久性節點2.持久性順序節點3.臨時性節點4.臨時性順序節點，
分布式鎖的實作主要思路就是：監控其他客戶端的狀態，來判斷自己是否可以獲得鎖，
采用臨時性順序節點的原因：
- zk服務器維護了客戶端的會話有效性，當會話失效的時候，其會話所創建的臨時性節點都會被洗掉，通過這一特點，可以通過watch臨時節點來監控其他客戶端的情況，方便自己做出相應動作，
- 因為zk對寫操作是順序性的，所以并發創建的順序節點會有一個唯一確定的序號，當前鎖是公平鎖的一種實作，所以依靠這種順序性可以很好的解釋—節點序列小的獲取到鎖并且可以采用watch自己的前一個節點來避免驚群現象（這樣watch事件的傳播是線性的），
測驗用例
1 public class ZKLock extends AbstractLock { 2 3 /** 4 * 1.Connect to zk 5 */ 6 private CuratorFramework client; 7 8 private InterProcessLock lock ; 9 10 11 public ZKLock(String zkAddress,String lockPath) { 12 // 1.Connect to zk 13 client = CuratorFrameworkFactory.newClient( 14 zkAddress, 15 new RetryNTimes(5, 5000) 16 ); 17 client.start(); 18 if(client.getState() == CuratorFrameworkState.STARTED){ 19 log.info("zk client start successfully!"); 20 log.info("zkAddress:{},lockPath:{}",zkAddress,lockPath); 21 }else{ 22 throw new RuntimeException("客戶端啟動失敗，，，"); 23 } 24 this.lock = defaultLock(lockPath); 25 } 26 27 private InterProcessLock defaultLock(String lockPath ){ 28 return new InterProcessMutex(client, lockPath); 29 } 30 @Override 31 public void lock() { 32 try { 33 this.lock.acquire(); 34 } catch (Exception e) { 35 throw new RuntimeException(e); 36 } 37 } 38 39 @Override 40 public boolean tryLock() { 41 boolean flag ; 42 try { 43 flag=this.lock.acquire(0,TimeUnit.SECONDS); 44 } catch (Exception e) { 45 throw new RuntimeException(e); 46 } 47 return flag; 48 } 49 50 @Override 51 public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { 52 boolean flag ; 53 try { 54 flag=this.lock.acquire(time,unit); 55 } catch (Exception e) { 56 throw new RuntimeException(e); 57 } 58 return flag; 59 } 60 61 @Override 62 public void unlock() { 63 try { 64 this.lock.release(); 65 } catch (Exception e) { 66 throw new RuntimeException(e); 67 } 68 } 69 70 } 71 private ExecutorService executorService = Executors.newCachedThreadPool(); 72 73 74 @Test 75 public void testLock() throws Exception{ 76 ZKLock zkLock = new ZKLock("xxxx:xxxx","/lockPath"); 77 int[] num = {0}; 78 long start = System.currentTimeMillis(); 79 for(int i=0;i<200;i++){ 80 executorService.submit(()->{ 81 try { 82 zkLock.lock(); 83 num[0]++; 84 } catch (Exception e){ 85 throw new RuntimeException(e); 86 } finally { 87 zkLock.unlock(); 88 } 89 }); 90 91 } 92 executorService.shutdown(); 93 executorService.awaitTermination(1, TimeUnit.HOURS); 94 log.info("耗時:{}",System.currentTimeMillis()-start); 95 System.out.println(num[0]); 96 }
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/463474.html
標籤：其他
上一篇：python資料處理-matplotlib入門(1)--安裝及繪制簡單的曲線
下一篇：如何用html寫書