ZooKeeper中怎樣實現(xiàn)分布式鎖

這篇文章將為大家詳細講解有關(guān)ZooKeeper中怎樣實現(xiàn)分布式鎖，文章內(nèi)容質(zhì)量較高，因此小編分享給大家做個參考，希望大家閱讀完這篇文章后對相關(guān)知識有一定的了解。

創(chuàng)新互聯(lián)是一家以網(wǎng)絡技術(shù)公司，為中小企業(yè)提供網(wǎng)站維護、成都網(wǎng)站建設、網(wǎng)站建設、網(wǎng)站備案、服務器租用、域名與空間、軟件開發(fā)、成都小程序開發(fā)等企業(yè)互聯(lián)網(wǎng)相關(guān)業(yè)務，是一家有著豐富的互聯(lián)網(wǎng)運營推廣經(jīng)驗的科技公司，有著多年的網(wǎng)站建站經(jīng)驗，致力于幫助中小企業(yè)在互聯(lián)網(wǎng)讓打出自已的品牌和口碑，讓企業(yè)在互聯(lián)網(wǎng)上打開一個面向全國乃至全球的業(yè)務窗口：建站聯(lián)系電話：028-86922220

基于ZooKeeper分布式鎖

在zookeeper指定節(jié)點（locks）下創(chuàng)建臨時順序節(jié)點node_n
獲取locks下所有子節(jié)點children
對子節(jié)點按節(jié)點自增序號從小到大排序
判斷本節(jié)點是不是第一個子節(jié)點，若是，則獲取鎖；若不是，則監(jiān)聽比該節(jié)點小的那個節(jié)點的刪除事件
若監(jiān)聽事件生效，則回到第二步重新進行判斷，直到獲取到鎖。

具體實現(xiàn)

下面就具體使用java和zookeeper實現(xiàn)分布式鎖，操作zookeeper使用的是apache提供的zookeeper的包。
通過實現(xiàn)Watch接口，實現(xiàn)process(WatchedEvent event)方法來實施監(jiān)控，使CountDownLatch來完成監(jiān)控，在等待鎖的時候使用CountDownLatch來計數(shù)，等到后進行countDown，停止等待，繼續(xù)運行。
以下整體流程基本與上述描述流程一致，只是在監(jiān)聽的時候使用的是CountDownLatch來監(jiān)聽前一個節(jié)點。

分布式鎖

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

/**
 * Created by liuyang on 2017/4/20.
 */
public class DistributedLock implements Lock, Watcher {
    //Zk對象操作
	private ZooKeeper zk = null;
    // 根節(jié)點
    private String ROOT_LOCK = "/locks";
    // 競爭的資源
    private String lockName;
    // 等待的前一個鎖
    private String WAIT_LOCK;
    // 當前鎖
    private String CURRENT_LOCK;
    // 計數(shù)器
    private CountDownLatch countDownLatch;
    private int sessionTimeout = 30000;
    private List<Exception> exceptionList = new ArrayList<Exception>();
    /**
     * 配置分布式鎖
     * @param config 連接的url
     * @param lockName 競爭資源
     */
    public DistributedLock(String config, String lockName) {
        this.lockName = lockName;
        try {
            // 連接zookeeper
            zk = new ZooKeeper(config, sessionTimeout, this);
            Stat stat = zk.exists(ROOT_LOCK, false);
            if (stat == null) {
                // 如果根節(jié)點不存在，則創(chuàng)建根節(jié)點
                zk.create(ROOT_LOCK, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, 
                 CreateMode.PERSISTENT);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }
    // 節(jié)點監(jiān)視器
    public void process(WatchedEvent event) {
        if (this.countDownLatch != null) {
            this.countDownLatch.countDown();
        }
    }
    public void lock() {
        if (exceptionList.size() > 0) {
            throw new LockException(exceptionList.get(0));
        }
        try {
            if (this.tryLock()) {
                System.out.println(Thread.currentThread().getName() + " " + lockName 
+ "獲得了鎖");
               return;
            } else {
                // 等待鎖
                waitForLock(WAIT_LOCK, sessionTimeout);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    public boolean tryLock() {
        try {
            String splitStr = "_lock_";
            if (lockName.contains(splitStr)) {
                throw new LockException("鎖名有誤");
            }
            // 創(chuàng)建臨時有序節(jié)點
            CURRENT_LOCK = zk.create(ROOT_LOCK + "/" + lockName + splitStr, 
new byte[0],ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            	System.out.println(CURRENT_LOCK + " 已經(jīng)創(chuàng)建");
            // 取所有子節(jié)點
            List<String> subNodes = zk.getChildren(ROOT_LOCK, false);
            // 取出所有l(wèi)ockName的鎖
            List<String> lockObjects = new ArrayList<String>();
            for (String node : subNodes) {
                String _node = node.split(splitStr)[0];
                if (_node.equals(lockName)) {
                    lockObjects.add(node);
                }
            }
            Collections.sort(lockObjects);
            System.out.println(Thread.currentThread().getName() + " 的鎖是 " 
+CURRENT_LOCK);
           // 若當前節(jié)點為最小節(jié)點，則獲取鎖成功
            if (CURRENT_LOCK.equals(ROOT_LOCK + "/" + lockObjects.get(0))) {
                return true;
            }
       // 若不是最小節(jié)點，則找到自己的前一個節(jié)點
       String prevNode = CURRENT_LOCK.substring(CURRENT_LOCK.lastIndexOf("/")+ 1);
       WAIT_LOCK = lockObjects.get(Collections.binarySearch(lockObjects, prevNode) - 1)
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
        return false;
    }

    public boolean tryLock(long timeout, TimeUnit unit) {
        try {
            if (this.tryLock()) {
                return true;
            }
            return waitForLock(WAIT_LOCK, timeout);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    // 等待鎖
    private boolean waitForLock(String prev, long waitTime) throws KeeperException, InterruptedException {
        Stat stat = zk.exists(ROOT_LOCK + "/" + prev, true);
        if (stat != null) {
            System.out.println(Thread.currentThread().getName() + "等待鎖 " + ROOT_LOCK + "/" + prev);
            this.countDownLatch = new CountDownLatch(1);
          // 計數(shù)等待，若等到前一個節(jié)點消失，則precess中進行countDown，停止等待，獲取鎖
            this.countDownLatch.await(waitTime, TimeUnit.MILLISECONDS);
            this.countDownLatch = null;
            System.out.println(Thread.currentThread().getName() + " 等到了鎖");
        }
        return true;
    }

    public void unlock() {
        try {
            System.out.println("釋放鎖 " + CURRENT_LOCK);
            zk.delete(CURRENT_LOCK, -1);
            CURRENT_LOCK = null;
            zk.close();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    public Condition newCondition() {
        return null;
    }

    public void lockInterruptibly() throws InterruptedException {
        this.lock();
    }


    public class LockException extends RuntimeException {
        private static final long serialVersionUID = 1L;
        public LockException(String e){
            super(e);
        }
        public LockException(Exception e){
            super(e);
        }
   }
}

測試代碼

public class Test {
    static int n = 500;

    public static void secskill() {
        System.out.println(--n);
    }

    public static void main(String[] args) {
        
        Runnable runnable = new Runnable() {
            public void run() {
                DistributedLock lock = null;
                try {
                    lock = new DistributedLock("127.0.0.1:2181", "test1");
                    lock.lock();
                    secskill();
                    System.out.println(Thread.currentThread().getName() + "正在運行");
                } finally {
                    if (lock != null) {
                        lock.unlock();
                    }
                }
            }
        };

        for (int i = 0; i < 10; i++) {
            Thread t = new Thread(runnable);
            t.start();
        }
    }
}

運行結(jié)果：

ZooKeeper中怎樣實現(xiàn)分布式鎖

關(guān)于ZooKeeper中怎樣實現(xiàn)分布式鎖就分享到這里了，希望以上內(nèi)容可以對大家有一定的幫助，可以學到更多知識。如果覺得文章不錯，可以把它分享出去讓更多的人看到。

網(wǎng)頁標題：ZooKeeper中怎樣實現(xiàn)分布式鎖
URL標題：http://muchs.cn/article30/phoepo.html

成都網(wǎng)站建設公司_創(chuàng)新互聯(lián)，為您提供商城網(wǎng)站、域名注冊、建站公司、營銷型網(wǎng)站建設、云服務器、小程序開發(fā)

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時需注明來源：創(chuàng)新互聯(lián)

猜你還喜歡下面的內(nèi)容