Zookeeper實現(xiàn)分布式鎖

之前寫過利用Redis實現(xiàn)分布式鎖的文章,剛好最近在學(xué)習(xí)ZooKpeer的一些知識,想著利用Zookeeper去實現(xiàn)分布式鎖,查閱了一些文章,發(fā)現(xiàn)大家一致認為高并發(fā)場景下并不適合用ZooKeeper去實現(xiàn),性能存在問題。為什么呢?ZooKeeper的性能問題,知乎上有人給出了兩條:

1. 本身不是為高可用性設(shè)計,master撐不住高流量容易導(dǎo)致系統(tǒng)crash。
2. Paxos算法的復(fù)雜性,選舉過程或許緩慢,結(jié)合上一點,動輒需要重新選舉master導(dǎo)致耗時過長。

有人將ZooKeeper類比于AQS(AbstractQueuedSynchronizer)。AQS是完成多線程協(xié)同一致的利器,ZooKeeper是完成跨機器多進程之間的協(xié)同一致性的利器,也確實是有那么點相似之處(這里純屬瞎扯啊,但不得不說佩服別人的聯(lián)想能力)。

image.png

image.png

Zookeeper實現(xiàn)分布式鎖的大致思路是:

1. 每個客戶端都去zookeeper上創(chuàng)建臨時的順序節(jié)點
2. 客戶端判斷當(dāng)前自己創(chuàng)建的節(jié)點是不是最小的
3. 如果是的話,就獲得了執(zhí)行當(dāng)前任務(wù)的鎖
4. 如果不是的話,就找到比自己小的節(jié)點,然后進行監(jiān)聽,如果被刪除的話,就可以獲得鎖

了解原理之后,我們得去編碼實現(xiàn)分布式鎖:

package com.meituan.ZooKeeperDistributeLock;

import org.apache.zookeeper.*;

import java.io.IOException;
import java.util.concurrent.*;

/**
 * @Author: taomin
 * @Create_time: 2018-03-11
 * @Description: zookeeper實現(xiàn)分布式鎖
 */
public class LockTest {

    private String zkQurom = "127.0.0.1:2181";
    private String lockNameSpace = "/home";
    private final String nodeString = lockNameSpace + "/test";
    private ZooKeeper zk;

    public LockTest() {
        try {
            zk = new ZooKeeper(zkQurom, 6000, new Watcher() {
                public void process(WatchedEvent watchedEvent) {
                    System.out.println("Receive event " + watchedEvent);
                    if (Event.KeeperState.SyncConnected == watchedEvent.getState())
                        System.out.println("connection is established...");
                }
            });
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private void watchNode(final String nodeString, final Thread thread) throws InterruptedException {
        try {
            zk.exists(nodeString, new Watcher() {
                public void process(WatchedEvent watchedEvent) {
                    System.out.println("==" + watchedEvent.toString());
                    if (watchedEvent.getType() == Event.EventType.NodeDeleted) {
                        System.out.println(Thread.currentThread().getName() + " released Lock ==============");
                        thread.interrupt();
                    }
                    try {
                        zk.exists(nodeString, new Watcher() {
                            public void process(WatchedEvent watchedEvent) {
                                System.out.println("==" + watchedEvent.toString());
                                if (watchedEvent.getType() == Event.EventType.NodeDeleted) {
                                    System.out.printf(Thread.currentThread().getName() + " released Lock ==============");
                                    thread.interrupt();
                                }
                                try {
                                    zk.exists(nodeString, true);
                                } catch (KeeperException e) {
                                    e.printStackTrace();
                                } catch (InterruptedException e) {
                                    e.printStackTrace();
                                }
                            }
                        });
                    } catch (KeeperException e) {
                        e.printStackTrace();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    private void ensureRootPath() throws InterruptedException {
        try {
            if (zk.exists(lockNameSpace, true) == null) {
                zk.create(lockNameSpace, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    public void unlock() {
        try {
            zk.delete(nodeString, -1);
            System.out.println(Thread.currentThread().getId() + " release Lock ==============");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    public boolean lock() throws InterruptedException {
        String path = null;
        ensureRootPath();
        watchNode(nodeString, Thread.currentThread());
        while (true) {
            try {
                path = zk.create(nodeString, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
            } catch (KeeperException e) {
                System.out.println(Thread.currentThread().getName() + "  getting Lock but can not get ==============");
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException ex) {
                    System.out.println("thread is notify");
                }
            }
            if (path != null && !path.trim().equals("")) {
                System.out.println(Thread.currentThread().getName() + "  get Lock ==============");
                return true;
            }
        }
    }

    public static void main(String args[]) throws InterruptedException {
        BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(10);
        ThreadPoolExecutor service = new ThreadPoolExecutor(5, 10,
                60, TimeUnit.MICROSECONDS, queue);
        for (int i = 0; i < 4; i++) {
            service.execute(new Runnable() {
                public void run() {
                    LockTest test = new LockTest();
                    try {
                        test.lock();
                        Thread.sleep(3000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    test.unlock();
                }
            });
        }
        service.shutdown();
    }
}

輸出日志,從日志里面可以清楚看到線程競爭鎖的過程。

Receive event WatchedEvent state:SyncConnected type:None path:null
connection is established...
Receive event WatchedEvent state:SyncConnected type:None path:null
connection is established...
Receive event WatchedEvent state:SyncConnected type:None path:null
connection is established...
Receive event WatchedEvent state:SyncConnected type:None path:null
connection is established...
==WatchedEvent state:SyncConnected type:NodeCreated path:/home/test
==WatchedEvent state:SyncConnected type:NodeCreated path:/home/test
==WatchedEvent state:SyncConnected type:NodeCreated path:/home/test
==WatchedEvent state:SyncConnected type:NodeCreated path:/home/test
pool-1-thread-2  get Lock ==============

pool-1-thread-1  getting Lock but can not get ==============

pool-1-thread-4  getting Lock but can not get ==============

pool-1-thread-3  getting Lock but can not get ==============

==WatchedEvent state:SyncConnected type:NodeDeleted path:/home/test
==WatchedEvent state:SyncConnected type:NodeDeleted path:/home/test
pool-1-thread-3-EventThread released Lock ==============
==WatchedEvent state:SyncConnected type:NodeDeleted path:/home/test
pool-1-thread-2-EventThread released Lock ==============
==WatchedEvent state:SyncConnected type:NodeDeleted path:/home/test
thread is notify
pool-1-thread-2 release Lock ==============

pool-1-thread-4-EventThread released Lock ==============
pool-1-thread-1-EventThread released Lock ==============
thread is notify
thread is notify
Receive event WatchedEvent state:SyncConnected type:NodeCreated path:/home/test
connection is established...
Receive event WatchedEvent state:SyncConnected type:NodeCreated path:/home/test
connection is established...
pool-1-thread-3  get Lock ==============

pool-1-thread-4  getting Lock but can not get ==============

pool-1-thread-1  getting Lock but can not get ==============

Receive event WatchedEvent state:SyncConnected type:NodeDeleted path:/home/test
connection is established...
Receive event WatchedEvent state:SyncConnected type:NodeDeleted path:/home/test
connection is established...
pool-1-thread-3 release Lock ==============

pool-1-thread-4  get Lock ==============

pool-1-thread-1  getting Lock but can not get ==============

pool-1-thread-4 release Lock ==============

pool-1-thread-1  get Lock ==============

pool-1-thread-1 release Lock ==============

接觸Zookeeper不久,理解不夠深刻,寫文章除了記錄自己的所見所聞外,同時也想借此來勉勵自己不放棄學(xué)習(xí),每天一點點進步,明天會有不一樣的自己。
end~

最后編輯于
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成,瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明:文章內(nèi)容(如有圖片或視頻亦包括在內(nèi))由作者上傳并發(fā)布,文章內(nèi)容僅代表作者本人觀點,簡書系信息發(fā)布平臺,僅提供信息存儲服務(wù)。

友情鏈接更多精彩內(nèi)容