前言： jdk提供的synchronized和ReentrantLock可以幫助我們在單進程中解決資源共享數據一致性，但是在分布式系統中是多進程多線程，這個時候僅僅使用jdk實現的鎖解決不了資源共享的問題，比如某商城中數據庫有10個商品，A用戶想要買走6個，B用戶想買走5個。如果系統運行在單臺機器上，我們使用Jdk提供的鎖，可以保證數據的一致性，但是當系統運行在多臺機器中，JDK實現的鎖就會失效，這個時候就應該使用分布式鎖，每次只能保證一臺機器在請求資源。分布式鎖有三種不同的方式實現，分別是數據庫提供的分布式鎖、redis、zookeeper實現，今天主要講zookeeper實現分布式鎖 在學習zk實現分布所之前，我們應該需要了解一些zk的知識 1、持久節點：客戶端斷開連接zk不刪除persistent類型節點 2、臨時節點：客戶端斷開連接zk刪除ephemeral類型節點 3、順序節點：節點后面會自動生成類似0000001的數字表示順序 4、節點變化的通知：客戶端注冊了監聽節點變化的時候，會調用回調方法 源碼地址：githut源碼地址 一、zk實現的簡單的分布式鎖 1、zk實現簡單的分布式鎖的思路，主要是抓住一下三點 （1）、當一個客戶端成功創建一個節點，另外一個客戶端是無法創建同名的節點（達到互斥的效果） （2）、我們注冊該節點的監聽時間，當節點刪除，會通知其他的客戶端，這個時候其他的客戶端可以重新去創建該節點（可以認為時拿到鎖的客戶端釋放鎖，其他的客戶端可以搶鎖） （3）、創建的節點應該時臨時節點，這樣保證我們在已經拿到鎖的客戶端掛掉了會自動釋放鎖 2、圖解  3、代碼實現 AbstractLock.java ``` package zklock; import org.I0Itec.zkclient.ZkClient; public abstract class AbstractLock { //zk地址和端口 public static final String ZK_ADDR = "192.168.0.230:2181"; //超時時間 public static final int SESSION_TIMEOUT = 10000; //創建zk protected ZkClient zkClient = new ZkClient(ZK_ADDR, SESSION_TIMEOUT); /** * 可以認為是模板模式，兩個子類分別實現它的抽象方法 * 1，簡單的分布式鎖 * 2，高性能分布式鎖 */ public void getLock() { String threadName = Thread.currentThread().getName(); if (tryLock()) { System.out.println(threadName+"-獲取鎖成功"); }else { System.out.println(threadName+"-獲取鎖失敗，進行等待..."); waitLock(); //遞歸重新獲取鎖 getLock(); } } public abstract void releaseLock(); public abstract boolean tryLock(); public abstract void waitLock(); } ``` ?AbstractLock類是個抽象類，里面getLock使用模板模式，子類分別是簡單的zk鎖和高性能的zk鎖 SimpleZkLock.java ``` package zklock; import java.util.concurrent.CountDownLatch; import org.I0Itec.zkclient.IZkDataListener; /** * @author hongtaolong * 簡單的分布式鎖的實現 */ public class SimpleZkLock extends AbstractLock { private static final String NODE_NAME = "/test_simple_lock"; private CountDownLatch countDownLatch; @Override public void releaseLock() { if (null != zkClient) { //刪除節點 zkClient.delete(NODE_NAME); zkClient.close(); System.out.println(Thread.currentThread().getName()+"-釋放鎖成功"); } } //直接創建臨時節點，如果創建成功，則表示獲取了鎖,創建不成功則處理異常 @Override public boolean tryLock() { if (null == zkClient) return false; try { zkClient.createEphemeral(NODE_NAME); return true; } catch (Exception e) { return false; } } @Override public void waitLock() { //監聽器 IZkDataListener iZkDataListener = new IZkDataListener() { //節點被刪除回調 @Override public void handleDataDeleted(String dataPath) throws Exception { if (countDownLatch != null) { countDownLatch.countDown(); } } //節點改變被回調 @Override public void handleDataChange(String dataPath, Object data) throws Exception { // TODO Auto-generated method stub } }; zkClient.subscribeDataChanges(NODE_NAME, iZkDataListener); //如果存在則阻塞 if (zkClient.exists(NODE_NAME)) { countDownLatch = new CountDownLatch(1); try { countDownLatch.await(); System.out.println(Thread.currentThread().getName()+" 等待獲取鎖..."); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } } //刪除監聽 zkClient.unsubscribeDataChanges(NODE_NAME, iZkDataListener); } } ``` SimpleZkLock 表示簡單的zk分布式鎖，邏輯還是相對比較簡單，下面看下測試 LockTest.java ``` package zklock; public class LockTest { public static void main(String[] args) { //模擬多個10個客戶端 for (int i=0;i<10;i++) { Thread thread = new Thread(new LockRunnable()); thread.start(); } } static class LockRunnable implements Runnable{ @Override public void run() { AbstractLock zkLock = new SimpleZkLock(); //AbstractLock zkLock = new HighPerformanceZkLock(); zkLock.getLock(); //模擬業務操作 try { Thread.sleep(500); } catch (InterruptedException e) { e.printStackTrace(); } zkLock.releaseLock(); } } } ``` 測試結果： ? ?打印比較多，還沒有全部截完...這個時候我們也能看出使用zk實現的簡單的分布式鎖存在的性能問題 二、高性能分布式鎖 上面使用zk實現的簡單的分布式鎖，實現比較簡單，但是存在性能問題，從上面的打印的結果可以看出、每一次客戶端釋放鎖的時候，其他的客戶端都會去搶鎖，這就造成了不必要的浪費。那么如果提升性能呢？ 1、思路：客戶端在搶鎖的時候進行排隊，客戶端只要監聽它前一個節點的變化就行，如果前一個節點釋放了鎖，客戶端才去進行搶鎖操作，這個時候我們就需要創建順序節點了 2、圖解 （1）客戶端排隊  ?（2）獲取鎖的邏輯  3、代碼實現 HighPerformanceZkLock .java ``` package zklock; import java.util.Collections; import java.util.List; import java.util.concurrent.CountDownLatch; import org.I0Itec.zkclient.IZkDataListener; /** * 高性能分布式鎖 * @author hongtaolong * */ public class HighPerformanceZkLock extends AbstractLock { private static final String PATH = "/highPerformance_zklock"; //當前節點路徑 private String currentPath; //前一個節點的路徑 private String beforePath; private CountDownLatch countDownLatch = null; public HighPerformanceZkLock() { //如果不存在這個節點，則創建持久節點 if (!zkClient.exists(PATH)) { zkClient.createPersistent(PATH); } } @Override public void releaseLock() { if (null != zkClient) { zkClient.delete(currentPath); zkClient.close(); } } @Override public boolean tryLock() { //如果currentPath為空則為第一次嘗試加鎖，第一次加鎖賦值currentPath if (null == currentPath || "".equals(currentPath)) { //在path下創建一個臨時的順序節點 currentPath = zkClient.createEphemeralSequential(PATH+"/", "lock"); } //獲取所有的臨時節點，并排序 List<String> childrens = zkClient.getChildren(PATH); Collections.sort(childrens); if (currentPath.equals(PATH+"/"+childrens.get(0))) { return true; }else {//如果當前節點不是排名第一，則獲取它前面的節點名稱，并賦值給beforePath int pathLength = PATH.length(); int wz = Collections.binarySearch(childrens, currentPath.substring(pathLength+1)); beforePath = PATH+"/"+childrens.get(wz-1); } return false; } @Override public void waitLock() { IZkDataListener lIZkDataListener = new IZkDataListener() { @Override public void handleDataDeleted(String dataPath) throws Exception { if (null != countDownLatch){ countDownLatch.countDown(); } } @Override public void handleDataChange(String dataPath, Object data) throws Exception { } }; //監聽前一個節點的變化 zkClient.subscribeDataChanges(beforePath, lIZkDataListener); if (zkClient.exists(beforePath)) { countDownLatch = new CountDownLatch(1); try { countDownLatch.await(); } catch (InterruptedException e) { e.printStackTrace(); } } zkClient.unsubscribeDataChanges(beforePath, lIZkDataListener); } } ``` 這里只要帖高性能鎖的代碼了，AbstractLock沒變化，LockTest中只要修改一行代碼 //AbstractLock zkLock = new SimpleZkLock(); AbstractLock zkLock = new HighPerformanceZkLock(); ?測試結果：  ? 上面是全部的打印結果，可以明顯看出要比上面簡單實現的分布式鎖少很多，這說明性能比上面的好，因為它不會去做無用功 ———————————————— 版權聲明：本文為CSDN博主「菜鳥的奮斗ing」的原創文章，遵循CC 4.0 BY-SA版權協議，轉載請附上原文出處鏈接及本聲明。 原文鏈接：https://blog.csdn.net/hongtaolong/article/details/88898875