zookeeper入门教程_ZooKeeper的事件机制原理

zookeeper入门教程_ZooKeeper的事件机制原理zookeeperwatcher架构zookeeper 配置中心分布式ID分布式锁集群搭建数据一致性协议:zab协议Zookeeper Leader选举Observer角色及其配置watcher架构客户端首先将Watcher注册到服务器,同时将Watch对象保存到客户端的Watch管理器中。当Zookeeper服务器监听到的数据发生变化时,服务器会通知客户端,接着客户端的Watch管理器会触发相关的Watcher来回调响应处理逻辑,从而完成整体的数据发布/订阅流程。javaAPIJava

大家好,又见面了,我是你们的朋友全栈君。如果您正在找激活码,请点击查看最新教程,关注关注公众号 “全栈程序员社区” 获取激活教程,可能之前旧版本教程已经失效.最新Idea2022.1教程亲测有效,一键激活。

Jetbrains全系列IDE使用 1年只要46元 售后保障 童叟无欺

watcher架构

在这里插入图片描述
客户端首先将Watcher注册到服务器,同时将Watch对象保存到客户端的Watch管理器中。当Zookeeper服务器监听到的数据发生变化时,服务器会通知客户端,接着客户端的Watch管理器会触发相关的Watcher来回调响应处理逻辑,从而完成整体的数据发布/订阅流程。
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
javaAPI
在这里插入图片描述
Java watch案例

public class WatcherDemo implements Watcher { 
   
    static ZooKeeper zooKeeper;
    static { 
   
        try { 
   
            zooKeeper = new ZooKeeper("192.168.3.39:2181", 4000,new WatcherDemo());
        } catch (IOException e) { 
   
            e.printStackTrace();
        }
    }
    @Override
    public void process(WatchedEvent event) { 
   
        System.out.println("eventType:"+event.getType());
        if(event.getType()==Event.EventType.NodeDataChanged){ 
   
            try { 
   
                zooKeeper.exists(event.getPath(),true);
            } catch (KeeperException e) { 
   
                e.printStackTrace();
            } catch (InterruptedException e) { 
   
                e.printStackTrace();
            }
        }
    }
    public static void main(String[] args) throws IOException, KeeperException, InterruptedException { 
   
        String path="/watcher";
        if(zooKeeper.exists(path,false)==null) { 
   
            zooKeeper.create("/watcher", "0".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
        Thread.sleep(1000);
        System.out.println("-----------");
        //true表示使用zookeeper实例中配置的watcher
        Stat stat=zooKeeper.exists(path,true);
        System.in.read();
    }
}

运行完程序,控制台显示
在这里插入图片描述
:此时启动 zookeeper 命令行终端,查看并且删除 watcher 节点:

在这里插入图片描述
IDE 控制台输出,触发了节点删除事件:
在这里插入图片描述

zookeeper 配置中心

使用zookeeper作为配置中心,通常情况下java应用程序会依赖很多配置文件,建议将配置信息配置在zookeeper中,将zookeeper作为服务器的配置中心,当配置发生变换之后,可以用zookeeperz中的watch机制捕获到发生变化的事件,以便客户端获得配置信息。

设计思路

  1. 连接zookeeper服务器
  2. 读取zookeeper中的配置信息,注册watcher监听器,存入本地变量
  3. 当zookeeper中的配置信息发生变化时,通过watcher的回调方法捕获数据变化事件
  4. 当zookeeper中的配置数据发生变化时,通过watcher的回调方法捕获数据变化事件
    java案例
package com.cc.duoxiancheng;

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

import java.util.concurrent.CountDownLatch;

public class MyConfigCenter implements Watcher { 
   
    // zookeeper地址
    String IP = "192.168.60.130:2181";
    // zookeeper是异步连接的,所以要使用countDownLatch 当响应返回的时候才继续改线程
    CountDownLatch countDownLatch = new CountDownLatch(1);
    static ZooKeeper zooKeeper;

    // 配置信息
    private String url;
    private String username;
    private String password;

    @Override
    public void process(WatchedEvent event) { 
   
        try { 
   
            if(event.getType() == Event.EventType.None) { 
   
                if(event.getState() == Event.KeeperState.SyncConnected){ 
   
                    System.out.println("连接成功");
                    countDownLatch.countDown();
                }
                else if(event.getState() == Event.KeeperState.Disconnected){ 
   
                    System.out.println("连接断开");
                }
                else if(event.getState() == Event.KeeperState.Expired){ 
   
                    System.out.println("连接超时");
                }
                else if(event.getState() == Event.KeeperState.AuthFailed){ 
   
                    System.out.println("认证失败");
                }
            }else if(event.getType() == Event.EventType.NodeDataChanged){ 
      //当有数据变华的时候,重新从配置信息中读取信息
                initValue();
            }
        }catch (Exception e){ 
   
            e.printStackTrace();
        }
    }
    private MyConfigCenter(){ 
   
        initValue();
    }
    public void initValue()
    { 
   
        try { 
   
            zooKeeper = new ZooKeeper(IP,50000,this);
            countDownLatch.await();
            //注册配置信息的watch道主watch中
            this.url = new String(zooKeeper.getData("/config/url",true,null));
            this.username = new String(zooKeeper.getData("/config/username",true,null));
            this.password = new String(zooKeeper.getData("/config/password",true,null));
        }catch (Exception e){ 
   
            e.printStackTrace();
        }
    }
}

分布式ID

在过去的单库单表系统中,通常可以使用数据库的AUTO_INCREMENT属性来自动为每条记录生成一个唯一的ID,d但是分库分表后,就无法在依靠数据库的auto_increment属性来唯一表示一条记录了。此时我们就可以用zookeeper在分布式环境下生成全局唯一id

设计思路

  1. 连接zookeeper服务器
  2. 指定路径下生成临时有序节点
  3. 取序列号及为分布式环境下的唯一ID

zookeeper中创建分布式唯一ID
在这里插入图片描述

public String getUniqueId(){ 
   
	String path = "";
	try{ 
   
		//生成zookeeper有序临时节点
		path = zookeeper.create(path,new byte[0],Ids.OPEN_ACL_UNSAFE,CREATEMode.EPHEMERAL_SEQUENTIAL);
	}catch(Exception e){ 
   
		e.printStackTrace();
	}
	//
	return path.subString(...);
}

分布式锁

分布式锁有多种实现方式,比如数据库,Redis都可以实现分布式锁,作为分布式协同工具zookeeper,当然也有着标准的实现方式。
设计思路

  1. 每个客户端往/Locks下创建l临时有序节点/Locks/Lock_,创建成功后/Locks下面会有每个客户端对应的节点。如/Locks/Lock_000000001
  2. 客户端获取/Locks下子节点,并进行排序,判断排在最前面的是否为自己,如果自己的锁节点排在第一位,代表获取锁成功
  3. 如果自己的锁节点不在第一位,则监听自己前一位的锁节点。例如,自己锁节点Lock_000000002,那么监听Lock_0000000001
  4. 当前一位锁节点(Lock_0000000001)对应的客户端执行完成,释放了锁,将会触发监听客户端
  5. 监听客户端重新执行第2步逻辑,判断自己是否获得了锁

java案例


private void createLock() throws Exception{ 
   
	//判断节点是否存在
	Stat stat = zooKeeper.exists(LOCK_ROOT_PATH,false);
	if(stat == null){ 
   
		zooKeeper.create(LOCK_ROOT_PATH,new byte[0],ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
	}
	//创建临时有序节点
	lockPath = zooKeeper.create(LOCK_ROOT_PATH + "/" + LOCK_NODE_NAME,new byte[0],ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
}

Watcher watch = new Watcher(){ 
   
	public void process(){ 
   
		if(event.getType == Event,EventType.NodeDeleted){ 
   
			synchronized(this){ 
   
				notifyAll();	
			}
		}
	}
}

private void attemptLock() throws Exception{ 
   
	//获取Locks节点下所有孩子节点
	List<String>list = zooKeeper.getChildren(LOCK_ROOT_PATH,false);
	//对节点进行排序
	Collection.sort(list);
	int index = list.indexOf(lockPath.subString(LOCK_ROOT_PATH.length() + 1));
	if(index == 0){ 
   	
		System.out.println("获取锁成功");
	}else { 
   
		//上一个节点路径
		String path = list.get(index-1);
		zooKeeper.exists(LOCK_ROOT_PATH + "/" + path,wathcer);
		if(stat == nukk){ 
   
			attemptLock();
		}else{ 
   
			synchronized(watcher){ 
   
				watcher.wait();
			}
			attemptLock();
		}
	}
}

public void releaseLock() throws Exception{ 
   
	zooKeeper.delete(this.lockPath,-1);
	zooKeeper.close();
	System,out.println("锁已经释放" + this.lockPath);
}

集群搭建

Ubuntu进行虚拟集群搭建。模拟共有3台服务器,端口分别为2181,2182,2183

  1. 基于zookeeper-3.4.10复制3份好的服务器文件
cp -r zookeeper-3.4.10 zookeeper2181
cp -r zookeeper-3.4.10 zookeeper2182
cp -r zookeeper-3.4.10 zookeeper2183
  1. 修改zookeeper2181,2182,2183服务器配置文件(以2181为例)
# 指定ip
clientPortAddress=192.168.0.120
# 配置集群
server.服务器编号=ip地址:Zookeeper服务器之间的通信端口:Leader选举的端口
server.1=192.168.0.120:2287:3387
server.2=192.168.0.120:2288:3388
server.3=192.168.0.120:2289:3389
  1. 在上一步dataDir指定的目录下,创建myid文件,然后在该文件添加上一步server配置的对应的服务器编号
# zookeeper2181对应的服务器编号是1
# /home/zookeeper/zookeeper2181/data目录下
echo "1" > myid
  1. 登录三个zookeeper服务器
# 启动
zkServer.sh start
# 状态
zkServer.sh status
# 登录
zkCli.sh -server 192.168.60.130:2181

数据一致性协议:zab协议

zab协议的全程是Zookeeper Atomic Broadcast,zookeeper是通过zab协议来保证分布式事务的最终一致性
基于zab协议,zookeeper集群中的角色主要有以下三类,如下表所示:
在这里插入图片描述
zab原子广播模式工作原理,通过类似两阶段提交协议的方法解决数据一致性:
在这里插入图片描述

  1. leader从客户端收到一个写请求
  2. leader生成一个新的事务并为这个事务生成一个唯一的ZXID
  3. leader将这个事务提议(propose)发送给所有的follow节点
  4. follower节点将受到的事务请求加入到历史队列(history queue)中,并发送ack给leader
  5. 当leader收到大多数follower(半数以上节点)的ack消息,leader会发送commit请求
  6. 当follow收到commit请求时,从历史队列中将事务请求commit

Zookeeper Leader选举

服务器状态

  • looking:寻找leader状态,当服务器处于该状态时,他会认为当前集群中没有leader,因此需要进入leader选举状态
  • leading:领导者状态。表明当前服务器是领导者
  • following:跟随者状态,表明当前服务器角色是follower
  • observing:观察者状态。表明当前服务角色是observer

服务器启动时的leader选举
每个节点启动的时候状态都是LOOKING,处于观望状态,接下来就开始进行选主流程

若进行Leader选举,则至少需要两台机器,这里选取3台机器组成的服务器集群为例。在集群初始化阶段,当有一台服务器Server1启动时,其单独无法进行和完成Leader选举,当第二台服务器Server2启动时,此时两台机器可以相互通信,每台机器都试图找到Leader,于是进入Leader选举过程。选举过程如下

  1. 每个Server发出一个投票。由于是初始情况,Server1和Server2都会将自己作为Leader服务器来进行投票,每次投票会包含所推举的服务器的myid和ZXID、epoch,使用(myid, ZXID,epoch)来表示,此时Server1的投票为(1, 0),Server2的投票为(2, 0),然后各自将这个投票发给集群中其他机器。

  2. 接受来自各个服务器的投票。集群的每个服务器收到投票后,首先判断该投票的有效性,如检查是否是本轮投票(epoch)、是否来自LOOKING状态的服务器。

  3. 处理投票。针对每一个投票,服务器都需要将别人的投票和自己的投票进行PK,PK规则如下

  • 优先比较epoch

  • 其次检查ZXID。ZXID比较大的服务器优先作为Leader

  • 如果ZXID相同,那么就比较myid。myid较大的服务器作为Leader服务器。

对于Server1而言,它的投票是(1, 0),接收Server2的投票为(2, 0),首先会比较两者的ZXID,均为0,再比较myid,此时Server2的myid最大,于是更新自己的投票为(2, 0),然后重新投票,对于Server2而言,其无须更新自己的投票,只是再次向集群中所有机器发出上一次投票信息即可。

  1. 统计投票。每次投票后,服务器都会统计投票信息,判断是否已经有过半机器接受到相同的投票信息,对于Server1、Server2而言,都统计出集群中已经有两台机器接受了(2, 0)的投票信息,此时便认为已经选出了Leader。

  2. 改变服务器状态。一旦确定了Leader,每个服务器就会更新自己的状态,如果是Follower,那么就变更为FOLLOWING,如果是Leader,就变更为LEADING。

运行时的leader选举
当集群中的leader服务器出现宕机或者不可用的情况时,那么整个集群将无法对外提供服务,而是进入新一轮的Leader选举,服务器运行期间的Leader选举和启动时期的Leader选举基本过程是一致的。

  1. 变更状态。Leader挂后,余下的非Observer服务器都会将自己的服务器状态变更为LOOKING,然后开始进入Leader选举过程。

  2. 每个Server会发出一个投票。在运行期间,每个服务器上的ZXID可能不同,此时假定Server1的ZXID为123,Server3的ZXID为122;在第一轮投票中,Server1和Server3都会投自己,产生投票(1, 123),(3, 122),然后各自将投票发送给集群中所有机器。接收来自各个服务器的投票。与启动时过程相同。

  3. 处理投票。与启动时过程相同,此时,Server1将会成为Leader。

  4. 统计投票。与启动时过程相同。

  5. 改变服务器的状态。与启动时过程相同

Observer角色及其配置

Observer角色特点:

  • 不参与集群的leader选举
  • 不参与集群中写数据时的ack反馈
    为了使用observer角色,在任何想变成observer角色的配置文件中加入如下配置:
peerType=observer

并在所有server的配置文件中,配置成observer模式的server的呐喊配置追加:observer,例如:

server.3=192.168.60.130:2289:3389:observer
版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/168967.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(0)


相关推荐

  • Google资深工程师深度讲解Go语言-内建容器(三)[通俗易懂]

    Google资深工程师深度讲解Go语言-内建容器(三)

  • Java Spring框架 (底层原理+入门)

    Java Spring框架 (底层原理+入门)JavaSpring框架Spring一、Spring的概述定义:Spring框架是一个开放源代码的J2EE应用程序框架,由RodJohnson发起,是针对bean的生命周期进行管理的轻量级容器(lightweightcontainer)。Spring解决了开发者在J2EE开发中遇到的许多常见的问题,提供了功能强大IOC、AOP及WebMVC等功能。Spring可以单独应用于构筑应用程序,也可以和Struts、Webwork、Tapestry等众多Web框架组合使用,并且可以与Swing

  • 进制之间的转换(二进制、八进制、十进制、十六进制)「建议收藏」

    进制之间的转换(二进制、八进制、十进制、十六进制)「建议收藏」二进制二进制就是计算机常用的进制,即逢二进一。例如:1010八进制八进制即逢八进一。例如:626十进制十进制就是我们在计算中常用的进制,所以就不再举例(即逢十进一)十六进制十六进制与其它进制有所不同,在10到15用英文字母进行表示。上面就是对进制的简单介绍,下面就是对进制转换而进行介绍。1.二进制转八进制拿二进制数10010110举例首先需要3个二进…

    2022年10月18日
  • svn 红叉叉图标解决方法

    svn 红叉叉图标解决方法

    2021年10月20日
  • 你知道Python是什么意思吗?

    你知道Python是什么意思吗?python中文就是蟒蛇的意思。在计算机中,它是一种编程语言。Python(英语发音:/ˈpaɪθən/),是一种面向对象、解释型计算机程序设计语言,由GuidovanRossum于1989年

  • MATLAB求解线性规划(含整数规划和0-1规划)问题[通俗易懂]

    MATLAB求解线性规划(含整数规划和0-1规划)问题[通俗易懂]线性规划是数学规划中的一类最简单规划问题,常见的线性规划是一个有约束的,变量范围为有理数的线性规划。如:对于这类线性规划问题,数学理论已经较为完善,可以有多种方法求解此类问题。但写这篇文章的目的并不是为了介绍数学理论,我们这里主要讲解如果利用工具求解这一类线性规划问题。最著名,同时也是最强大的数学最优化软件是LINGO/LINDO软件包,它能够求解多种的数学规划问题,同时还提供了多

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号