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

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

基于ZooKeeper分布式鎖

1. 在zookeeper指定節(jié)點(diǎn)（locks）下創(chuàng)建臨時(shí)順序節(jié)點(diǎn)node_n

2. 獲取locks下所有子節(jié)點(diǎn)children

3. 對(duì)子節(jié)點(diǎn)按節(jié)點(diǎn)自增序號(hào)從小到大排序

4. 判斷本節(jié)點(diǎn)是不是第一個(gè)子節(jié)點(diǎn)，若是，則獲取鎖；若不是，則監(jiān)聽比該節(jié)點(diǎn)小的那個(gè)節(jié)點(diǎn)的刪除事件

5. 若監(jiān)聽事件生效，則回到第二步重新進(jìn)行判斷，直到獲取到鎖。

具體實(shí)現(xiàn)

• 下面就具體使用java和zookeeper實(shí)現(xiàn)分布式鎖，操作zookeeper使用的是apache提供的zookeeper的包。

• 通過實(shí)現(xiàn)Watch接口，實(shí)現(xiàn)process(WatchedEvent event) 方法來實(shí)施監(jiān)控，使CountDownLatch來完成監(jiān)控，在等待鎖的時(shí)候使用CountDownLatch來計(jì)數(shù)，等到后進(jìn)行countDown，停止等待，繼續(xù)運(yùn)行。

• 以下整體流程基本與上述描述流程一致，只是在監(jiān)聽的時(shí)候使用的是CountDownLatch來監(jiān)聽前一個(gè)節(jié)點(diǎn)。

分布式鎖

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對(duì)象操作
	private ZooKeeper zk = null;
    // 根節(jié)點(diǎn)
    private String ROOT_LOCK = "/locks";
    // 競(jìng)爭(zhēng)的資源
    private String lockName;
    // 等待的前一個(gè)鎖
    private String WAIT_LOCK;
    // 當(dāng)前鎖
    private String CURRENT_LOCK;
    // 計(jì)數(shù)器
    private CountDownLatch countDownLatch;
    private int sessionTimeout = 30000;
    private List<Exception> exceptionList = new ArrayList<Exception>();
    /**
     * 配置分布式鎖
     * @param config 連接的url
     * @param lockName 競(jìng)爭(zhēng)資源
     */
    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é)點(diǎn)不存在，則創(chuàng)建根節(jié)點(diǎn)
                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é)點(diǎn)監(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)建臨時(shí)有序節(jié)點(diǎn)
            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é)點(diǎn)
            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);
           // 若當(dāng)前節(jié)點(diǎn)為最小節(jié)點(diǎn)，則獲取鎖成功
            if (CURRENT_LOCK.equals(ROOT_LOCK + "/" + lockObjects.get(0))) {
                return true;
            }
       // 若不是最小節(jié)點(diǎn)，則找到自己的前一個(gè)節(jié)點(diǎn)
       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);
          // 計(jì)數(shù)等待，若等到前一個(gè)節(jié)點(diǎn)消失，則precess中進(jìn)行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);
        }
   }
}

測(cè)試代碼

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() + "正在運(yùn)行");
                } finally {
                    if (lock != null) {
                        lock.unlock();
                    }
                }
            }
        };

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

運(yùn)行結(jié)果：

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