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

时间： 2024-12-29 12:00:38

zookeeper分布式锁的实现的相关文章

ZooKeeper分布式锁浅谈（一）

一.概述 清明节的时候写了一篇分布式锁概述,里面介绍了分布式锁实现的几种方式,其实那时候我一直沉迷于使用redis的悲观锁和乐观锁来实现分布式锁,直到一个血案的引发才让我重新认识了redis分布式锁的弊端,所以才痛定思痛潜心研究Zookeeper:自己装了三台Centos虚拟机,搭建了ZooKeeper集群.二.ZooKeeper基本概念1.前言 ZooKeeper 是一个开源的分布式协调服务,由雅虎创建,是 Google Chubby 的开源实现.分布式应用程序可以基于 ZooKeeper 实

[转载] zookeeper 分布式锁服务

转载自http://www.cnblogs.com/shanyou/archive/2012/09/22/2697818.html 分布式锁服务在大家的项目中或许用的不多,因为大家都把排他放在数据库那一层来挡.当大量的行锁.表锁.事务充斥着数据库的时候.一般web应用很多的瓶颈都在数据库上,这里给大家介绍的是减轻数据库锁负担的一种方案,使用zookeeper分布式锁服务. zookeeper是hadoop下面的一个子项目, 用来协调跟hadoop相关的一些分布式的框架, 如hadoop, hiv

分布式锁（一） Zookeeper分布式锁

什么是Zookeeper? Zookeeper(业界简称zk)是一种提供配置管理.分布式协同以及命名的中心化服务,这些提供的功能都是分布式系统中非常底层且必不可少的基本功能,但是如果自己实现这些功能而且要达到高吞吐.低延迟同时还要保持一致性和可用性,实际上非常困难.因此zookeeper提供了这些功能,开发者在zookeeper之上构建自己的各种分布式系统. 虽然zookeeper的实现比较复杂,但是它提供的模型抽象却是非常简单的.Zookeeper提供一个多层级的节点命名空间(节点称为znod

ZooKeeper分布式锁简单实践

ZooKeeper分布式锁的实现原理 在分布式解决方案中,Zookeeper是一个分布式协调工具.当多个JVM客户端,同时在ZooKeeper上创建相同的一个临时节点,因为临时节点路径是保证唯一,只要谁能够创建节点成功,谁就能够获取到锁.没有创建成功节点,就会进行等待,当释放锁的时候,采用事件通知给客户端重新获取锁资源.如果请求超时直接返回给客户端超时,重新请求即可. 代码实现为了更好的展现效果,我这里设置每个线程请求需要1s,请求超时时间为30s. 首先我们先写一个测试类,模拟多线程多客户端请

关于分布式锁原理的一些学习与思考-redis分布式锁，zookeeper分布式锁

首先分布式锁和我们平常讲到的锁原理基本一样,目的就是确保,在多个线程并发时,只有一个线程在同一刻操作这个业务或者说方法.变量. 在一个进程中,也就是一个jvm 或者说应用中,我们很容易去处理控制,在jdk java.util 并发包中已经为我们提供了这些方法去加锁, 比如synchronized 关键字 或者Lock 锁,都可以处理. 但是我们现在的应用程序如果只部署一台服务器,那并发量是很差的,如果同时有上万的请求那么很有可能造成服务器压力过大,而瘫痪. 想想双十一 和 三十晚上十点分支付宝红

死磕 java同步系列之zookeeper分布式锁

问题 (1)zookeeper如何实现分布式锁? (2)zookeeper分布式锁有哪些优点? (3)zookeeper分布式锁有哪些缺点? 简介 zooKeeper是一个分布式的,开放源码的分布式应用程序协调服务,它可以为分布式应用提供一致性服务,它是Hadoop和Hbase的重要组件,同时也可以作为配置中心.注册中心运用在微服务体系中. 本章我们将介绍zookeeper如何实现分布式锁运用在分布式系统中. 基础知识 什么是znode? zooKeeper操作和维护的为一个个数据节点,称为 z

Curator实现zookeeper分布式锁的基本原理

一.写在前面 之前写过一篇文章(<拜托,面试请不要再问我Redis分布式锁的实现原理>),给大家说了一下Redisson这个开源框架是如何实现Redis分布式锁原理的,这篇文章再给大家聊一下ZooKeeper实现分布式锁的原理. 同理,我是直接基于比较常用的Curator这个开源框架,聊一下这个框架对ZooKeeper(以下简称zk)分布式锁的实现. 一般除了大公司是自行封装分布式锁框架之外,建议大家用这些开源框架封装好的分布式锁实现,这是一个比较快捷省事儿的方式. 二.ZooKeeper分布

彻底讲清楚ZooKeeper分布式锁的实现原理

一.写在前面 之前写过一篇文章(<拜托,面试请不要再问我Redis分布式锁的实现原理>),给大家说了一下Redisson这个开源框架是如何实现Redis分布式锁原理的,这篇文章再给大家聊一下ZooKeeper实现分布式锁的原理. 同理,我是直接基于比较常用的Curator这个开源框架,聊一下这个框架对ZooKeeper(以下简称zk)分布式锁的实现. 一般除了大公司是自行封装分布式锁框架之外,建议大家用这些开源框架封装好的分布式锁实现,这是一个比较快捷省事儿的方式. 二.ZooKeeper分布

高级java必会系列一：zookeeper分布式锁

方案1: 算法思路:利用名称唯一性,加锁操作时,只需要所有客户端一起创建/test/Lock节点,只有一个创建成功,成功者获得锁.解锁时,只需删除/test/Lock节点,其余客户端再次进入竞争创建节点,直到所有客户端都获得锁.特点:这种方案的正确性和可靠性是ZooKeeper机制保证的,实现简单.缺点是会产生"惊群"效应,假如许多客户端在等待一把锁,当锁释放时候所有客户端都被唤醒,仅仅有一个客户端得到锁.方案2:算法思路:临时顺序节点实现共享锁        客户端调用create(

zookeeper 分布式锁原理：

1 大家也许都很熟悉了多个线程或者多个进程间的共享锁的实现方式了,但是在分布式场景中我们会面临多个Server之间的锁的问题,实现的复杂度比较高.利用基于google chubby原理开发的开源的zookeeper,可以使得这个问题变得简单很多.下面介绍几种可能的实现方式,并且对比每种实现方式的优缺点. 1. 利用节点名称的唯一性来实现共享锁 ZooKeeper抽象出来的节点结构是一个和unix文件系统类似的小型的树状的目录结构.ZooKeeper机制规定:同一个目录下只能有一个唯一的文件名.例