散仙,在关于zookeeper的前几篇文章中,除了记录了zookeeper的一些基础知识,也介绍了怎么使用zookeeper来完成,配置文件同步,和主从自动切换的功能,那么,本篇散仙将会介绍下如何使用,zookeeper来完成分布式锁的功能,其实本质上是与主从切换的实现代码是非常类似的,但是功能上强调的重点不一样。
至于,为什么需要分布式锁(公平锁)?为什么不使用JAVA 自带的锁的应用?
1,为什么需要分布式锁? 因为在分布式环境下,可能会出现一些事务,这时候我们除了可以在存储层的数据库进行控制,也可以在应用层控制,举个例子来讲,中国的飞机路线,我们都知道任何时候,都只能由一架飞机通过,而这个控制这个由谁通过,什么时候通过,是由一个信号控制台来决定的,分布式的环境下由于节点分散在各个地方,各个区域,所以控制起来比较麻烦,这时候我们就可以使用zookeeper来轻松的完成,分布式锁的功能。
2,为什么不使用JAVA自带的锁?JAVA JDK提供了公平锁,与非公平锁,但这种实现是基于同一个JVM来说的,如果同一台机器上,不同的JVM,则可以使用文件锁,来实现,但是这些并不是分布式的模式,虽然可以通过RMI的方式来实现,但比较繁琐。
使用zookeeper来完成分布式锁的步骤如下:
序号 | 内容 |
1 | 创建一个持久znode |
2 | 多个程序并发的去zk服务上,创建一个短暂有时序性的节点路径。 |
3 | 各个节点监听,比它小的里面,最大的节点的动态。 |
4 | 如果发现,比它小的里面,最大的节点发生锁释放或退出,就自动接替为独占锁 |
5 | 没发生改变的节点,继续重复步骤,2,3,4 |
拓扑图如下所示:
注意上图中的master指的就是,获取锁的实例,这其实跟集群环境里只能有一个master的道理一样。
代码如下:
package com.test;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.text.SimpleDateFormat;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.data.Stat;
/***
* 基于zookeeper实现的
* 分布式公平锁
*
* @author qin dong liang
* QQ技术群交流:324714439
*
* */
public class Lock1 implements Watcher {
/**
* ZK实例
* */
private ZooKeeper zk;
/**原子计数锁,防止在zk没有连上前,执行CURD操作*/
private CountDownLatch down=new CountDownLatch(1);
public Lock1() {
// TODO Auto-generated constructor stub
}
public Lock1(String host)throws Exception {
this.zk=new ZooKeeper(host, 5000 , new Watcher() {
@Override
public void process(WatchedEvent event) {
// TODO Auto-generated method stub
/**链接上zk服务,岂可取消阻塞计数**/
if(event.getState()==KeeperState.SyncConnected){
down.countDown();
}
}
});
}
/**
* 字符编码
*
* **/
private static final Charset CHARSET=StandardCharsets.UTF_8;
/***
*
* 此方法是写入数据
* 如果不存在此节点
* 就会新建,已存在就是
* 更新
*
* **/
public void write(String path,String value)throws Exception{
Stat stat=zk.exists(path, false);
if(stat==null){
zk.create(path, value.getBytes(CHARSET), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}else{
zk.setData(path, value.getBytes(CHARSET), -1);
}
}
/**
*
* 切换锁
*
* **/
public void check()throws Exception{
List<String> list=zk.getChildren("/a", null);
Collections.sort(list);//排序使得节点有次序
if(list.isEmpty()){
System.out.println("此父路径下面没有节点,分布式锁任务完成或还没启动!");
}else{
String start=list.get(0);//获取第一个节点
String data=new String(zk.getData("/a/"+start, false,null));
if(data.equals("a")){//等于本身就启动作为Master
if(list.size()==1){
startMaster();//作为Master启动
}else{
automicSwitch();//对于非第一个启动的节点,会调用此方法,因为他的第一个挂了
//或释放锁了,所以它是抢占的
}
}else{
//非当前节点,就打印当前节点,监控的节点
for(int i=0;i<list.size();i++){
//获取那个节点存的此客户端的模拟IP
String temp=new String(zk.getData("/a/"+list.get(i), false, null));
if(temp.equals("a")){
//因为前面作为首位判断,所以这个出现的位置不可能是首位
//需要监听小节点里面的最大的一个节点
String watchPath=list.get(i-1);
System.out.println("Lock1监听的是: "+watchPath);
zk.exists("/a/"+watchPath, this);//监听此节点的详细情况,如果发生节点注销事件
//则会触发自身的process方法
break;//结束循环
}
}
}
}
}
@Override
public void process(WatchedEvent event) {
// TODO Auto-generated method stub
if(event.getType()==Event.EventType.NodeDeleted){
//如果发现,监听的节点,挂掉了,那么就重新,进行监听
try{
System.out.println("注意有锁退出或释放,公平锁开始抢占........");
check();
}catch(Exception e){
e.printStackTrace();
}
}
}
/**
*
* 读取数据,给定一个路径和
* 监听事件
*
* ***/
public String read(String path,Watcher watch)throws Exception{
byte[] data=zk.getData(path, watch, null);
return new String(data,CHARSET);
}
SimpleDateFormat f=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
/**
* 关闭zk连接
*
* **/
public void close()throws Exception{
zk.close();
}
/**
* 释放锁
* @throws Exception
*/
public void automicSwitch()throws Exception{
// System.out.println("有节点释放锁,Lock1锁占入......., 时间 "+f.format(new Date()));
System.out.println("Lock1的上级锁节点退出或释放锁了,Lock1锁占入......., 时间 "+f.format(new Date()));
}
/**
* 创建一个持久node,
*
* **/
public void createPersist()throws Exception{
zk.create("/a", "主节点".getBytes(), Ids.OPEN_ACL_UNSAFE , CreateMode.PERSISTENT);
System.out.println("创建主节点成功........");
}
/***
* 创建锁node,注意是抢占 的
*
*
* */
public void createTemp()throws Exception{
zk.create("/a/b", "a".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
System.out.println("Lock1注册锁成功,进入公平队列...........");
}
public static void main(String[] args)throws Exception {
//Slave s=new Slave("192.168.120.128:2181");
Lock1 lock=new Lock1("192.168.120.128:2181");
// lock.createPersist();//创建主节点
lock.createTemp();//注册临时有序节点
lock.check();
Thread.sleep(Long.MAX_VALUE);
//lock.close();
}
/***
* 获取锁成功
*
* */
public void startMaster(){
System.out.println("Lock1节点获取锁了,其他节点等待........");
}
}
代码如上,所示,测试的时候,需要搭建一个3个节点的zookeeper集群,关于怎么搭建zookeeper集群,散仙前面的文章里有介绍,需要注意的是myid文件不要漏掉。
上面这个类,需要拷贝多份,并改变里面的节点的值,放在不同的eclipse中,进行模拟测试。
分享到:
相关推荐
分布式之Zookeeper(高清视频教程).rar
ubuntu系统部署完全分布式hadoop时,先部署了zookeeper和hbase
分布式框架Zookeeper之服务注册与订阅.........................
分布式架构Zookeeper入门实战到底层原理剖析
Zookeeper伪分布式集群搭建描述伪分布式集群就是在一台机器部署多个zk应用,部署之前服务器需要有jdk环境 java -version可以显示相关java
互联网大厂必会的高薪技能之分布式架构Zookeeper入门实战到底层原理剖析
server程序入口(启动类在zookeeper-server/src/main/java文件夹中,org.apache.zookeeper.server.quorum.QuorumPeerMain),program argument为conf/zoo.cfg,将conf/log4j.properties配置拷贝到zookeeper-server/...
三台主机搭建分布式Zookeeper,完全分布式搭建zookeeper集群,及其简单的操作
分布式服务框架 Zookeeper -- 管理分布式环境中的数据
本文来自于cnblogs,说到分布式开发Zookeeper是必须了解和掌握的,分布式消息服务kafka、hbase到hadoop等分布式大数据处理都会用到Zookeeper,所以在此将Zookeeper作为基础来讲解。Zookeeper是分布式服务框架,主要...
#资源达人分享计划#
详解Zookeeper伪分布式集群环境搭建过程
基于zookeeper的分布式锁简单实现,包含测试代码,实用工具类
ZooKeeper 是一个开源的分布式 应用程序协调服务器,其为分布式系统提供一致性服务。其一致性是通过基于 Paxos 算法的 ZAB 协议完成的。其主要功能包括:配置维护、域名服务、分布式同步、集群管理等。 zookeeper 的...
#资源达人分享计划#
从Paxos到Zookeeper分布式一致性原理与实践 + ZooKeeper-分布式过程协同技术详解
在分布式高并发的情况下,多个访问之间拥有唯一的一个订单编号,多次提交的订单不会出现重复的,只能是唯一的,ZooKeeper就很好的解决了唯一性
分布式服务框架zookeeper 3.4.13版本,主要用于构建RPC服务与常见集群等。
从传统JavaWeb转入分布式系统应用,再到接触分布式协调框架ZooKeeper,通过痛苦的思维逻辑和理念转变,历经一个月时间,小伙伴们终于把ZooKeeper嵌入到了BoCloud博云的BeyondContainer中,并在其上进行相应功能的...