文档章节

ZooKeeper分布式锁应用

会跳舞的机器人
 会跳舞的机器人
发布于 2017/08/26 14:59
字数 1425
阅读 22
收藏 0

一、Zookeeper是什么

Zookeeper是一个高性能的分布式系统的协调服务。它在一个简单的接口里暴露公共服务:像命名、配置管理、同步、和群组服务,所以你没有必要从头开始实现它们。你可以使用现成的Zookeeper去实现共识、群组管理、领导人选举和业务协议。并且你可以在它的基础之上建立自己特定的需求。

二、Zookeeper分布式锁的实现原理

利用临时顺序节点实现Zookeeper分布式锁。

  • 1、首先为一个lock场景,在zookeeper中指定对应的一个根节点,用于记录资源竞争的内容,称之为/lock_node
  • 2、每个lock创建后,会lazy在zookeeper中创建一个node节点,表明对应的资源竞争标识。 (小技巧:node节点为EPHEMERAL_SEQUENTIAL,自增长的临时节点)。比如有两个客户端创建znode,那分别为/lock_node/lock-1、/lock_node/lock-2
  • 3、进行lock操作时,获取对应lock根节点下的所有字节点,也即处于竞争中的资源标识
  • 4、按照Fair竞争的原则,按照对应的自增内容做排序,取出编号最小的一个节点做为lock的owner,判断自己的节点id是否就为owner id,如果是则返回,lock成功。
  • 5、如果自己非owner id,按照排序的结果找到序号比自己前一位的id,关注它锁释放的操作(也就是exist watcher),形成一个链式的触发过程。
unlock过程
  • 6、将自己id对应的节点删除即可,对应的下一个排队的节点就可以收到Watcher事件,从而被唤醒得到锁后退出

ZooKeeper的几个特性让它非常合适作为分布式锁服务

  • zookeeper支持watcher机制,这样实现阻塞锁,可以watch锁数据,等到数据被删除,zookeeper会通知客户端去重新竞争锁。
  • zookeeper的数据可以支持临时节点的概念,即客户端写入的数据是临时数据,在客户端宕机后,临时数据会被删除,这样就实现了锁的异常释放。使用这样的方式,就不需要给锁增加超时自动释放的特性了。

三、Zookeeper分布式锁应用

Curator是Netflix公司开源的一个Zookeeper客户端,与Zookeeper提供的原生客户端相比,Curator的抽象层次更高,简化了Zookeeper客户端的开发量。

使用场景

1、创建ZooKeeperConnector.java

package com.robot.zookeeper.components;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author: 会跳舞的机器人
 * @date: 2017/6/21 18:07
 * @description:ZooKeeper连接器
 */
public class ZooKeeperConnector {
    private Logger logger = LoggerFactory.getLogger(this.getClass());

    /**
     * 连接IP端口信息,格式为:10.1.74.75:2281,10.1.74.75:2282,10.1.74.75:2283
     */
    private String hosts;

    private CuratorFramework client;

    private static final int DEFAULT_SESSION_TIMEOUT_MS = 30 * 1000;
    private static final int DEFAULT_CONNECTION_TIMEOUT_MS = 10 * 1000;

    private int sessionTimeout = DEFAULT_SESSION_TIMEOUT_MS;
    private int connectionTimeout = DEFAULT_CONNECTION_TIMEOUT_MS;

    /**
     * 连接ZooKeeper
     */
    public void connect() {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        client = CuratorFrameworkFactory.newClient(hosts, sessionTimeout, connectionTimeout, retryPolicy);
        client.start();
        logger.info("Successfully connected to Zookeeper [{}] ", hosts);
    }

    /**
     * 关闭ZooKeeper的连接
     */
    public void close() {
        CloseableUtils.closeQuietly(client);
    }

    /**
     * 重连
     *
     * @return
     */
    public CuratorFramework reConnect() {
        connect();
        return client;
    }

    public String getHosts() {
        return hosts;
    }

    public void setHosts(String hosts) {
        this.hosts = hosts;
    }

    public CuratorFramework getClient() {
        if (client == null) {
            connect();
        }
        return client;
    }

    public void setClient(CuratorFramework client) {
        this.client = client;
    }

    public int getSessionTimeout() {
        return sessionTimeout;
    }

    public void setSessionTimeout(int sessionTimeout) {
        this.sessionTimeout = sessionTimeout;
    }

    public int getConnectionTimeout() {
        return connectionTimeout;
    }

    public void setConnectionTimeout(int connectionTimeout) {
        this.connectionTimeout = connectionTimeout;
    }
}

2、创建AccountLock.java

package com.robot.zookeeper.utils;

import com.robot.zookeeper.components.ZooKeeperConnector;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 * @author: 会跳舞的机器人
 * @date: 2017/6/22 10:16
 * @description:账户分布式锁
 */
public class AccountLock {
    private Logger logger = LoggerFactory.getLogger(this.getClass());
    /**
     * 共享锁
     */
    InterProcessMutex lock;

    /**
     * 是否获取到锁
     */
    boolean wasAcquired = false;

    /**
     * 构造器
     *
     * @param path
     * @param zooKeeperConnector
     */
    public AccountLock(String path, ZooKeeperConnector zooKeeperConnector) {
        lock = new InterProcessMutex(zooKeeperConnector.getClient(), path);
    }

    /**
     * 尝试加锁并等待
     *
     * @param timeOut 超时时间(秒),-1表示不等待
     * @return true表示获取锁成功,false表示获取锁失败
     */
    public boolean acquire(int timeOut) {
        try {
            if (timeOut == -1) {
                wasAcquired = lock.acquire(-1, null);
            } else {
                wasAcquired = lock.acquire(timeOut, TimeUnit.SECONDS);
            }
        } catch (Exception e) {
            logger.error("Get lock time out error", e.getMessage());
            wasAcquired = false;
        }
        return wasAcquired;
    }

    /**
     * 尝试加锁并等待
     *
     * @param timeOut  timeOut 超时时间,-1表示不等待
     * @param timeUnit 超时时间单位
     * @return true表示获取锁成功,false表示获取锁失败
     */
    public boolean acquire(int timeOut, TimeUnit timeUnit) {
        try {
            wasAcquired = lock.acquire(timeOut, timeUnit);
        } catch (Exception e) {
            logger.error("Get lock time out error", e.getMessage());
            wasAcquired = false;
        }
        return wasAcquired;
    }

    /**
     * 释放锁
     */
    public void release() {
        if (wasAcquired) {
            try {
                lock.release();
            } catch (Exception e) {
                logger.error("release lock error", e.getMessage());
            }
        }
    }

}

3、测试类

package com.robot.zookeeper;

import com.robot.zookeeper.components.ZooKeeperConnector;
import com.robot.zookeeper.utils.AccountLock;

/**
 * @author: 会跳舞的机器人
 * @date: 2017/6/22 10:34
 * @description:
 */
public class Test {
    public static void main(String[] args) {
        final ZooKeeperConnector zooKeeperConnector = new ZooKeeperConnector();
        zooKeeperConnector.setHosts("192.168.133.128:2181,192.168.133.129:2182,192.168.133.130:2183");
        zooKeeperConnector.connect();

        /**
         * 创建4个线程去获取锁
         */
        for (int i = 1; i < 5; i++) {
            Thread thread = new Thread() {
                @Override
                public void run() {
                    AccountLock accountLock = new AccountLock("/ACCOUNT/221890", zooKeeperConnector);
                    boolean wasAcquired = accountLock.acquire(10);
                    if (wasAcquired) {
                        System.out.println("线程" + Thread.currentThread().getName() + "获取到锁");
                        try {
                            Thread.sleep(3000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        accountLock.release();
                        System.out.println("线程" + Thread.currentThread().getName() + "释放锁");
                    }
                }
            };
            thread.start();
        }
    }
}

输出结果:

线程Thread-3获取到锁
线程Thread-3释放锁
线程Thread-4获取到锁
线程Thread-4释放锁
线程Thread-2获取到锁
线程Thread-2释放锁
线程Thread-1获取到锁
线程Thread-1释放锁

账户加锁的时候,我们针对用户的ID进行加锁,在测试类中,我们创建了4个线程去获取锁,从输出结果可以看出每次只有一个线程能获取到锁,并且在该线程释放锁之后,其他的线程才能获取到锁。

当然,测试类中的ZooKeeperConnector的初始化一般都是通过Spring进行管理

<beans>
	<bean id="zkConnector" class="com.baibei.component.zk.ZooKeeperConnector"
		init-method="connect" lazy-init="false">
		<property name="hosts" value="#{environment['ZOOKEEPER.CONNECTION.HOSTS']}" />
	</bean>
</beans>

Demo中所需要的maven配置如下:

 <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-framework</artifactId>
            <version>2.10.0</version>
        </dependency>

        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>2.10.0</version>
        </dependency>

        <!-- curator的内嵌包版本存在问题,所以用这个版本来替代-->
        <dependency>
            <groupId>com.google.guava</groupId>
            <artifactId>guava</artifactId>
            <version>18.0</version>
        </dependency>

© 著作权归作者所有

会跳舞的机器人
粉丝 6
博文 23
码字总数 28541
作品 0
广州
私信 提问
分布式锁与实现(二)基于ZooKeeper实现

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

rechardchensir
2018/10/08
0
0
大数据教程(3.3):zookeeper简介

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

em_aaron
2018/07/28
0
0
Zookeeper 概述及应用场景

一、概述 分布式协调技术,主要用来解决分布式环境当中多个进程之间的同步控制,让他们有序的去访问某种临界资源,防止造成"脏数据"的后果。 ZooKeeper是一个分布式的,开放源码的分布式应用...

PeakFang-BOK
2018/10/20
0
0
分布式锁与实现(二)——基于ZooKeeper实现

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

刘祖鹏
2018/05/08
0
0
分布式设计与开发(三)------高一致性服务ZooKeeper

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

山哥
2012/03/19
0
0

没有更多内容

加载失败,请刷新页面

加载更多

蚂蚁区块链BaaS:开放在云端,落地于实体

自 2018 年 6 月上线以来,蚂蚁区块链 BaaS 不断在技术上实现突破,形成自身独特的优势,并做为行业领军者在不同领域的几十个场景实现落地。在区块链的商用时代加速到来之际,通过开放自身的...

阿里云官方博客
27分钟前
3
0
Idea cannot access【好使】

Idea中无法访问类中public方法。。 解决: idea点击左下角电脑图标,打开右边栏各种功能键 右边栏Maven Projects中点一下刷新,就是那个两个蓝的箭头组成的环 ----------------------------...

Airship
28分钟前
3
0
Spark内置图像数据源初探

作者:林武康,花名知瑕, 阿里巴巴计算平台事业部EMR团队的高级开发工程师,Apache HUE Contributor, 参与了多个开源项目的研发工作,对于分布式系统设计应用有较丰富的经验,目前主要专注于...

阿里云云栖社区
29分钟前
1
0
【面试被虐】游戏中的敏感词过滤是如何实现的?

小秋今天去面试了,面试官问了一个与敏感词过滤算法相关的问题,然而小秋对敏感词过滤算法一点也没听说过。于是,有了以下事情的发生….. 面试官开怼 面试官:玩过王者荣耀吧?了解过敏感词过...

爱编程的浪子
31分钟前
4
0
springboot之maven属性引入

<plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-resources-plugin</artifactId><version>2.7</version><configuration><delimiters><delimite......

Online_Reus
32分钟前
2
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部