文档章节

zookeeper分布式锁的实现

天天顺利
 天天顺利
发布于 2015/07/03 17:00
字数 2017
阅读 3267
收藏 33

zookeeper分布式锁的实现

本文主要讲述在使用ZooKeeper进行分布式锁的实现过程中,如何有效的避免“羊群效应(herdeffect)”的出现。

最后有实现了代码,仅供参考

本文参考了《Hadoop权威指南》以及以下网页内容实现:

http://aliapp.blog.51cto.com/8192229/1328018

 

实现过程介绍如下:

一般的分布式锁实现

这里简单的讲下一般的分布式锁如何实现。具体的代码实现可以在这里看到:https://svn.apache.org/repos/asf/zookeeper/trunk/src/recipes/lock/

在之前的《ZooKeepe数据模型》一文中提到过,zookeeper中节点的创建类型有4类,这里我们重点关注下临时顺序节点。这种类型的节点有几下几个特性:

  1. 节点的生命周期和客户端会话绑定,即创建节点的客户端会话一旦失效,那么这个节点也会被清除。

  2. 每个父节点都会负责维护其子节点创建的先后顺序,并且如果创建的是顺序节点(SEQUENTIAL)的话,父节点会自动为这个节点分配一个整形数值,以后缀的形式自动追加到节点名中,作为这个节点最终的节点名。

利用上面这两个特性,我们来看下获取实现分布式锁的基本逻辑:

  1. 客户端调用create()方法创建名为“_locknode_/guid-lock-”的节点,需要注意的是,这里节点的创建类型需要设置为EPHEMERAL_SEQUENTIAL。

  2. 客户端调用getChildren(“_locknode_”)方法来获取所有已经创建的子节点,同时在这个节点上注册上子节点变更通知的Watcher。

  3. 客户端获取到所有子节点path之后,如果发现自己在步骤1中创建的节点是所有节点中序号最小的,那么就认为这个客户端获得了锁。

  4. 如果在步骤3中发现自己并非是所有子节点中最小的,说明自己还没有获取到锁,就开始等待,直到下次子节点变更通知的时候,再进行子节点的获取,判断是否获取锁。

释放锁的过程相对比较简单,就是删除自己创建的那个子节点即可。

问题所在

上面这个分布式锁的实现中,大体能够满足了一般的分布式集群竞争锁的需求。这里说的一般性场景是指集群规模不大,一般在10台机器以内。

不 过,细想上面的实现逻辑,我们很容易会发现一个问题,步骤4,“即获取所有的子点,判断自己创建的节点是否已经是序号最小的节点”,这个过程,在整个分布 式锁的竞争过程中,大量重复运行,并且绝大多数的运行结果都是判断出自己并非是序号最小的节点,从而继续等待下一次通知——这个显然看起来不怎么科学。客 户端无端的接受到过多的和自己不相关的事件通知,这如果在集群规模大的时候,会对Server造成很大的性能影响,并且如果一旦同一时间有多个节点的客户 端断开连接,这个时候,服务器就会像其余客户端发送大量的事件通知——这就是所谓的羊群效应。而这个问题的根源在于,没有找准客户端真正的关注点。

我们再来回顾一下上面的分布式锁竞争过程,它的核心逻辑在于:判断自己是否是所有节点中序号最小的。于是,很容易可以联想的到的是,每个节点的创建者只需要关注比自己序号小的那个节点。

改进后的分布式锁实现

下面是改进后的分布式锁实现,和之前的实现方式唯一不同之处在于,这里设计成每个锁竞争者,只需要关注”_locknode_”节点下序号比自己小的那个节点是否存在即可。实现如下:

  1. 客户端调用create()方法创建名为“_locknode_/guid-lock-”的节点,需要注意的是,这里节点的创建类型需要设置为EPHEMERAL_SEQUENTIAL。

  2. 客户端调用getChildren(“_locknode_”)方法来获取所有已经创建的子节点,注意,这里不注册任何Watcher。

  3. 客户端获取到所有子节点path之后,如果发现自己在步骤1中创建的节点序号最小,那么就认为这个客户端获得了锁。

  4. 如果在步骤3中发现自己并非所有子节点中最小的,说明自己还没有获取到锁。此时客户端需要找到比自己小的那个节点,然后对其调用exist()方法,同时注册事件监听。

  5. 之后当这个被关注的节点被移除了,客户端会收到相应的通知。这个时候客户端需要再次调用getChildren(“_locknode_”)方法来获取所有已经创建的子节点,确保自己确实是最小的节点了,然后进入步骤3

 

代码实现:

Zookeeper连接类代码

import java.io.IOException;
import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher.Event.KeeperState;

public class ConnectionWatcher implements Watcher
{
    private static final int SESSION_TIMEOUT = 5000;
    protected ZooKeeper zk;
    private CountDownLatch connectedSignal = new CountDownLatch(1);

    public void connect(String hosts) throws IOException, InterruptedException
    {
        zk = new ZooKeeper(hosts, SESSION_TIMEOUT, this);
        connectedSignal.await();
    }

    @Override
    public void process(WatchedEvent event)
    {
        if (event.getState() == KeeperState.SyncConnected)
        {
            connectedSignal.countDown();
        }
    }

    public void close() throws InterruptedException
    {
        zk.close();
    }
}

Java代码

import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

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.Ids;
import org.apache.zookeeper.data.Stat;

/**
 * http://graduter.iteye.com/blog/2024190
 * 本类简单实现了分布式锁的机制
 * 采用单机zookeeper服务器测试;运行多次本机程序,相当于多个客户端用户 全部启动完成后,第一个客户端的会话需要手动中断,相当于触发客户端宕机现象
 * 本类实现的分布式锁避免羊群效应(Herd Effect),具体可详见代码
 */
public class DistributedLock extends ConnectionWatcher
{
    public String join(String groupPath) throws KeeperException,InterruptedException
    {
        String path = groupPath + "/lock-" + zk.getSessionId() + "-";
        // 建立一个顺序临时节点
        String createdPath = zk.create(path, null/* data */, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println("Created " + createdPath);

        return createdPath;
    }

    /**
     * 检查本客户端是否得到了分布式锁
     *
     * @param groupPath
     * @param myName
     * @return
     * @throws KeeperException
     * @throws InterruptedException
     */
    public boolean checkState(String groupPath, String myName) throws KeeperException, InterruptedException
    {
        System.out.println("groupPath = " + groupPath);
        System.out.println("myName = " + myName);
        List<String> childList = zk.getChildren(groupPath, false);
        // myName = /zkRoot/_locknode_/lock-94103680129368072-0000000003
        String[] myStr = myName.split("-");
        long myId = Long.parseLong(myStr[2]);

        boolean minId = true;
        int index = 0;
        for (String childName : childList)
        {
            System.out.println(index + " \t " + childName);
            String[] str = childName.split("-");
            long id = Long.parseLong(str[2]);
            if (id < myId)
            {
                minId = false;
                break;
            }
            index ++;
        }

        if (minId)
        {
            System.out.println(new Date() + "我得到了分布锁,哈哈! myId:" + myId);
            return true;
        }
        else
        {
            System.out.println(new Date() + "继续努力吧,  myId:" + myId);
            return false;
        }
    }

    /**
     * 若本客户端没有得到分布式锁,则进行监听本节点前面的节点(避免羊群效应)
     *
     * @param groupPath
     * @param myName
     * @throws KeeperException
     * @throws InterruptedException
     */
    public void listenNode(final String groupPath, final String myName) throws KeeperException, InterruptedException
    {
        List<String> childList = zk.getChildren(groupPath, false);

        String[] myStr = myName.split("-");
        long myId = Long.parseLong(myStr[2]);

        List<Long> idList = new ArrayList<Long>();
        Map<Long, String> sessionMap = new HashMap<Long, String>();

        for (String childName : childList)
        {
            String[] str = childName.split("-");
            long id = Long.parseLong(str[2]);
            idList.add(id);
            sessionMap.put(id, str[1] + "-" + str[2]);
        }

        Collections.sort(idList);
        int i = idList.indexOf(myId);
        if (i <= 0)
        {
            throw new IllegalArgumentException("数据错误!");
        }

        // 得到前面的一个节点
        long headId = idList.get(i - 1);

        String headPath = groupPath + "/lock-" + sessionMap.get(headId);
        // 添加监听:/zkRoot/_locknode_/lock-94103680129368071-0000000002
        System.out.println("添加监听:" + headPath);

        Stat stat = zk.exists(headPath, new Watcher() {
            @Override
            public void process(WatchedEvent event)
            {
                System.out.println("已经触发了" + event.getType() + "事件!");
                try
                {
                    while (true)
                    {
                        if (checkState(groupPath, myName))
                        {
                            Thread.sleep(3000);
                            System.out.println(new Date() + " 系统关闭!");
                            System.exit(0);
                        }
                        Thread.sleep(3000);
                    }
                } catch (KeeperException e)
                {
                    e.printStackTrace();
                } catch (InterruptedException e)
                {
                    e.printStackTrace();
                }
            }
        });
        System.out.println(stat);
    }
    /**
     * 1.
     * Exception in thread "main" org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /zkRoot/_locknode_/lock-94103680129368068-
     * 2.
     * Exception in thread "main" org.apache.zookeeper.KeeperException$NoChildrenForEphemeralsException: KeeperErrorCode = NoChildrenForEphemerals for /zkRoot/_locknode_
     *
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception
    {
        DistributedLock joinGroup = new DistributedLock();
        joinGroup.connect("localhost:" + "2181");
        
        // zookeeper的根节点;运行本程序前,需要提前生成
        String groupName = "zkRoot";
        String memberName = "_locknode_";
        Stat stat = joinGroup.zk.exists("/" + groupName, true);
        if(stat == null)
        {
            joinGroup.zk.create("/" + groupName, groupName.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
        stat = joinGroup.zk.exists("/" + groupName + "/" + memberName, true);
        if(stat == null)
        {
            joinGroup.zk.create("/" + groupName + "/" + memberName, memberName.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
        String path = "/" + groupName + "/" + memberName;

        String myName = joinGroup.join(path);
        if (!joinGroup.checkState(path, myName)) {
            joinGroup.listenNode(path, myName);
        }

        Thread.sleep(Integer.MAX_VALUE);
        joinGroup.close();
    }
}

本文转载自:http://graduter.iteye.com/blog/2024190

天天顺利
粉丝 45
博文 167
码字总数 73007
作品 0
海淀
技术主管
私信 提问
加载中

评论(1)

素人派
素人派
学习了。
http://surenpi.com
分布式锁与实现(二)基于ZooKeeper实现

引言 ZooKeeper是一个分布式的,开放源码的分布式应用程序协调服务,是Google的Chubby一个开源的实现,是Hadoop和Hbase的重要组件。它是一个为分布式应用提供一致性服务的软件,提供的功能包...

rechardchensir
2018/10/08
0
0
分布式锁与实现(二)——基于ZooKeeper实现

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

刘祖鹏
2018/05/08
0
0
详解分布式协调服务 ZooKeeper

原文链接:https://draveness.me/zookeeper-chubby 在 2006 年,Google 发表了一篇名为《The Chubby lock service for loosely-coupled distributed systems[1]》的论文,其中描述了一个分布...

Docker
2018/11/27
0
0
大数据教程(3.3):zookeeper简介

一、概念 ZooKeeper是一个分布式的,开放源码的分布式应用程序协调服务,是Google的Chubby一个开源的实现,是Hadoop和Hbase的重要组件。它是一个为分布式应用提供一致性服务的软件,提供的功...

em_aaron
2018/07/28
0
0
分布式设计与开发(三)------高一致性服务ZooKeeper

分布式环境中大多数服务是允许部分失败,也允许数据不一致,但有些最基础的服务是需要高可靠性,高一致性的,这些服务是其他分布式服务运转的基础,比如naming service、分布式lock等,这些分...

山哥
2012/03/19
0
0

没有更多内容

加载失败,请刷新页面

加载更多

JWT学习总结

官方 https://jwt.io 英文原版 https://www.ietf.org/rfc/rfc7519.txt 或 https://tools.ietf.org/html/rfc7519 中文翻译 https://www.jianshu.com/p/10f5161dd9df 1. 概述 JSON Web Token(......

冷基
52分钟前
3
0
AOP的学习(1)

AOP 理解AOP编程思想(面向方法、面向切面) spring AOP的概念 方面 -- 功能 目标 -- 原有方法 通知 -- 对原有方法增强的方法 连接点 -- 可以用来连接通知的地方(方法) 切入点 -- 将用来插入...

太猪-YJ
今天
4
0
一张图看懂亮度、明度、光度、光亮度、明亮度

亮度、明度、光亮度,Luminance和Brightness、lightness其实都是一个意思,只是起名字太难了。 提出一个颜色模型后,由于明度的取值与别人的不同,为了表示区别所以就另想一个词而已。 因此在...

linsk1998
昨天
8
0
Python应用:python链表示例

前言 python链表应用源码示例,需要用到python os模块方法、函数和类的应用。 首先,先简单的来了解下什么是链表?链表是一种物理存储单元上非连续、非顺序的存储结构,数据元素的逻辑顺序是...

python小白1
昨天
4
0
Source Insight加载源码

Source Insight是一个图形化的源代码查看工具(当然也可以作为编译工具)。如果一个项目的源代码较多,此工具可以很方便地查找到源代码自建的依赖关系。 1.创建工程 下图为Snort源代码的文件...

天王盖地虎626
昨天
5
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部