基于Zookeeper的分布式事务锁

ZooKeeper是一个分布式的,开放源码的分布式应用程序协调服务,是Google的Chubby一个开源的实现,是Hadoop和Hbase的重要组件。它是一个为分布式应用提供一致性服务的软件,提供的功能包括:配置维护、域名服务、分布式同步、组服务等。

ZooKeeper的架构通过冗余服务实现高可用性。因此,如果第一次无应答,客户端就可以询问另一台ZooKeeper主机。ZooKeeper节点将它们的数据存储于一个分层的命名空间,非常类似于一个文件系统或一个前缀树结构。客户端可以在节点读写,从而以这种方式拥有一个共享的配置服务。更新是全序的。

基于ZooKeeper分布式锁的流程

(1)在zookeeper指定节点(locks)下创建临时顺序节点node_n

(2)获取locks下所有子节点children

(3)对子节点按节点自增序号从小到大排序

(4)判断本节点是不是第一个子节点,若是,则获取锁;若不是,则监听比该节点小的那个节点的删除事件

若监听事件生效,则回到第二步重新进行判断,直到获取到锁。

具体实现

下面就具体使用java和zookeeper实现分布式锁,操作zookeeper使用的是apache提供的zookeeper的包。

通过实现Watch接口,实现process(WatchedEvent event)方法来实施监控,使CountDownLatch来完成监控,在等待锁的时候使用CountDownLatch来计数,等到后进行countDown,停止等待,继续运行。

以下整体流程基本与上述描述流程一致,只是在监听的时候使用的是CountDownLatch来监听前一个节点。

代码如下

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;

public class  DistributedLock implements Lock,Watcher{

    private ZooKeeper zk = null;

    // 根节点    private String ROOT_LOCK = “/locks”;

    // 竞争的资源    private String lockName;

    // 等待的前一个锁    private String WAIT_LOCK;

    // 当前锁    private String CURRENT_LOCK;

    // 计数器    private CountDownLatch countDownLatch;

    private int sessionTimeout = 30000;

    private List<Exception> exceptionList = new ArrayList<Exception>();

    /**    * 配置分布式锁    *@paramconfig 连接的url    *@paramlockName 竞争资源    */           

   publicDistributedLock(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) {

                // 如果根节点不存在,则创建根节点                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();

        }

    }

    // 节点监视器    publicvoidprocess(WatchedEvent event){

        if (this.countDownLatch != null) {

            this.countDownLatch.countDown();

        }

    }

    publicvoidlock(){

        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();

        }

    }

    publicbooleantryLock(){

        try {

            String splitStr = “_lock_”;

            if (lockName.contains(splitStr)) {

                throw new LockException(“锁名有误”);

            }

            // 创建临时有序节点            CURRENT_LOCK = zk.create(ROOT_LOCK + “/” + lockName + splitStr, new byte[0],

                    ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

            System.out.println(CURRENT_LOCK + ” 已经创建”);

            // 取所有子节点            List<String> subNodes = zk.getChildren(ROOT_LOCK, false);

            // 取出所有lockName的锁            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);

            // 若当前节点为最小节点,则获取锁成功            if (CURRENT_LOCK.equals(ROOT_LOCK + “/” + lockObjects.get(0))) {

                return true;

            }

            // 若不是最小节点,则找到自己的前一个节点            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;

    }

    publicbooleantryLock(longtimeout, TimeUnit unit){

        try {

            if (this.tryLock()) {

                return true;

            }

            return waitForLock(WAIT_LOCK, timeout);

        } catch (Exception e) {

            e.printStackTrace();

        }

        return false;

    }

    // 等待锁    privatebooleanwaitForLock(String prev,longwaitTime)throwsKeeperException, 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);

            // 计数等待,若等到前一个节点消失,则precess中进行countDown,停止等待,获取锁            this.countDownLatch.await(waitTime, TimeUnit.MILLISECONDS);

            this.countDownLatch = null;

            System.out.println(Thread.currentThread().getName() + ” 等到了锁”);

        }

        return true;

    }

    publicvoidunlock(){

        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();

        }

    }

    publicConditionnewCondition(){

        return null;

    }

    publicvoidlockInterruptibly()throwsInterruptedException{

        this.lock();

    }

    public classLockExceptionextendsRuntimeException{

        private static final long serialVersionUID = 1L;

        publicLockException(String e){

            super(e);

        }

        publicLockException(Exception e){

            super(e);

        }

    }

}

测试代码

public classTest{

    static int n = 500;

    publicstaticvoidsecskill(){

        System.out.println(–n);

    }

    publicstaticvoidmain(String[] args){

        Runnable runnable = new Runnable() {

            publicvoidrun(){

                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();

        }

    }

}

总体来说,如果了解到整个实现流程,使用zookeeper实现分布式锁并不是很困难,不过这也只是一个简单的实现,与前面实现Redis实现相比,本实现的稳定性更强,这是因为zookeeper的特性所致,在外界看来,zookeeper集群中每一个节点都是一致的。

https://www.cnblogs.com/liuyang0/p/6800538.html

    原文作者:代码之尖
    原文地址: https://www.jianshu.com/p/5870515e4cf6
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞