Zookeeper(5)---分布式锁

基于临时序号节点来实现分布式锁

为什么要用临时节点呢？如果拿到锁的服务宕机了，会话失效ZK自己也会删除掉临时的序号节点，这样也不会阻塞其他服务。

流程：

1.在一个持久节点下面创建临时的序号节点作为锁节点，如：/lock/lockId00000001 /lock/lockId00000002

2.获取持久节点下面所有子节点，判断其最小的是不是自己，如果是自己则表示获取锁成功。

3.如果最小的结点不是自己，则阻塞等待，并对lock结点添加监听，如果结点数发生变化了，则说明有释放了锁。但是这里如果存在大量监听，当结点发生变化的时候，可能就会出现羊群效应。因为大家都监听了，同时会通知很多客户端，会造成ZK性能突然下降。

所以这里可以只监听自己前面的那个节点，排队执行，03监听02,02监听01

4.当锁释放，节点监听到变化之后，执行第2步。

释放锁只需要删除对应的临时节点，如果获取到锁的服务宕机了，因为是临时节点也会自己删除的。

代码实现：

package com.nijunyang.zookeeper.zklock;

import lombok.Data;

import org.I0Itec.zkclient.IZkDataListener;

import org.I0Itec.zkclient.ZkClient;

import java.util.List;

import java.util.stream.Collectors;

/**

 * Description:

 * Created by nijunyang on 2020/11/29 21:51

 */

public class ZKLock {

    private static final String SERVER = "192.168.0.67:2181";

    private ZkClient zkClient;

    private static final String ROOT_PATH = "/lock";

    public ZKLock() {

        zkClient = new ZkClient(SERVER, 5000, 20000);

        buildRoot();

    }

    private void buildRoot() {

        if (!zkClient.exists(ROOT_PATH)) {

            zkClient.createPersistent(ROOT_PATH);

        }

    }

    /**

     * 加锁

     *

     * @param lockId

     * @param timeout

     * @return

     */

    public Lock lock(String lockId, long timeout) {

        Lock lockNode = createLock(lockId);

        // 尝试激活锁

        tryActiveLock(lockNode);

        if (!lockNode.isActive()) {

            try {

                synchronized (lockNode) {

                    lockNode.wait(timeout);

                }

            } catch (InterruptedException e) {

                throw new RuntimeException(e);

            }

        }

        if (!lockNode.isActive()) {

            throw new RuntimeException("获取锁超时");

        }

        return lockNode;

    }

    /**

     * 释放锁

     *

     * @param lock

     */

    public void unlock(Lock lock) {

        if (lock.isActive()) {

            zkClient.delete(lock.getPath());

        }

    }

    /**

     * 激活锁

     * @param lockNode

     */

    private void tryActiveLock(Lock lockNode) {

        // 判断当前是否为最小节点

        List<String> list = zkClient.getChildren(ROOT_PATH)

                .stream()

                .sorted()

                .map(p -> ROOT_PATH + "/" + p)

                .collect(Collectors.toList());

        String firstNodePath = list.get(0);

        //第一个节点是当前节点，将锁设置为激活

        if (firstNodePath.equals(lockNode.getPath())) {

            lockNode.setActive(true);

        } else {

            //第一个节点不是当前节点，则监听前面一个节点

            String upNodePath = list.get(list.indexOf(lockNode.getPath()) - 1);

            zkClient.subscribeDataChanges(upNodePath, new IZkDataListener() {

                @Override

                public void handleDataChange(String dataPath, Object data) throws Exception {

                }

                @Override

                public void handleDataDeleted(String dataPath) throws Exception {

                    //监听之后继续尝试加锁，加锁成功唤醒之前的等待

                    tryActiveLock(lockNode);

                    synchronized (lockNode) {

                        if (lockNode.isActive()) {

                            lockNode.notify();

                        }

                    }

                    zkClient.unsubscribeDataChanges(upNodePath, this);

                }

            });

        }

    }

    /**

     * 创建锁节点

     *

     * @param lockId

     * @return Lock

     */

    private Lock createLock(String lockId) {

        String nodePath = zkClient.createEphemeralSequential(ROOT_PATH + "/" + lockId, "write");

        return new Lock(lockId, nodePath);

    }

    @Data

    public static class Lock {

        private String lockId;

        private String path;

        /**

         * 是否激活锁

         */

        private boolean active;

        public Lock(String lockId, String path) {

            this.lockId = lockId;

            this.path = path;

        }

    }

}

巴特西

Zookeeper(5)---分布式锁

最新文章

热门文章