Zk實現分布式鎖

Zookeeper實現分布式鎖

zookeeper實現分布式鎖，主要得益於ZooKeeper保證了數據的強一致性這一特性。鎖服務可以分為兩類，一個是保持獨占，另一個是控制時序。

1. 保持獨占，就是所有試圖來獲取這個鎖的客戶端，最終只有一個可以成功獲得這把鎖。通常的做法是把zk上的一個znode看作是一把鎖，通過 create znode 的方式來實現。所有客戶端都去創建 /distribute_lock 節點，最終成功創建的那個客戶端也即擁有了這把鎖。

2. 控制時序，就是所有試圖來獲取這個鎖的客戶端，最終都是會被安排執行，只是有個全局時序了。做法和上面基本類似，只是這里 /distribute_lock預先已經存在，客戶端在它下面創建臨時有序節點。Zk 的父節點（/distribute_lock）維持一份 sequence, 保證子節點創建的時序性，從而也形成了每個客戶端的全局時序。

 

分布式鎖的產生的原因：

1.單純的Lock鎖或者synchronize只能解決單個jvm線程安全問題

2.分布式 Session 一致性問題

3.分布式全局id（也可以使用分布式鎖） 

換個角度來說，分布式鎖產生的原因就是集群。

 

在單台服務器上，如何生唯一的訂單號，方案有UUid+時間戳方式，redis方式。

生成訂單號， 秒殺搶購時候，首先如果預測是100w訂單號，生成放在redis。客戶端下單，直接redis去獲取即可。因為redis是單線程的，如果實際是150w用戶，當redis剩下50w訂單號時候，繼續生成補充。 

但是在集群環境下，這種方式其實並不能保證其唯一性。

import java.text.SimpleDateFormat;
import java.util.Date;

//生成訂單號 時間戳
public class OrderNumGenerator {
  //區分不同的訂單號
    private static int count = 0;
//單台服務器，多個線程同時生成訂單號
    public String getNumber(){
        try {
            Thread.sleep(300);
        } catch (Exception e) {
          
        }
        SimpleDateFormat simpt = new SimpleDateFormat("yyyy-MM-dd-HH-mm-ss");  
        return simpt.format(new Date()) + "-" + ++count;  //時間戳后面加了 count
    }
}

開啟100個線程調用：

public class OrderService implements  Runnable {

     private OrderNumGenerator orderNumGenerator  = new OrderNumGenerator(); 

     public void run() {
        getNumber();     
    }

    public void getNumber(){
    String number =    orderNumGenerator.getNumber();
    System.out.println(Thread.currentThread().getName()+"num"+number);
    }

    public static void main(String[] args) {
        OrderService orderService = new OrderService();
        //開啟100個線程
        for (int i = 0; i <100; i++) {  
                    new Thread(orderService).start();
        }    
    }
}

結果：

 因為多個線程共享同一個全局變量，會產生線程安全問題！

 解決方案當然就是可以加鎖：

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class OrderService implements Runnable {

    private OrderNumGenerator orderNumGenerator = new OrderNumGenerator();
    private Lock lock = new ReentrantLock();

    public void run() {
        getNumber();
    }

    public void getNumber() {
        //加鎖
        lock.lock();
        String number = orderNumGenerator.getNumber();
        System.out.println(Thread.currentThread().getName() + "生成訂單：" + number);
        //釋放鎖
        lock.unlock();
    }

    public static void main(String[] args) {
        OrderService orderService = new OrderService();
        // 開啟100個線程
        for (int i = 0; i < 100; i++) {
            new Thread(orderService).start();
        }
    }
}

但是這種方式效率很低！

 

如果是集群環境下： 

     每台jvm都有一個count，都有自增的代碼操作這個count， 三個不同的jvm獨立的用戶請 過來 映射到哪個就操作哪個，這時候就產生分布式鎖的問題。

這時候需要分布式鎖，共享一個count

jvm1 操作時候 其他的jvm2 和 jvm3 不可以操作他。

 

分布式鎖：保證分布式領域中共享數據安全問題，它的實現方式可以有這些：

1、數據庫實現（效率很低）

2、redis實現（使用redission實現，但是需要考慮釋放問題。也比較麻煩）

3、Zookeeper實現（使用臨時節點，效率高，失效時間可以控制）

4、Spring Cloud實現全局鎖（內置的）

 

下面用一個業務場景加上代碼來說明：

業務場景

在分布式情況，生成全局訂單號ID

 

產生問題

在分布式集群環境下，每台機器不能實現同步，在分布式場景下使用時間戳生成訂單號可能會重復

 

Zookeeper實現分布式鎖原理

     使用zookeeper創建臨時序列節點來實現分布式鎖，大體思路就是創建臨時序列節點，找出最小的序列節點，獲取分布式鎖，程序執行完成之后此序列節點消失，通過watch來監控節點的變化，從剩下的節點的找到最小的序列節點，獲取分布式鎖，執行相應處理，依次類推……

      因為zk節點唯一的，不能重復，節點類型為臨時節點， 一台zk服務器創建成功時候，另外的zk服務器創建節點時候就會報錯，該節點已經存在。這時候其他的zk服務器就會開始監聽並等待。讓這台zk服務器的程序現在執行完畢，釋放鎖。關閉當前會話。臨時節點就會消失，並且事件通知Watcher，其他的就會來創建。

 

代碼實現

　創建鎖的接口

public interface ExtLock {  

    //ExtLock基於zk實現分布式鎖
    public void  getLock();

    //釋放鎖
    public void unLock()；

}

實現zk分布式鎖：

import java.util.concurrent.CountDownLatch;
import org.I0Itec.zkclient.IZkDataListener;

public class ZookeeperDistrbuteLock implements ExtLock{

    private static final String CONNECTION="192.168.2.222:2181";
    private ZkClient zkClient = new ZkClient(CONNECTION);
    private String lockPath="/distribute_lock";
    private CountDownLatch countDownLatch;

     //獲取鎖
      public void getLock() { 
          // 如果節點創建成果，直接執行業務邏輯，如果節點創建失敗，進行等待 
          if (tryLock()) {
            System.out.println("#####成功獲取鎖######");
        }else {
            //進行等待
            waitLock();
        }   
    }

    //釋放鎖
      public void unLock() {
        //執行完畢 直接連接
          if (zkClient != null) {
            zkClient.close();
            System.out.println("######釋放鎖完畢######");
        }
    }

     public boolean tryLock() {
        try {
            zkClient.createEphemeral(lockPath);
            return true;
        } catch (Exception e) {
            // 如果失敗 直接catch
            return false;
        }
    }

    public void waitLock() {
        IZkDataListener iZkDataListener = new IZkDataListener() {
            // 節點被刪除
            public void handleDataDeleted(String arg0) throws Exception {
                if (countDownLatch != null) {
                    countDownLatch.countDown(); // 計數器為0的情況，await 后面的繼續執行
                }
            }

     // 節點被修改
      public void handleDataChange(String arg0, Object arg1) throws Exception {
        System.out.println("########節點被修改#######");
            }
        };

        // 監聽事件通知
        zkClient.subscribeDataChanges(lockPath, iZkDataListener);
        // 控制程序的等待
        if (zkClient.exists(lockPath)) {  //如果檢查出已經被創建了就等待
            countDownLatch = new CountDownLatch(1);
            try {
                countDownLatch.wait(); //當為0時候，后面的繼續執行
            } catch (Exception e) {
            }
        }
        //后面代碼繼續執行
        //刪除該事件監聽
        zkClient.unsubscribeDataChanges(lockPath, iZkDataListener);
    }
}

生產訂單號：

import java.text.SimpleDateFormat;
import java.util.Date;

//生成訂單號 時間戳
public class OrderNumGenerator {

  //區分不同的訂單號
    private static int count = 0;

//單台服務器，多個線程 同事生成訂單號
    public String getNumber(){
        SimpleDateFormat simpt = new SimpleDateFormat("yyyy-MM-dd-HH-mm-ss");  
        return simpt.format(new Date()) + "-" + ++count;  //時間戳后面加了 count
    }
}

運行方法：

public class OrderService implements Runnable {

    private OrderNumGenerator orderNumGenerator = new OrderNumGenerator(); 
    private ExtLock lock = new ZookeeperDistrbuteLock();

    public void run() {
        getNumber();
    }

    public void getNumber() { 
            lock.getLock();
            String number = orderNumGenerator.getNumber();
            System.out.println(Thread.currentThread().getName() + ",number" + number);
           try {
            Thread.sleep(10000);//為了看效果
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
             lock.unLock();
    }

    public static void main(String[] args) {
        for (int i = 0; i < 100; i++) { // 開啟100個線程
            //模擬分布式鎖的場景
            new Thread(new OrderService()).start();
        }
    }
}

運行結果：