zookeeper实现分布式锁怎么实现?步骤有哪些?

TheDisguiser 2020-05-03 15:33:36 java常见问答 6336

在一般项目开发中都需要解决并发问题,一般情况下,JDK自带的lock与同步已经足够解决这个问题,但在分布式下就非常乏力了,下面就跟小编一起来看看如何在zookeeper下实现分布式锁吧。

Zookeeper实现分布式锁

当多个进程一起访问共享资源时,我们可以通过zookeeper来实现分布式锁,下面是步骤:

a.建立一个节点,节点类型为持久节点;

b.当进程需要访问共享资源时,调用分布式锁的lock()或tryLock()方法获得锁,这个时候会在第一步创建的lock节点下建立相应的顺序子节点,节点类型为临时顺序节点(EPHEMERAL_SEQUENTIAL),通过组成特定的名字name+lock+顺序号。

c.建立子节点后,对lock下面的所有以name开头的子节点进行排序,判断刚刚建立的子节点顺序号是否是最小的节点,假如是最小节点,则获得该锁对资源进行访问。

d.如果不是该节点,就获得该节点的上一顺序节点,并给该节点是否存在注册监听事件。同时在这里阻塞。等待监听事件的发生,获得锁控制权。

e.当调用完共享资源后,调用unlock()方法,关闭zookeeper,进而可以引发监听事件,释放该锁。

例:

package cn.wpeace.zktest;
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;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.data.Stat;
/**
 * @author peace
 *
 */
public class DistributedLock implements Lock, Watcher
{
    private ZooKeeper zk;
    private String root = "/locks"; //根
    private String lockName; //竞争资源的标志
    private String waitNode; //等待前一个锁
    private String myZnode; //当前锁
    private CountDownLatch latch; //计数器
    private CountDownLatch connectedSignal = new CountDownLatch(1);
    private int sessionTimeout = 30000;
    /**
     * 创建分布式锁,使用前请确认config配置的zookeeper服务可用
     * @param config 192.168.1.127:2181
     * @param lockName 竞争资源标志,lockName中不能包含单词_lock_
     */
    public DistributedLock(String config, String lockName)
    {
        this.lockName = lockName;
        // 创建一个与服务器的连接
        try
        {
            zk = new ZooKeeper(config, sessionTimeout, this);
            connectedSignal.await();
            Stat stat = zk.exists(root, false); //此去不执行 Watcher
            if (stat == null)
            {
                // 创建根节点
                zk.create(root, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        }
        catch (IOException e)
        {
            throw new LockException(e);
        }
        catch (KeeperException e)
        {
            throw new LockException(e);
        }
        catch (InterruptedException e)
        {
            throw new LockException(e);
        }
    }
    /**
     * zookeeper节点的监视器
     */
    public void process(WatchedEvent event)
    {
        //建立连接用
        if (event.getState() == KeeperState.SyncConnected)
        {
            connectedSignal.countDown();
            return;
        }
        //其他线程放弃锁的标志
        if (this.latch != null)
        {
            this.latch.countDown();
        }
    }
    public void lock()
    {
        try
        {
            if (this.tryLock())
            {
                System.out.println("Thread " + Thread.currentThread()
                    .getId() + " " + myZnode + " get lock true");
                return;
            }
            else
            {
                waitForLock(waitNode, sessionTimeout); //等待锁
            }
        }
        catch (KeeperException e)
        {
            throw new LockException(e);
        }
        catch (InterruptedException e)
        {
            throw new LockException(e);
        }
    }
    public boolean tryLock()
    {
        try
        {
            String splitStr = "_lock_";
            if (lockName.contains(splitStr))
                throw new LockException("lockName can not contains \u000B");
            //创建临时子节点
            myZnode = zk.create(root + "/" + lockName + splitStr, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            System.out.println(myZnode + " is created ");
            //取出所有子节点
            List < String > subNodes = zk.getChildren(root, false);
            //取出所有lockName的锁
            List < String > lockObjNodes = new ArrayList < String > ();
            for (String node: subNodes)
            {
                String _node = node.split(splitStr)[0];
                if (_node.equals(lockName))
                {
                    lockObjNodes.add(node);
                }
            }
            Collections.sort(lockObjNodes);
            if (myZnode.equals(root + "/" + lockObjNodes.get(0)))
            {
                //如果是最小的节点,则表示取得锁
                System.out.println(myZnode + "==" + lockObjNodes.get(0));
                return true;
            }
            //如果不是最小的节点,找到比自己小1的节点
            String subMyZnode = myZnode.substring(myZnode.lastIndexOf("/") + 1);
            waitNode = lockObjNodes.get(Collections.binarySearch(lockObjNodes, subMyZnode) - 1); //找到前一个子节点
        }
        catch (KeeperException e)
        {
            throw new LockException(e);
        }
        catch (InterruptedException e)
        {
            throw new LockException(e);
        }
        return false;
    }
    public boolean tryLock(long time, TimeUnit unit)
    {
        try
        {
            if (this.tryLock())
            {
                return true;
            }
            return waitForLock(waitNode, time);
        }
        catch (Exception e)
        {
            e.printStackTrace();
        }
        return false;
    }
    private boolean waitForLock(String lower, long waitTime) throws InterruptedException, KeeperException
    {
        Stat stat = zk.exists(root + "/" + lower, true); //同时注册监听。
        //判断比自己小一个数的节点是否存在,如果不存在则无需等待锁,同时注册监听
        if (stat != null)
        {
            System.out.println("Thread " + Thread.currentThread()
                .getId() + " waiting for " + root + "/" + lower);
            this.latch = new CountDownLatch(1);
            this.latch.await(waitTime, TimeUnit.MILLISECONDS); //等待,这里应该一直等待其他线程释放锁
            this.latch = null;
        }
        return true;
    }
    public void unlock()
    {
        try
        {
            System.out.println("unlock " + myZnode);
            zk.delete(myZnode, -1);
            myZnode = null;
            zk.close();
        }
        catch (InterruptedException e)
        {
            e.printStackTrace();
        }
        catch (KeeperException e)
        {
            e.printStackTrace();
        }
    }
    public void lockInterruptibly() throws InterruptedException
    {
        this.lock();
    }
    public Condition newCondition()
    {
        return null;
    }
    public class LockException extends RuntimeException
    {
        private static final long serialVersionUID = 1 L;
        public LockException(String e)
        {
            super(e);
        }
        public LockException(Exception e)
        {
            super(e);
        }
    }
}

测试:

 DistributedLock lock = new DistributedLock("192.168.1.127:2181", "lock");
 lock.lock();
 //共享资源
 if (lock != null)
     lock.unlock();

以上就是zookeeper的全部内容了,更多Java常见问答内容请持续关注我们了解吧。