ZooKeeper入門教程三分布式鎖實(shí)現(xiàn)及完整運(yùn)行源碼

更新時(shí)間：2022年01月28日 14:43:52 作者：愛碼叔（稀有氣體）

本文是ZooKeeper入門系列教程，分布式鎖有多種實(shí)現(xiàn)方式，比如通過數(shù)據(jù)庫、redis都可實(shí)現(xiàn)。作為分布式協(xié)同工具ZooKeeper，當(dāng)然也有著標(biāo)準(zhǔn)的實(shí)現(xiàn)方式。本文介紹在zookeeper中如何實(shí)現(xiàn)排他鎖

1.0版本

首先我們先介紹一個(gè)簡單的zookeeper實(shí)現(xiàn)分布式鎖的思路：

用zookeeper中一個(gè)臨時(shí)節(jié)點(diǎn)代表鎖，比如在/exlusive_lock下創(chuàng)建臨時(shí)子節(jié)點(diǎn)/exlusive_lock/lock。

所有客戶端爭相創(chuàng)建此節(jié)點(diǎn)，但只有一個(gè)客戶端創(chuàng)建成功。
創(chuàng)建成功代表獲取鎖成功，此客戶端執(zhí)行業(yè)務(wù)邏輯
未創(chuàng)建成功的客戶端，監(jiān)聽/exlusive_lock變更
獲取鎖的客戶端執(zhí)行完成后，刪除/exlusive_lock/lock，表示鎖被釋放
鎖被釋放后，其他監(jiān)聽/exlusive_lock變更的客戶端得到通知，再次爭相創(chuàng)建臨時(shí)子節(jié)點(diǎn)/exlusive_lock/lock。此時(shí)相當(dāng)于回到了第2步。

我們的程序按照上述邏輯直至搶占到鎖，執(zhí)行完業(yè)務(wù)邏輯。

上述是較為簡單的分布式鎖實(shí)現(xiàn)方式。能夠應(yīng)付一般使用場景，但存在著如下兩個(gè)問題：

1、鎖的獲取順序和最初客戶端爭搶順序不一致，這不是一個(gè)公平鎖。每次鎖獲取都是當(dāng)次最先搶到鎖的客戶端。

2、羊群效應(yīng)，所有沒有搶到鎖的客戶端都會監(jiān)聽/exlusive_lock變更。當(dāng)并發(fā)客戶端很多的情況下，所有的客戶端都會接到通知去爭搶鎖，此時(shí)就出現(xiàn)了羊群效應(yīng)。

為了解決上面的問題，我們重新設(shè)計(jì)。

2.0版本

我們在2.0版本中，讓每個(gè)客戶端在/exlusive_lock下創(chuàng)建的臨時(shí)節(jié)點(diǎn)為有序節(jié)點(diǎn)，這樣每個(gè)客戶端都在/exlusive_lock下有自己對應(yīng)的鎖節(jié)點(diǎn)，而序號排在最前面的節(jié)點(diǎn)，代表對應(yīng)的客戶端獲取鎖成功。排在后面的客戶端監(jiān)聽自己前面一個(gè)節(jié)點(diǎn)，那么在他前序客戶端執(zhí)行完成后，他將得到通知，獲得鎖成功。邏輯修改如下：

每個(gè)客戶端往/exlusive_lock下創(chuàng)建有序臨時(shí)節(jié)點(diǎn)/exlusive_lock/lock_。創(chuàng)建成功后/exlusive_lock下面會有每個(gè)客戶端對應(yīng)的節(jié)點(diǎn)，如/exlusive_lock/lock_000000001
客戶端取得/exlusive_lock下子節(jié)點(diǎn)，并進(jìn)行排序，判斷排在最前面的是否為自己。如果自己的鎖節(jié)點(diǎn)在第一位，代表獲取鎖成功，此客戶端執(zhí)行業(yè)務(wù)邏輯
如果自己的鎖節(jié)點(diǎn)不在第一位，則監(jiān)聽自己前一位的鎖節(jié)點(diǎn)。例如，自己鎖節(jié)點(diǎn)lock_000000002，那么則監(jiān)聽lock_000000001.
當(dāng)前一位鎖節(jié)點(diǎn)（lock_000000001）對應(yīng)的客戶端執(zhí)行完成，釋放了鎖，將會觸發(fā)監(jiān)聽客戶端（lock_000000002）的邏輯。
監(jiān)聽客戶端重新執(zhí)行第2步邏輯，判斷自己是否獲得了鎖。

如此修改后，每個(gè)客戶端只關(guān)心自己前序鎖是否釋放，所以每次只會有一個(gè)客戶端得到通知。而且，所有客戶端的執(zhí)行順序和最初鎖創(chuàng)建的順序是一致的。解決了1.0版本的兩個(gè)問題。

接下來我們看看代碼如何實(shí)現(xiàn)。

LockSample類

此類是分布式鎖類，實(shí)現(xiàn)了2個(gè)分布式鎖的相關(guān)方法：

1、獲取鎖

2、釋放鎖

主要程序邏輯圍繞著這兩個(gè)方法的實(shí)現(xiàn)，特別是獲取鎖的邏輯。我們先看一下該類的成員變量：

private ZooKeeper zkClient;
private static final String LOCK_ROOT_PATH = "/Locks";
private static final String LOCK_NODE_NAME = "Lock_";
private String lockPath;

定義了zkClient，用來操作zookeeper。

鎖的根路徑，及自增節(jié)點(diǎn)的前綴。此處生產(chǎn)環(huán)境應(yīng)該由客戶端傳入。

當(dāng)前鎖的路徑。

構(gòu)造方法

public LockSample() throws IOException {
    zkClient= new ZooKeeper("localhost:2181", 10000, new Watcher() {
        @Override
        public void process(WatchedEvent event) {
            if(event.getState()== Event.KeeperState.Disconnected){
                System.out.println("失去連接");
 
            }
        }
    });
}

創(chuàng)建zkClient，同時(shí)創(chuàng)建了狀態(tài)監(jiān)聽。此監(jiān)聽可以去掉，這里只是打印出失去連接狀態(tài)。

獲取鎖實(shí)現(xiàn)

暴露出來的獲取鎖的方法為acquireLock()，邏輯很簡單：

public  void acquireLock() throws InterruptedException, KeeperException {
    //創(chuàng)建鎖節(jié)點(diǎn)
    createLock();
    //嘗試獲取鎖
    attemptLock();
}

首先創(chuàng)建鎖節(jié)點(diǎn)，然后嘗試去取鎖。真正的邏輯都在這兩個(gè)方法中。

createLock()

先判斷鎖的根節(jié)點(diǎn)/Locks是否存在，不存在的話創(chuàng)建。然后在/Locks下創(chuàng)建有序臨時(shí)節(jié)點(diǎn)，并設(shè)置當(dāng)前的鎖路徑變量lockPath。

代碼如下：

private void createLock() throws KeeperException, InterruptedException {
    //如果根節(jié)點(diǎn)不存在，則創(chuàng)建根節(jié)點(diǎn)
    Stat stat = zkClient.exists(LOCK_ROOT_PATH, false);
    if (stat == null) {
        zkClient.create(LOCK_ROOT_PATH, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }
 
    // 創(chuàng)建EPHEMERAL_SEQUENTIAL類型節(jié)點(diǎn)
    String lockPath = zkClient.create(LOCK_ROOT_PATH + "/" + LOCK_NODE_NAME,
            Thread.currentThread().getName().getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,
            CreateMode.EPHEMERAL_SEQUENTIAL);
    System.out.println(Thread.currentThread().getName() + " 鎖創(chuàng)建: " + lockPath);
    this.lockPath=lockPath;
}

attemptLock()

這是最核心的方法，客戶端嘗試去獲取鎖，是對2.0版本邏輯的實(shí)現(xiàn)，這里就不再重復(fù)邏輯，直接看代碼：

private void attemptLock() throws KeeperException, InterruptedException {
    // 獲取Lock所有子節(jié)點(diǎn)，按照節(jié)點(diǎn)序號排序
    List<String> lockPaths = null;
    lockPaths = zkClient.getChildren(LOCK_ROOT_PATH, false);
    Collections.sort(lockPaths);
    int index = lockPaths.indexOf(lockPath.substring(LOCK_ROOT_PATH.length() + 1));
    // 如果lockPath是序號最小的節(jié)點(diǎn)，則獲取鎖
    if (index == 0) {
        System.out.println(Thread.currentThread().getName() + " 鎖獲得, lockPath: " + lockPath);
        return ;
    } else {
        // lockPath不是序號最小的節(jié)點(diǎn)，監(jiān)聽前一個(gè)節(jié)點(diǎn)
        String preLockPath = lockPaths.get(index - 1);
 
        Stat stat = zkClient.exists(LOCK_ROOT_PATH + "/" + preLockPath, watcher);
 
        // 假如前一個(gè)節(jié)點(diǎn)不存在了，比如說執(zhí)行完畢，或者執(zhí)行節(jié)點(diǎn)掉線，重新獲取鎖
        if (stat == null) {
            attemptLock();
        } else { // 阻塞當(dāng)前進(jìn)程，直到preLockPath釋放鎖，被watcher觀察到，notifyAll后，重新acquireLock
            System.out.println(" 等待前鎖釋放，prelocakPath："+preLockPath);
            synchronized (watcher) {
                watcher.wait();
            }
            attemptLock();
        }
    }
}

注意這一行代碼

Stat stat = zkClient.exists(LOCK_ROOT_PATH + "/" + preLockPath, watcher);

我們在獲取前一個(gè)節(jié)點(diǎn)的時(shí)候，同時(shí)設(shè)置了監(jiān)聽watcher。如果前鎖存在，則阻塞主線程。

watcher定義代碼如下：

private Watcher watcher = new Watcher() {
    @Override
    public void process(WatchedEvent event) {
        System.out.println(event.getPath() + " 前鎖釋放");
        synchronized (this) {
            notifyAll();
        }
    }
};

watcher只是notifyAll，讓主線程繼續(xù)執(zhí)行，以便再次調(diào)用attemptLock()，去嘗試獲取lock。如果沒有異常情況的話，此時(shí)當(dāng)前客戶端應(yīng)該能夠成功獲取鎖。

釋放鎖實(shí)現(xiàn)

釋放鎖原語實(shí)現(xiàn)很簡單，參照releaseLock()方法。代碼如下：

public void releaseLock() throws KeeperException, InterruptedException {
    zkClient.delete(lockPath, -1);
    zkClient.close();
    System.out.println(" 鎖釋放：" + lockPath);
}

關(guān)于分布式鎖的代碼到此就講解完了，我們再看下客戶端如何使用它。

我們創(chuàng)建一個(gè)TicketSeller類，作為客戶端來使用分布式鎖。

TicketSeller類

sell()

不帶鎖的業(yè)務(wù)邏輯方法，代碼如下：

private void sell(){
    System.out.println("售票開始");
    // 線程隨機(jī)休眠數(shù)毫秒，模擬現(xiàn)實(shí)中的費(fèi)時(shí)操作
    int sleepMillis = (int) (Math.random() * 2000);
    try {
        //代表復(fù)雜邏輯執(zhí)行了一段時(shí)間
        Thread.sleep(sleepMillis);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    System.out.println("售票結(jié)束");
}

僅是為了演示，sleep了一段時(shí)間。

sellTicketWithLock()

此方法中，加鎖后執(zhí)行業(yè)務(wù)邏輯，代碼如下：

public void sellTicketWithLock() throws KeeperException, InterruptedException, IOException {
    LockSample lock = new LockSample();
    lock.acquireLock();
    sell();
    lock.releaseLock();
}

測試入口

接下來我們寫一個(gè)main函數(shù)做測試：

public static void main(String[] args) throws KeeperException, InterruptedException, IOException {
    TicketSeller ticketSeller = new TicketSeller();
    for(int i=0;i<1000;i++){
        ticketSeller.sellTicketWithLock();
    }
}

main函數(shù)中我們循環(huán)調(diào)用ticketSeller.sellTicketWithLock()，執(zhí)行加鎖后的賣票邏輯。

測試方法

1、先啟動一個(gè)java程序運(yùn)行，可以看到日志輸出如下：

main 鎖創(chuàng)建: /Locks/Lock_0000000391
main 鎖獲得, lockPath: /Locks/Lock_0000000391
售票開始
售票結(jié)束
 鎖釋放：/Locks/Lock_0000000391
main 鎖創(chuàng)建: /Locks/Lock_0000000392
main 鎖獲得, lockPath: /Locks/Lock_0000000392
售票開始
售票結(jié)束
 鎖釋放：/Locks/Lock_0000000392
main 鎖創(chuàng)建: /Locks/Lock_0000000393
main 鎖獲得, lockPath: /Locks/Lock_0000000393
售票開始
售票結(jié)束
 鎖釋放：/Locks/Lock_0000000393

可見每次執(zhí)行都是按照鎖的順序執(zhí)行，而且由于只有一個(gè)進(jìn)程，并沒有鎖的爭搶發(fā)生。

2、我們再啟動一個(gè)同樣的程序，鎖的爭搶此時(shí)發(fā)生了，可以看到雙方的日志輸出如下：

程序1:

main 鎖獲得, lockPath: /Locks/Lock_0000000471
售票開始
售票結(jié)束
 鎖釋放：/Locks/Lock_0000000471
main 鎖創(chuàng)建: /Locks/Lock_0000000473
 等待前鎖釋放，prelocakPath：Lock_0000000472
/Locks/Lock_0000000472 前鎖釋放
main 鎖獲得, lockPath: /Locks/Lock_0000000473
售票開始
售票結(jié)束
 鎖釋放：/Locks/Lock_0000000473

可以看到Lock_0000000471執(zhí)行完成后，該進(jìn)程獲取的鎖為Lock_0000000473，這說明Lock_0000000472被另外一個(gè)進(jìn)程創(chuàng)建了。此時(shí)Lock_0000000473在等待前鎖釋放。Lock_0000000472釋放后，Lock_0000000473才獲得鎖，然后才執(zhí)行業(yè)務(wù)邏輯。

我們再看程序2的日志：

main 鎖獲得, lockPath: /Locks/Lock_0000000472
售票開始
售票結(jié)束
 鎖釋放：/Locks/Lock_0000000472
main 鎖創(chuàng)建: /Locks/Lock_0000000474
 等待前鎖釋放，prelocakPath：Lock_0000000473
/Locks/Lock_0000000473 前鎖釋放
main 鎖獲得, lockPath: /Locks/Lock_0000000474
售票開始
售票結(jié)束
 鎖釋放：/Locks/Lock_0000000474

可以看到，確實(shí)是進(jìn)程2獲取了Lock_0000000472。

zookeeper實(shí)現(xiàn)分布式鎖就先講到這。注意代碼只做演示用，并不適合生產(chǎn)環(huán)境使用。

代碼清單如下：

1、LockSample

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
 
import java.io.IOException;
import java.util.Collections;
import java.util.List;
 
public class LockSample {
 
    //ZooKeeper配置信息
    private ZooKeeper zkClient;
    private static final String LOCK_ROOT_PATH = "/Locks";
    private static final String LOCK_NODE_NAME = "Lock_";
    private String lockPath;
 
    // 監(jiān)控lockPath的前一個(gè)節(jié)點(diǎn)的watcher
    private Watcher watcher = new Watcher() {
        @Override
        public void process(WatchedEvent event) {
            System.out.println(event.getPath() + " 前鎖釋放");
            synchronized (this) {
                notifyAll();
            }
 
        }
    };
 
    public LockSample() throws IOException {
        zkClient= new ZooKeeper("localhost:2181", 10000, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                if(event.getState()== Event.KeeperState.Disconnected){
                    System.out.println("失去連接");
 
                }
            }
        });
    }
 
    //獲取鎖的原語實(shí)現(xiàn).
    public  void acquireLock() throws InterruptedException, KeeperException {
        //創(chuàng)建鎖節(jié)點(diǎn)
        createLock();
        //嘗試獲取鎖
        attemptLock();
    }
 
    //創(chuàng)建鎖的原語實(shí)現(xiàn)。在lock節(jié)點(diǎn)下創(chuàng)建該線程的鎖節(jié)點(diǎn)
    private void createLock() throws KeeperException, InterruptedException {
        //如果根節(jié)點(diǎn)不存在，則創(chuàng)建根節(jié)點(diǎn)
        Stat stat = zkClient.exists(LOCK_ROOT_PATH, false);
        if (stat == null) {
            zkClient.create(LOCK_ROOT_PATH, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
 
        // 創(chuàng)建EPHEMERAL_SEQUENTIAL類型節(jié)點(diǎn)
        String lockPath = zkClient.create(LOCK_ROOT_PATH + "/" + LOCK_NODE_NAME,
                Thread.currentThread().getName().getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println(Thread.currentThread().getName() + " 鎖創(chuàng)建: " + lockPath);
        this.lockPath=lockPath;
    }
 
    private void attemptLock() throws KeeperException, InterruptedException {
        // 獲取Lock所有子節(jié)點(diǎn)，按照節(jié)點(diǎn)序號排序
        List<String> lockPaths = null;
 
        lockPaths = zkClient.getChildren(LOCK_ROOT_PATH, false);
 
        Collections.sort(lockPaths);
 
        int index = lockPaths.indexOf(lockPath.substring(LOCK_ROOT_PATH.length() + 1));
 
        // 如果lockPath是序號最小的節(jié)點(diǎn)，則獲取鎖
        if (index == 0) {
            System.out.println(Thread.currentThread().getName() + " 鎖獲得, lockPath: " + lockPath);
            return ;
        } else {
            // lockPath不是序號最小的節(jié)點(diǎn)，監(jiān)控前一個(gè)節(jié)點(diǎn)
            String preLockPath = lockPaths.get(index - 1);
 
            Stat stat = zkClient.exists(LOCK_ROOT_PATH + "/" + preLockPath, watcher);
 
            // 假如前一個(gè)節(jié)點(diǎn)不存在了，比如說執(zhí)行完畢，或者執(zhí)行節(jié)點(diǎn)掉線，重新獲取鎖
            if (stat == null) {
                attemptLock();
            } else { // 阻塞當(dāng)前進(jìn)程，直到preLockPath釋放鎖，被watcher觀察到，notifyAll后，重新acquireLock
                System.out.println(" 等待前鎖釋放，prelocakPath："+preLockPath);
                synchronized (watcher) {
                    watcher.wait();
                }
                attemptLock();
            }
        }
    }
 
    //釋放鎖的原語實(shí)現(xiàn)
    public void releaseLock() throws KeeperException, InterruptedException {
        zkClient.delete(lockPath, -1);
        zkClient.close();
        System.out.println(" 鎖釋放：" + lockPath);
    }
 
 
}

2、TicketSeller

import org.apache.zookeeper.KeeperException;
import java.io.IOException;
public class TicketSeller {
    private void sell(){
        System.out.println("售票開始");
        // 線程隨機(jī)休眠數(shù)毫秒，模擬現(xiàn)實(shí)中的費(fèi)時(shí)操作
        int sleepMillis = (int) (Math.random() * 2000);
        try {
            //代表復(fù)雜邏輯執(zhí)行了一段時(shí)間
            Thread.sleep(sleepMillis);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("售票結(jié)束");
    }
 
    public void sellTicketWithLock() throws KeeperException, InterruptedException, IOException {
        LockSample lock = new LockSample();
        lock.acquireLock();
        sell();
        lock.releaseLock();
    }
 
    public static void main(String[] args) throws KeeperException, InterruptedException, IOException {
        TicketSeller ticketSeller = new TicketSeller();
        for(int i=0;i<1000;i++){
            ticketSeller.sellTicketWithLock();
 
        }
    }
}

以上就是ZooKeeper入門教程三分布式鎖實(shí)現(xiàn)及完整運(yùn)行源碼的詳細(xì)內(nèi)容，更多關(guān)于ZooKeeper分布式鎖實(shí)現(xiàn)源碼的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章:

最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线

ZooKeeper入門教程三分布式鎖實(shí)現(xiàn)及完整運(yùn)行源碼

目錄

1.0版本

2.0版本

LockSample類

構(gòu)造方法

獲取鎖實(shí)現(xiàn)

createLock()

attemptLock()

釋放鎖實(shí)現(xiàn)

TicketSeller類

sell()

sellTicketWithLock()

測試入口

測試方法

代碼清單如下：

1、LockSample

2、TicketSeller

相關(guān)文章

最新評論

大家感興趣的內(nèi)容

最近更新的內(nèi)容

常用在線小工具

最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线

ZooKeeper入門教程三分布式鎖實(shí)現(xiàn)及完整運(yùn)行源碼

目錄

1.0版本

2.0版本

LockSample類

構(gòu)造方法

獲取鎖實(shí)現(xiàn)

createLock()

attemptLock()

釋放鎖實(shí)現(xiàn)

TicketSeller類

sell()

sellTicketWithLock()

測試入口

測試方法

代碼清單如下：

1、LockSample

2、TicketSeller

相關(guān)文章

最新評論

大家感興趣的內(nèi)容

最近更新的內(nèi)容

常用在線小工具

2、TicketSeller